src/chrome/test/functional/prefetch.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2011 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """This functional test spawns a web server, and runs chrome to point
   7 at that web server.
   8
   9 The content served contains prefetch requests, and the tests assert that the
  10 webserver logs reflect that.
  11
  12 Run like any functional test:
  13 $ python chrome/test/functional/prefetch.py
  14 in a repo with a built pyautolib
  15
  16 The import of multiprocessing implies python 2.6 is required
  17 """
  18
  19 import os
  20 import time
  21 import multiprocessing
  22 import Queue
  23 import string
  24 from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
  25
  26 import pyauto_functional  # Must be imported before pyauto
  27 import pyauto
  28
  29 # this class handles IPC retrieving server "logs" from our integral
  30 # server.  Each test should clear() the log, and then run asserts on
  31 # the retrieval list.
  32
  33 # at startup, the server puts an int in the queue which is its port,
  34 # we store that for subsequent tests
  35
  36 class ServerLog:
  37   def clear(self):
  38     self.log = {}
  39
  40   def __init__(self,queue):
  41     self.clear()
  42     self.port = None
  43     self.queue = queue
  44
  45   def _readQueue(self):
  46     try:
  47       while True:
  48         queueval = self.queue.get(False)
  49         if isinstance(queueval,int):
  50           self.port = queueval
  51         else:
  52          self.log[queueval] = True
  53     except Queue.Empty:
  54       return
  55
  56   def getPort(self):
  57     if not self.port:
  58       self._readQueue()
  59     return self.port
  60
  61   def isRetrieved(self,path):
  62     self._readQueue()
  63     try:
  64       return self.log[path]
  65     except KeyError:
  66       return None
  67
  68 #
  69 # The next few classes run a simple web server that returns log information
  70 # via a multiprocessing.Queue.
  71 #
  72 class AbstractPrefetchServerHandler(BaseHTTPRequestHandler):
  73   content = {
  74       "prefetch-origin.html":
  75         (200, """<html><head>
  76 <link rel="prefetch" href="static-prefetch-target.html">
  77 <script type="text/javascript">
  78 function changeParagraph()
  79 {
  80   var newPara = document.createElement("p");
  81   newPara.innerHTML =
  82         "<link rel=\\"prefetch\\" href=\\"dynamic-prefetch-target.html\\">" +
  83        "<p>This paragraph contains a dynamic link prefetch.  " +
  84        "The target of this prefetch is " +
  85        "<a href=\\"dynamic-prefetch-target.html\\">this document.</a>";
  86   var para = document.getElementById("p1");
  87   document.body.insertBefore(newPara,para);
  88 }
  89 </script>
  90 </head>
  91 <body onload="changeParagraph()">
  92 <p id="p1">This is a document that contains a link prefetch.  The target of
  93 that prefetch is <a href="static-prefetch-target.html">this document.</a>
  94 </body>"""),
  95       "static-prefetch-target.html":
  96         (200, "<html><head></head><body>empty</body>"),
  97       "dynamic-prefetch-target.html":
  98         (200, "<html><head></head><body>empty</body>")}
  99
 100   def do_GET(self):
 101     self.queue.put(self.path[1:])
 102     try:
 103       response_code, response = self.content[self.path[1:]]
 104       self.send_response(response_code)
 105       self.end_headers()
 106       self.wfile.write(response)
 107     except KeyError:
 108       self.send_response(404)
 109       self.end_headers()
 110
 111 def run_web_server(queue_arg):
 112   class PrefetchServerHandler(AbstractPrefetchServerHandler):
 113     queue = queue_arg
 114   server = HTTPServer(('',0), PrefetchServerHandler)
 115   queue.put(server.server_port)
 116   server.serve_forever()
 117
 118 #
 119 # Here's the test itself
 120 #
 121 queue = multiprocessing.Queue()
 122 server_log = ServerLog(queue)
 123
 124 class PrefetchTest(pyauto.PyUITest):
 125   """Testcase for Prefetching"""
 126   def testBasic(self):
 127     server_log.clear()
 128     url = "http://localhost:%d/prefetch-origin.html" % server_log.getPort()
 129     self.NavigateToURL(url)
 130     self.assertEqual(True, server_log.isRetrieved("prefetch-origin.html"))
 131     time.sleep(0.1)  # required since prefetches occur after onload
 132     self.assertEqual(True, server_log.isRetrieved(
 133         "static-prefetch-target.html"))
 134     self.assertEqual(True, server_log.isRetrieved(
 135         "dynamic-prefetch-target.html"))
 136
 137 if __name__ == '__main__':
 138   web_server = multiprocessing.Process(target=run_web_server,args=(queue,))
 139   web_server.daemon = True
 140   web_server.start()
 141   pyauto_functional.Main()