blob: 945ccbf5ca6c5a33e8bd7e3ded9a799db9cee20d [file] [log] [blame]
# Copyright 2016 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.
import argparse
import json
import logging
import os
import re
import socket
import shlex
import sys
import time
import traceback
import unittest
import urlparse
sys.path.append(os.path.join(os.path.dirname(__file__), os.pardir, os.pardir,
os.pardir, 'third_party', 'webdriver', 'pylib'))
from selenium import webdriver
from import Options
# These network condition values are used in SetNetworkConnection()
'4G': {
'latency': 20,
'upload_throughput': 4096 * 1024,
'download_throughput': 4096 * 1024,
'offline': False,
'3G': {
'latency': 425,
'upload_throughput': 750 * 1024,
'download_throughput': 750 * 1024,
'offline': False,
'2G': {
'latency': 1650,
'upload_throughput': 250 * 1024,
'download_throughput': 250 * 1024,
'offline': False,
'latency': 0,
'upload_throughput': 0 * 1024,
'download_throughput': 0 * 1024,
# This stays false so that Chrome won't disconnect from ChromeDriver.
'offline': False,
def ParseFlags():
"""Parses the given command line arguments.
A new Namespace object with class properties for each argument added below.
See pydoc for argparse.
def TestFilter(v):
# The filtering here allows for any number of * wildcards with a required
# . separator between classname and methodname, but no other special
# characters.
return re.match(r'^([A-Za-z_0-9\*]+\.)?[A-Za-z_0-9\*]+$', v).group(0)
raise argparse.ArgumentTypeError('Test filter "%s" is not a valid filter'
% v)
parser = argparse.ArgumentParser()
parser.add_argument('--browser_args', type=str, help='Override browser flags '
'in code with these flags')
default='1.1 Chrome-Compression-Proxy', help='What the via should match to '
'be considered valid')
parser.add_argument('--android', help='If given, attempts to run the test on '
'Android via adb. Ignores usage of --chrome_exec', action='store_true')
default='', help='Set the android package for Chrome')
parser.add_argument('--chrome_exec', type=str, help='The path to '
'the Chrome or Chromium executable')
parser.add_argument('chrome_driver', type=str, help='The path to '
'the ChromeDriver executable. If not given, the default system chrome '
'will be used.')
parser.add_argument('--disable_buffer', help='Causes stdout and stderr from '
'tests to output normally. Otherwise, the standard output and standard '
'error streams are buffered during the test run, and output from a '
'passing test is discarded. Output will always be echoed normally on test '
'fail or error and is added to the failure messages.', action='store_true')
parser.add_argument('-c', '--catch', help='Control-C during the test run '
'waits for the current test to end and then reports all the results so '
'far. A second Control-C raises the normal KeyboardInterrupt exception.',
parser.add_argument('-f', '--failfast', help='Stop the test run on the first '
'error or failure.', action='store_true')
parser.add_argument('--test_filter', '--gtest_filter', type=TestFilter,
help='The filter to use when discovering tests to run, in the form '
'<class name>.<method name> Wildcards (*) are accepted. Default=*',
parser.add_argument('--logging_level', choices=['DEBUG', 'INFO', 'WARN',
'ERROR', 'CRIT'], default='WARN', help='The logging verbosity for log '
'messages, printed to stderr. To see stderr logging output during a '
'successful test run, also pass --disable_buffer. Default=ERROR')
parser.add_argument('--log_file', help='If given, write logging statements '
'to the given file instead of stderr.')
parser.add_argument('--skip_slow', action='store_true', help='If set, tests '
'marked as slow will be skipped.', default=False)
parser.add_argument('--chrome_start_time', type=int, default=0, help='The '
'number of attempts to check if Chrome has fetched a proxy client config '
'before starting the test. Each check takes about one second.')
return parser.parse_args(sys.argv[1:])
def GetLogger(name='common'):
"""Creates a Logger instance with the given name and returns it.
If a logger has already been created with the same name, that instance is
returned instead.
name: The name of the logger to return.
A logger with the given name.
logger = logging.getLogger(name)
if hasattr(logger, "initialized"):
return logger
logging_level = ParseFlags().logging_level
if logging_level == 'DEBUG':
elif logging_level == 'INFO':
elif logging_level == 'WARN':
elif logging_level == 'ERROR':
elif logging_level == 'CRIT':
formatter = logging.Formatter('%(asctime)s %(funcName)s() %(levelname)s: '
if ParseFlags().log_file:
fh = logging.FileHandler(ParseFlags().log_file)
ch = logging.StreamHandler(sys.stderr)
logger.initialized = True
return logger
class TestDriver:
"""The main driver for an integration test.
This class is the tool that is used by every integration test to interact with
the Chromium browser and validate proper functionality. This class sits on top
of the Selenium Chrome Webdriver with added utility and helper functions for
Chrome-Proxy. This class should be used with Python's 'with' operator.
_flags: A Namespace object from the call to ParseFlags()
_driver: A reference to the driver object from the Chrome Driver library.
_chrome_args: A set of string arguments to start Chrome with.
_url: The string URL that Chrome will navigate to for this test.
_has_logs: Boolean flag set when a page is loaded and cleared when logs are
_control_network_connection: Boolean signal that this chromedriver instance
was meant to support network connection control, and thus had to enable
mobile emulation
_network_connection: The connection type to use on start up
def __init__(self, control_network_connection=False):
self._flags = ParseFlags()
self._driver = None
self._chrome_args = set()
self._url = ''
self._logger = GetLogger(name='TestDriver')
self._has_logs = False
self._control_network_connection = control_network_connection
self._network_connection = None
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, tb):
if self._driver:
def _OverrideChromeArgs(self):
"""Overrides any given arguments in the code with those given on the command
Arguments that need to be overridden may contain different values for
a flag given in the code. In that case, check by the flag whether to
override the argument.
def GetDictKey(argument):
return argument.split('=', 1)[0]
if self._flags.browser_args and len(self._flags.browser_args) > 0:
# Build a dict of flags mapped to the whole argument.
original_args = {}
for arg in self._chrome_args:
original_args[GetDictKey(arg)] = arg
# Override flags given in code with any command line arguments.
for override_arg in shlex.split(self._flags.browser_args):
arg_key = GetDictKey(override_arg)
if (arg_key in original_args
and original_args[arg_key] in self._chrome_args):
self._chrome_args.remove(original_args[arg_key])'Removed Chrome flag. %s', original_args[arg_key])
self._chrome_args.add(override_arg)'Added Chrome flag. %s', override_arg)
# Always add the flag that allows histograms to be queried in javascript.
def _StartDriver(self):
"""Parses the flags to pass to Chromium, then starts the ChromeDriver.
If running Android, the Android package name is passed to ChromeDriver here.
capabilities = {
'loggingPrefs': {'performance': 'INFO'},
chrome_options = Options()
if self._control_network_connection:
'networkConnectionEnabled': True,
'mobileEmulationEnabled': True,
if not
{'deviceName': 'Google Nexus 5'})
for arg in self._chrome_args:
chrome_options.add_argument(arg)'Starting Chrome with these flags: %s',
self._logger.debug('Will use Chrome on Android')
elif self._flags.chrome_exec:
chrome_options.binary_location = self._flags.chrome_exec'Using the Chrome binary at this path: %s',
self._logger.debug('ChromeOptions will be parsed into these capabilities: '
'%s', json.dumps(chrome_options.to_capabilities()))
driver = webdriver.Chrome(executable_path=self._flags.chrome_driver,
desired_capabilities=capabilities, chrome_options=chrome_options)
'getAvailableLogTypes': ('GET', '/session/$sessionId/log/types'),
'getLog': ('POST', '/session/$sessionId/log'),
('POST', '/session/$sessionId/chromium/network_conditions')})
self._driver = driver
if self._control_network_connection:
# Set network connection if it was called before LoadURL()
def _StopDriver(self):
"""Nicely stops the ChromeDriver.
self._logger.debug('Stopping ChromeDriver')
self._driver = None
def AddChromeArgs(self, args):
"""Adds multiple arguments that will be passed to Chromium at start.
args: An iterable of strings, each an argument to pass to Chrome at start.
for arg in args:
def AddChromeArg(self, arg):
"""Adds a single argument that will be passed to Chromium at start.
arg: a string argument to pass to Chrome at start
self._logger.debug('Adding Chrome arg: %s', arg)
def RemoveChromeArgs(self, args):
"""Removes multiple arguments that will no longer be passed to Chromium at
args: An iterable of strings to no longer use the next time Chrome
for arg in args:
def RemoveChromeArg(self, arg):
"""Removes a single argument that will no longer be passed to Chromium at
arg: A string flag to no longer use the next time Chrome starts.
self._logger.debug('Removing Chrome arg: %s', arg)
def ClearChromeArgs(self):
"""Removes all arguments from Chromium at start.
self._logger.debug('Clearing all Chrome args')
def ClearCache(self):
"""Clears the browser cache.
Important note: ChromeDriver automatically starts
a clean copy of Chrome on every instantiation.
res = self.ExecuteJavascript('if( && chrome.benchmarking && '
'chrome.benchmarking.clearCache){chrome.benchmarking.clearCache(); '
'clearHostResolverCache();}')'Cleared browser cache. Returned=%s', str(res))
def SetNetworkConnection(self, connection_type):
"""Changes the emulated connection type.
connection_type: the connection type to use according to the dict near the
top of the file OR a dictionary specifying the network conditions
if not self._control_network_connection:
raise Exception('SetNetworkConnection can only be used with a TestDriver '
'initalized with control_network_connection=True')
self._network_connection = connection_type
network = (NETWORKS[self._network_connection]
if connection_type in NETWORKS else connection_type)
if self._driver and self._network_connection:
{'network_conditions': network})
def LoadURL(self, url, timeout=30):
"""Starts Chromium with any arguments previously given and navigates to the
given URL.
url: The URL to navigate to.
timeout: The time in seconds to load the page before timing out.
self._url = url
if (len(urlparse.urlparse(url).netloc) == 0 and
len(urlparse.urlparse(url).scheme) == 0):
self._logger.warn('Invalid URL: "%s". Did you forget to prepend '
'"http://"? See RFC 1808 for more information', url)
if not self._driver:
self._logger.debug('Set page load timeout to %f seconds', timeout)
self._logger.debug('Loaded page %s', url)
self._has_logs = True
def FindElement(self, by, value):
"""Finds an element on the page.
Uses the By selector and value given.
by: the selenium.webdriver.common.By selector
value: the value
a WebElement object
return self._driver.find_element(by=by, value=value)
def ExecuteJavascript(self, script, timeout=30):
"""Executes the given javascript in the browser's current page in an
anonymous function.
If you expect a result and don't get one, try adding a return statement or
using ExecuteJavascriptStatement() below.
script: A string of Javascript code.
timeout: Timeout for the Javascript code to return in seconds.
A string of the verbatim output from the Javascript execution.
if not self._driver:
# TODO(robertogden): Use 'driver.set_script_timeout(timeout)' instead after
# crbug/672114 is fixed.
default_timeout = socket.getdefaulttimeout()
self._logger.debug('Set socket timeout to %f seconds', timeout)
script_result = self._driver.execute_script(script)
self._logger.debug('Executed Javascript in browser: %s', script)
self._logger.debug('Set socket timeout to %s', str(default_timeout))
return script_result
def ExecuteJavascriptStatement(self, script, timeout=30):
"""Wraps ExecuteJavascript() for use with a single statement.
Behavior is analogous to 'function(){ return <script> }();'
script: A string of Javascript code.
timeout: Timeout for the Javascript code to return in seconds.
A string of the verbatim output from the Javascript execution.
return self.ExecuteJavascript("return " + script, timeout)
def GetHistogram(self, histogram, timeout=30):
"""Gets a Chrome histogram as a dictionary object.
histogram: the name of the histogram to fetch
timeout: timeout for the underlying Javascript query.
A dictionary object containing information about the histogram.
js_query = 'statsCollectionController.getBrowserHistogram("%s")' % histogram
string_response = self.ExecuteJavascriptStatement(js_query, timeout)
self._logger.debug('Got %s histogram=%s', histogram, string_response)
return json.loads(string_response)
def WaitForJavascriptExpression(self, expression, timeout, min_poll=0.1,
"""Waits for the given Javascript expression to evaluate to True within the
given timeout. This method polls the Javascript expression within the range
of |min_poll| and |max_poll|.
expression: The Javascript expression to poll, as a string.
min_poll: The most frequently to poll as a float.
max_poll: The least frequently to poll as a float.
Returns: The result of the expression.
poll_interval = max(min(max_poll, float(timeout) / 10.0), min_poll)
self._logger.debug('Poll interval=%f seconds', poll_interval)
result = self.ExecuteJavascriptStatement(expression)
total_waited_time = 0
while not result and total_waited_time < timeout:
total_waited_time += poll_interval
result = self.ExecuteJavascriptStatement(expression)
if not result:
self._logger.error('%s not true after %f seconds' % (expression, timeout))
raise Exception('%s not true after %f seconds' % (expression, timeout))
return result
def GetPerformanceLogs(self, method_filter=r'Network\.responseReceived'):
"""Returns all logged Performance events from Chrome. Raises an Exception if
no pages have been loaded since the last time this function was called.
method_filter: A regex expression to match the method of logged events
against. Only logs who's method matches the regex will be returned.
Performance logs as a list of dicts, since the last time this function was
if not self._has_logs:
raise Exception('No pages loaded since last Network log query!')
all_messages = []
for log in self._driver.execute('getLog', {'type': 'performance'})['value']:
message = json.loads(log['message'])['message']
self._logger.debug('Got Performance log: %s', log['message'])
if re.match(method_filter, message['method']):
all_messages.append(message)'Got %d performance logs with filter method=%s',
len(all_messages), method_filter)
self._has_logs = False
return all_messages
def SleepUntilHistogramHasEntry(self, histogram_name, sleep_intervals=10):
"""Polls if a histogram exists in 1-6 second intervals for 10 intervals.
Allows script to run with a timeout of 5 seconds, so the default behavior
allows up to 60 seconds until timeout.
histogram_name: The name of the histogram to wait for
sleep_intervals: The number of polling intervals, each polling cycle takes
no more than 6 seconds.
Whether the histogram exists
histogram = {}
while(not histogram and sleep_intervals > 0):
histogram = self.GetHistogram(histogram_name, 5)
if (not histogram):
sleep_intervals -= 1
return bool(histogram)
def GetHTTPResponses(self, include_favicon=False, skip_domainless_pages=True,
"""Parses the Performance Logs and returns a list of HTTPResponse objects.
Use caution when calling this function multiple times. Only responses
since the last time this function was called are returned (or since Chrome
started, whichever is later). An Exception will be raised if no page was
loaded since the last time this function was called.
include_favicon: A bool that if True will include responses for favicons.
skip_domainless_pages: If True, only responses with a net_loc as in RFC
1808 will be included. Pages such as about:blank will be skipped.
override_has_logs: Allows the _has_logs property to be set if there was
not a page load but an XHR was expected instead.
A list of HTTPResponse objects, each representing a single completed HTTP
transaction by Chrome.
if override_has_logs:
self._has_logs = True
def MakeHTTPResponse(log_dict):
params = log_dict['params']
response_dict = params['response']
http_response_dict = {
'response_headers': response_dict['headers'] if 'headers' in
response_dict else {},
'request_headers': response_dict['requestHeaders'] if 'requestHeaders'
in response_dict else {},
'url': response_dict['url'] if 'url' in response_dict else '',
'protocol': response_dict['protocol'] if 'protocol' in response_dict
else '',
'port': response_dict['remotePort'] if 'remotePort' in response_dict
else -1,
'status': response_dict['status'] if 'status' in response_dict else -1,
'request_type': params['type'] if 'type' in params else ''
return HTTPResponse(**http_response_dict)
all_responses = []
for message in self.GetPerformanceLogs():
response = MakeHTTPResponse(message)
self._logger.debug('New HTTPResponse: %s', str(response))
is_favicon = response.url.endswith('favicon.ico')
has_domain = len(urlparse.urlparse(response.url).netloc) > 0
if (not is_favicon or include_favicon) and (not skip_domainless_pages or
else:"Skipping HTTPResponse with url=%s in returned logs.",
response.url)'%d new HTTPResponse objects found in the logs %s '
'favicons', len(all_responses), ('including' if include_favicon else
'not including'))
return all_responses
class HTTPResponse:
"""This class represents a single HTTP transaction (request and response) by
This class also includes several convenience functions for ChromeProxy
specific assertions.
_response_headers: A dict of response headers.
_request_headers: A dict of request headers.
_url: the fetched url
_protocol: The protocol used to get the response.
_port: The remote port number used to get the response.
_status: The integer status code of the response
_request_type: What caused this request (Document, XHR, etc)
_flags: A Namespace object from ParseFlags()
def __init__(self, response_headers, request_headers, url, protocol, port,
status, request_type):
self._response_headers = {}
self._request_headers = {}
self._url = url
self._protocol = protocol
self._port = port
self._status = status
self._request_type = request_type
self._flags = ParseFlags()
# Make all header names lower case.
for name in response_headers:
self._response_headers[name.lower()] = response_headers[name]
for name in request_headers:
self._request_headers[name.lower()] = request_headers[name]
def __str__(self):
self_dict = {
'response_headers': self._response_headers,
'request_headers': self._request_headers,
'url': self._url,
'protocol': self._protocol,
'port': self._port,
'status': self._status,
'request_type': self._request_type
return json.dumps(self_dict, indent=2)
def response_headers(self):
return self._response_headers
def request_headers(self):
return self._request_headers
def url(self):
return self._url
def protocol(self):
return self._protocol
def port(self):
return self._port
def status(self):
return self._status
def request_type(self):
return self._request_type
def ResponseHasViaHeader(self):
return 'via' in self._response_headers and (self._response_headers['via'] ==
def WasXHR(self):
return self.request_type == 'XHR'
def UsedHTTP(self):
return self._protocol == 'http/1.1'
def UsedHTTP2(self):
return self._protocol == 'h2'
class IntegrationTest(unittest.TestCase):
"""This class adds ChromeProxy-specific assertions to the generic
unittest.TestCase class.
def assertHasChromeProxyViaHeader(self, http_response):
"""Asserts that the Via header in the given HTTPResponse matches the
expected value as given on the command line.
http_response: The HTTPResponse object to check.
self.assertIn('via', http_response.response_headers)
expected_via_header = ParseFlags().via_header_value
actual_via_headers = http_response.response_headers['via'].split(',')
self.assertIn(expected_via_header, actual_via_headers, "Via header not in "
"response headers! Expected: %s, Actual: %s" %
(expected_via_header, actual_via_headers))
def assertNotHasChromeProxyViaHeader(self, http_response):
"""Asserts that the Via header in the given HTTPResponse does not match the
expected value as given on the command line.
http_response: The HTTPResponse object to check.
if 'via' in http_response.response_headers:
expected_via_header = ParseFlags().via_header_value
actual_via_headers = http_response.response_headers['via'].split(',')
self.assertNotIn(expected_via_header, actual_via_headers, "Via header "
"found in response headers! Not expected: %s, Actual: %s" %
(expected_via_header, actual_via_headers))
def checkLoFiResponse(self, http_response, expected_lo_fi):
"""Asserts that if expected the response headers contain the Lo-Fi directive
then the request headers do too. If the CPAT header contains if-heavy, the
request should not be LoFi. If-heavy will be deprecated in the future. Also
checks that the content size is less than 100 if |expected_lo_fi|.
Otherwise, checks that the response and request headers don't contain the
Lo-Fi directive and the content size is greater than 100.
http_response: The HTTPResponse object to check.
expected_lo_fi: Whether the response should be Lo-Fi.
Whether the response was Lo-Fi.
if (expected_lo_fi) :
content_length = http_response.response_headers['content-length']
cpat_request = http_response.request_headers[
cpct_response = http_response.response_headers[
if ('empty-image' in cpct_response):
self.assertIn('empty-image', cpat_request)
self.assertTrue(int(content_length) < 100)
return True;
return False;
if ('chrome-proxy-accept-transform' in http_response.request_headers):
cpat_request = http_response.request_headers[
if ('empty-image' in cpat_request):
self.assertIn('if-heavy', cpat_request)
content_length = http_response.response_headers['content-length']
self.assertTrue(int(content_length) > 100)
return False;
def checkLitePageResponse(self, http_response):
"""Asserts that if the response headers contain the Lite Page directive then
the request headers do too.
http_response: The HTTPResponse object to check.
Whether the response was a Lite Page.
if ('chrome-proxy-content-transform' not in http_response.response_headers):
return False;
cpct_response = http_response.response_headers[
cpat_request = http_response.request_headers[
if ('lite-page' in cpct_response):
self.assertIn('lite-page', cpat_request)
return True;
return False;
def RunAllTests(run_all_tests=False):
"""A simple helper method to run all tests using unittest.main().
run_all_tests: If True, all tests in the directory will be run, Otherwise
only the tests in the file given on the command line will be run.
the TestResult object from the test runner
flags = ParseFlags()
logger = GetLogger()
logger.debug('Command line args: %s', str(sys.argv))'sys.argv parsed to %s', str(flags))
if flags.catch:
# Use python's test discovery to locate tests files that have subclasses of
# unittest.TestCase and methods beginning with 'test'.
pattern = '*.py' if run_all_tests else os.path.basename(sys.argv[0])
loader = unittest.TestLoader()
test_suite_iter =,
# Match class and method name on the given test filter from --test_filter.
tests = unittest.TestSuite()
test_filter_re = flags.test_filter.replace('.', r'\.').replace('*', '.*')
for test_suite in test_suite_iter:
for test_case in test_suite:
for test in test_case:
# Drop the file name in the form <filename>.<classname>.<methodname>
test_id =['.') + 1:]
if re.match(test_filter_re, test_id):
testRunner = unittest.runner.TextTestRunner(verbosity=2,
failfast=flags.failfast, buffer=(not flags.disable_buffer))