#!/usr/bin/python
# Copyright (c) 2010 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, """
This is a document that contains a link prefetch. The target of
that prefetch is this document.
"""),
"static-prefetch-target.html":
(200, "
empty"),
"dynamic-prefetch-target.html":
(200, "empty")}
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()