| #!/usr/bin/env python |
| # Copyright (c) 2011 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """This functional test spawns a web server, and runs chrome to point |
| at that web server. |
| |
| The content served contains prefetch requests, and the tests assert that the |
| webserver logs reflect that. |
| |
| Run like any functional test: |
| $ python chrome/test/functional/prefetch.py |
| in a repo with a built pyautolib |
| |
| The import of multiprocessing implies python 2.6 is required |
| """ |
| |
| import os |
| import time |
| import multiprocessing |
| import Queue |
| import string |
| from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer |
| |
| import pyauto_functional # Must be imported before pyauto |
| import pyauto |
| |
| # this class handles IPC retrieving server "logs" from our integral |
| # server. Each test should clear() the log, and then run asserts on |
| # the retrieval list. |
| |
| # at startup, the server puts an int in the queue which is its port, |
| # we store that for subsequent tests |
| |
| class ServerLog: |
| def clear(self): |
| self.log = {} |
| |
| def __init__(self,queue): |
| self.clear() |
| self.port = None |
| self.queue = queue |
| |
| def _readQueue(self): |
| try: |
| while True: |
| queueval = self.queue.get(False) |
| if isinstance(queueval,int): |
| self.port = queueval |
| else: |
| self.log[queueval] = True |
| except Queue.Empty: |
| return |
| |
| def getPort(self): |
| if not self.port: |
| self._readQueue() |
| return self.port |
| |
| def isRetrieved(self,path): |
| self._readQueue() |
| try: |
| return self.log[path] |
| except KeyError: |
| return None |
| |
| # |
| # The next few classes run a simple web server that returns log information |
| # via a multiprocessing.Queue. |
| # |
| class AbstractPrefetchServerHandler(BaseHTTPRequestHandler): |
| content = { |
| "prefetch-origin.html": |
| (200, """<html><head> |
| <link rel="prefetch" href="static-prefetch-target.html"> |
| <script type="text/javascript"> |
| function changeParagraph() |
| { |
| var newPara = document.createElement("p"); |
| newPara.innerHTML = |
| "<link rel=\\"prefetch\\" href=\\"dynamic-prefetch-target.html\\">" + |
| "<p>This paragraph contains a dynamic link prefetch. " + |
| "The target of this prefetch is " + |
| "<a href=\\"dynamic-prefetch-target.html\\">this document.</a>"; |
| var para = document.getElementById("p1"); |
| document.body.insertBefore(newPara,para); |
| } |
| </script> |
| </head> |
| <body onload="changeParagraph()"> |
| <p id="p1">This is a document that contains a link prefetch. The target of |
| that prefetch is <a href="static-prefetch-target.html">this document.</a> |
| </body>"""), |
| "static-prefetch-target.html": |
| (200, "<html><head></head><body>empty</body>"), |
| "dynamic-prefetch-target.html": |
| (200, "<html><head></head><body>empty</body>")} |
| |
| def do_GET(self): |
| self.queue.put(self.path[1:]) |
| try: |
| response_code, response = self.content[self.path[1:]] |
| self.send_response(response_code) |
| self.end_headers() |
| self.wfile.write(response) |
| except KeyError: |
| self.send_response(404) |
| self.end_headers() |
| |
| def run_web_server(queue_arg): |
| class PrefetchServerHandler(AbstractPrefetchServerHandler): |
| queue = queue_arg |
| server = HTTPServer(('',0), PrefetchServerHandler) |
| queue.put(server.server_port) |
| server.serve_forever() |
| |
| # |
| # Here's the test itself |
| # |
| queue = multiprocessing.Queue() |
| server_log = ServerLog(queue) |
| |
| class PrefetchTest(pyauto.PyUITest): |
| """Testcase for Prefetching""" |
| def testBasic(self): |
| server_log.clear() |
| url = "http://localhost:%d/prefetch-origin.html" % server_log.getPort() |
| self.NavigateToURL(url) |
| self.assertEqual(True, server_log.isRetrieved("prefetch-origin.html")) |
| time.sleep(0.1) # required since prefetches occur after onload |
| self.assertEqual(True, server_log.isRetrieved( |
| "static-prefetch-target.html")) |
| self.assertEqual(True, server_log.isRetrieved( |
| "dynamic-prefetch-target.html")) |
| |
| if __name__ == '__main__': |
| web_server = multiprocessing.Process(target=run_web_server,args=(queue,)) |
| web_server.daemon = True |
| web_server.start() |
| pyauto_functional.Main() |
