| # 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 selenium.webdriver.chrome.options import Options |
| |
| def ParseFlags(): |
| """Parses the given command line arguments. |
| |
| Returns: |
| A new Namespace object with class properties for each argument added below. |
| See pydoc for argparse. |
| """ |
| def TestFilter(v): |
| try: |
| # 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) |
| except: |
| 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') |
| parser.add_argument('--via_header_value', |
| 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') |
| parser.add_argument('--android_package', |
| default='com.android.chrome', 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.', |
| action='store_true') |
| 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=*', |
| 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.') |
| 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. |
| |
| Args: |
| name: The name of the logger to return. |
| Returns: |
| 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': |
| logger.setLevel(logging.DEBUG) |
| elif logging_level == 'INFO': |
| logger.setLevel(logging.INFO) |
| elif logging_level == 'WARN': |
| logger.setLevel(logging.WARNING) |
| elif logging_level == 'ERROR': |
| logger.setLevel(logging.ERROR) |
| elif logging_level == 'CRIT': |
| logger.setLevel(logging.CRITICAL) |
| else: |
| logger.setLevel(logging.NOTSET) |
| formatter = logging.Formatter('%(asctime)s %(funcName)s() %(levelname)s: ' |
| '%(message)s') |
| if ParseFlags().log_file: |
| fh = logging.FileHandler(ParseFlags().log_file) |
| fh.setFormatter(formatter) |
| logger.addHandler(fh) |
| else: |
| ch = logging.StreamHandler(sys.stderr) |
| ch.setFormatter(formatter) |
| logger.addHandler(ch) |
| 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. |
| |
| Attributes: |
| _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 |
| fetched. |
| """ |
| |
| def __init__(self): |
| self._flags = ParseFlags() |
| self._driver = None |
| self._chrome_args = set() |
| self._url = '' |
| self._logger = GetLogger(name='TestDriver') |
| self._has_logs = False |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, exc_type, exc_value, tb): |
| if self._driver: |
| self._StopDriver() |
| |
| def _OverrideChromeArgs(self): |
| """Overrides any given arguments in the code with those given on the command |
| line. |
| |
| 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: |
| self._chrome_args.remove(original_args[arg_key]) |
| self._logger.info('Removed Chrome flag. %s', original_args[arg_key]) |
| self._chrome_args.add(override_arg) |
| self._logger.info('Added Chrome flag. %s', override_arg) |
| # Always add the flag that allows histograms to be queried in javascript. |
| self._chrome_args.add('--enable-stats-collection-bindings') |
| |
| 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. |
| """ |
| self._OverrideChromeArgs() |
| capabilities = { |
| 'loggingPrefs': {'performance': 'INFO'} |
| } |
| chrome_options = Options() |
| for arg in self._chrome_args: |
| chrome_options.add_argument(arg) |
| self._logger.info('Starting Chrome with these flags: %s', |
| str(self._chrome_args)) |
| if self._flags.android: |
| chrome_options.add_experimental_option('androidPackage', |
| self._flags.android_package) |
| self._logger.debug('Will use Chrome on Android') |
| elif self._flags.chrome_exec: |
| chrome_options.binary_location = self._flags.chrome_exec |
| self._logger.info('Using the Chrome binary at this path: %s', |
| self._flags.chrome_exec) |
| 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) |
| driver.command_executor._commands.update({ |
| 'getAvailableLogTypes': ('GET', '/session/$sessionId/log/types'), |
| 'getLog': ('POST', '/session/$sessionId/log')}) |
| self._driver = driver |
| |
| def _StopDriver(self): |
| """Nicely stops the ChromeDriver. |
| """ |
| self._logger.debug('Stopping ChromeDriver') |
| self._driver.quit() |
| self._driver = None |
| |
| def AddChromeArgs(self, args): |
| """Adds multiple arguments that will be passed to Chromium at start. |
| |
| Args: |
| args: An iterable of strings, each an argument to pass to Chrome at start. |
| """ |
| for arg in args: |
| self._chrome_args.add(arg) |
| |
| def AddChromeArg(self, arg): |
| """Adds a single argument that will be passed to Chromium at start. |
| |
| Args: |
| arg: a string argument to pass to Chrome at start |
| """ |
| self._chrome_args.add(arg) |
| self._logger.debug('Adding Chrome arg: %s', arg) |
| |
| def RemoveChromeArgs(self, args): |
| """Removes multiple arguments that will no longer be passed to Chromium at |
| start. |
| |
| Args: |
| args: An iterable of strings to no longer use the next time Chrome |
| starts. |
| """ |
| for arg in args: |
| self._chrome_args.discard(arg) |
| |
| def RemoveChromeArg(self, arg): |
| """Removes a single argument that will no longer be passed to Chromium at |
| start. |
| |
| Args: |
| arg: A string flag to no longer use the next time Chrome starts. |
| """ |
| self._chrome_args.discard(arg) |
| self._logger.debug('Removing Chrome arg: %s', arg) |
| |
| def ClearChromeArgs(self): |
| """Removes all arguments from Chromium at start. |
| """ |
| self._chrome_args.clear() |
| 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(window.chrome && chrome.benchmarking && ' |
| 'chrome.benchmarking.clearCache){chrome.benchmarking.clearCache(); ' |
| 'chrome.benchmarking.clearPredictorCache();chrome.benchmarking.' |
| 'clearHostResolverCache();}') |
| self._logger.info('Cleared browser cache. Returned=%s', str(res)) |
| |
| def LoadURL(self, url, timeout=30): |
| """Starts Chromium with any arguments previously given and navigates to the |
| given URL. |
| |
| Args: |
| 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._StartDriver() |
| self._driver.set_page_load_timeout(timeout) |
| self._logger.debug('Set page load timeout to %f seconds', timeout) |
| self._driver.get(self._url) |
| self._logger.debug('Loaded page %s', url) |
| self._has_logs = True |
| |
| 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. |
| |
| Args: |
| script: A string of Javascript code. |
| timeout: Timeout for the Javascript code to return in seconds. |
| Returns: |
| A string of the verbatim output from the Javascript execution. |
| """ |
| if not self._driver: |
| self._StartDriver() |
| # TODO(robertogden): Use 'driver.set_script_timeout(timeout)' instead after |
| # crbug/672114 is fixed. |
| default_timeout = socket.getdefaulttimeout() |
| socket.setdefaulttimeout(timeout) |
| 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) |
| socket.setdefaulttimeout(default_timeout) |
| 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> }();' |
| |
| Args: |
| script: A string of Javascript code. |
| timeout: Timeout for the Javascript code to return in seconds. |
| Returns: |
| 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. |
| |
| Args: |
| histogram: the name of the histogram to fetch |
| timeout: timeout for the underlying Javascript query. |
| |
| Returns: |
| 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, |
| max_poll=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|. |
| |
| Args: |
| 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: |
| time.sleep(poll_interval) |
| 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. |
| |
| Args: |
| method_filter: A regex expression to match the method of logged events |
| against. Only logs who's method matches the regex will be returned. |
| Returns: |
| Performance logs as a list of dicts, since the last time this function was |
| called. |
| """ |
| 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) |
| self._logger.info('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. |
| |
| Args: |
| 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. |
| Returns: |
| Whether the histogram exists |
| """ |
| histogram = {} |
| while(not histogram and sleep_intervals > 0): |
| histogram = self.GetHistogram(histogram_name, 5) |
| if (not histogram): |
| time.sleep(1) |
| 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. |
| |
| Args: |
| 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. |
| Returns: |
| A list of HTTPResponse objects, each representing a single completed HTTP |
| transaction by Chrome. |
| """ |
| 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 |
| has_domain): |
| all_responses.append(response) |
| else: |
| self._logger.info("Skipping HTTPResponse with url=%s in returned logs.", |
| response.url) |
| self._logger.info('%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 |
| Chrome. |
| |
| This class also includes several convenience functions for ChromeProxy |
| specific assertions. |
| |
| Attributes: |
| _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) |
| |
| @property |
| def response_headers(self): |
| return self._response_headers |
| |
| @property |
| def request_headers(self): |
| return self._request_headers |
| |
| @property |
| def url(self): |
| return self._url |
| |
| @property |
| def protocol(self): |
| return self._protocol |
| |
| @property |
| def port(self): |
| return self._port |
| |
| @property |
| def status(self): |
| return self._status |
| |
| @property |
| def request_type(self): |
| return self._request_type |
| |
| def ResponseHasViaHeader(self): |
| return 'via' in self._response_headers and (self._response_headers['via'] == |
| self._flags.via_header_value) |
| |
| 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. |
| |
| Args: |
| http_response: The HTTPResponse object to check. |
| """ |
| expected_via_header = ParseFlags().via_header_value |
| self.assertIn('via', http_response.response_headers) |
| self.assertEqual(expected_via_header, http_response.response_headers['via']) |
| |
| 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. |
| |
| Args: |
| http_response: The HTTPResponse object to check. |
| """ |
| expected_via_header = ParseFlags().via_header_value |
| self.assertNotIn('via', http_response.response_headers) |
| if 'via' in http_response.response_headers: |
| self.assertNotIn(expected_via_header, |
| http_response.response_headers['via']) |
| |
| 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. 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. |
| |
| Args: |
| http_response: The HTTPResponse object to check. |
| expected_lo_fi: Whether the response should be Lo-Fi. |
| |
| Returns: |
| Whether the response was Lo-Fi. |
| """ |
| |
| if (expected_lo_fi) : |
| self.assertHasChromeProxyViaHeader(http_response) |
| content_length = http_response.response_headers['content-length'] |
| cpat_request = http_response.request_headers[ |
| 'chrome-proxy-accept-transform'] |
| cpct_response = http_response.response_headers[ |
| 'chrome-proxy-content-transform'] |
| if ('empty-image' in cpct_response): |
| self.assertIn('empty-image', cpat_request) |
| self.assertTrue(int(content_length) < 100) |
| return True; |
| return False; |
| else: |
| self.assertNotIn('chrome-proxy-accept-transform', |
| http_response.request_headers) |
| self.assertNotIn('chrome-proxy-content-transform', |
| http_response.response_headers) |
| 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. |
| |
| Args: |
| http_response: The HTTPResponse object to check. |
| |
| Returns: |
| Whether the response was a Lite Page. |
| """ |
| |
| self.assertHasChromeProxyViaHeader(http_response) |
| if ('chrome-proxy-content-transform' not in http_response.response_headers): |
| return False; |
| cpct_response = http_response.response_headers[ |
| 'chrome-proxy-content-transform'] |
| cpat_request = http_response.request_headers[ |
| 'chrome-proxy-accept-transform'] |
| if ('lite-page' in cpct_response): |
| self.assertIn('lite-page', cpat_request) |
| return True; |
| return False; |
| |
| @staticmethod |
| def RunAllTests(run_all_tests=False): |
| """A simple helper method to run all tests using unittest.main(). |
| |
| Args: |
| 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. |
| """ |
| flags = ParseFlags() |
| logger = GetLogger() |
| logger.debug('Command line args: %s', str(sys.argv)) |
| logger.info('sys.argv parsed to %s', str(flags)) |
| if flags.catch: |
| unittest.installHandler() |
| # 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 = loader.discover(os.path.dirname(__file__), |
| pattern=pattern) |
| # 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 = test.id()[test.id().find('.') + 1:] |
| if re.match(test_filter_re, test_id): |
| tests.addTest(test) |
| testRunner = unittest.runner.TextTestRunner(verbosity=2, |
| failfast=flags.failfast, buffer=(not flags.disable_buffer)) |
| testRunner.run(tests) |
| |
| # Platform-specific decorators. |
| # These decorators can be used to only run a test function for certain platforms |
| # by annotating the function with them. |
| |
| def AndroidOnly(func): |
| def wrapper(*args, **kwargs): |
| if ParseFlags().android: |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test runs on Android only.') |
| return wrapper |
| |
| def NotAndroid(func): |
| def wrapper(*args, **kwargs): |
| if not ParseFlags().android: |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test does not run on Android.') |
| return wrapper |
| |
| def WindowsOnly(func): |
| def wrapper(*args, **kwargs): |
| if sys.platform == 'win32': |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test runs on Windows only.') |
| return wrapper |
| |
| def NotWindows(func): |
| def wrapper(*args, **kwargs): |
| if sys.platform != 'win32': |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test does not run on Windows.') |
| return wrapper |
| |
| def LinuxOnly(func): |
| def wrapper(*args, **kwargs): |
| if sys.platform.startswith('linux'): |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test runs on Linux only.') |
| return wrapper |
| |
| def NotLinux(func): |
| def wrapper(*args, **kwargs): |
| if sys.platform.startswith('linux'): |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test does not run on Linux.') |
| return wrapper |
| |
| def MacOnly(func): |
| def wrapper(*args, **kwargs): |
| if sys.platform == 'darwin': |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test runs on Mac OS only.') |
| return wrapper |
| |
| def NotMac(func): |
| def wrapper(*args, **kwargs): |
| if sys.platform == 'darwin': |
| func(*args, **kwargs) |
| else: |
| args[0].skipTest('This test does not run on Mac OS.') |
| return wrapper |
