blob: e82aacd924333ca33735348f0f0fe723face4103 [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 inspect
import logging
import re
import unittest
from py_utils import cloud_storage
from telemetry.internal.browser import browser_finder
from telemetry.internal.browser import browser_finder_exceptions
from telemetry.internal.platform import android_device
from telemetry.internal.results import artifact_logger
from telemetry.testing import browser_test_context
from typ import json_results
from typ import test_case
DEFAULT_LOG_FORMAT = (
'(%(levelname)s) %(asctime)s %(module)s.%(funcName)s:%(lineno)d '
'%(message)s')
class SeriallyExecutedBrowserTestCase(test_case.TestCase):
# Below is a reference to the typ.Runner instance. It will be used in
# member functions like GetExpectationsForTest() to get test information
# from the typ.Runner instance running the test.
_typ_runner = None
def __init__(self, methodName):
super(SeriallyExecutedBrowserTestCase, self).__init__(methodName)
self._private_methodname = methodName
def shortName(self):
"""Returns the method name this test runs, without the package prefix."""
return self._private_methodname
def set_artifacts(self, artifacts):
super(SeriallyExecutedBrowserTestCase, self).set_artifacts(artifacts)
artifact_logger.RegisterArtifactImplementation(artifacts)
def setUp(self):
if hasattr(self, 'browser') and self.browser:
self.browser.CleanupUnsymbolizedMinidumps()
def tearDown(self):
if hasattr(self, 'browser') and self.browser:
self.browser.CleanupUnsymbolizedMinidumps(fatal=True)
@classmethod
def Name(cls):
return cls.__name__
@classmethod
def AddCommandlineArgs(cls, parser):
pass
@classmethod
def SetUpProcess(cls):
""" Set up testing logic before running the test case.
This is guaranteed to be called only once for all the tests before the test
suite runs.
"""
finder_options = browser_test_context.GetCopy().finder_options
cls._finder_options = finder_options
# Set up logging based on the verbosity passed from the parent to
# the child process.
if finder_options.verbosity >= 2:
logging.getLogger().setLevel(logging.DEBUG)
elif finder_options.verbosity:
logging.getLogger().setLevel(logging.INFO)
else:
logging.getLogger().setLevel(logging.WARNING)
logging.basicConfig(format=DEFAULT_LOG_FORMAT)
cls.platform = None
cls.browser = None
cls._browser_to_create = None
cls._browser_options = None
@classmethod
def SetBrowserOptions(cls, browser_options):
"""Sets the browser option for the browser to create.
Args:
browser_options: Browser options object for the browser we want to test.
"""
cls._browser_options = browser_options
cls._browser_to_create = browser_finder.FindBrowser(browser_options)
if not cls._browser_to_create:
raise browser_finder_exceptions.BrowserFinderException(
'Cannot find browser of type %s. \n\nAvailable browsers:\n%s\n' % (
browser_options.browser_options.browser_type,
'\n'.join(browser_finder.GetAllAvailableBrowserTypes(
browser_options))))
if not cls.platform:
cls.platform = cls._browser_to_create.platform
cls.platform.SetPerformanceMode(android_device.HIGH_PERFORMANCE_MODE)
cls.platform.network_controller.Open(
browser_options.browser_options.wpr_mode)
else:
assert cls.platform == cls._browser_to_create.platform, (
'All browser launches within same test suite must use browsers on '
'the same platform')
@classmethod
def StartWPRServer(cls, archive_path=None, archive_bucket=None):
"""Start a webpage replay server.
Args:
archive_path: Path to the WPR file. If there is a corresponding sha1 file,
this archive will be automatically downloaded from Google Storage.
archive_bucket: The bucket to look for the WPR archive.
"""
assert cls._browser_options, (
'Browser options must be set with |SetBrowserOptions| prior to '
'starting WPR')
assert not cls.browser, 'WPR must be started prior to browser being started'
cloud_storage.GetIfChanged(archive_path, archive_bucket)
cls.platform.network_controller.StartReplay(archive_path)
@classmethod
def StopWPRServer(cls):
cls.platform.network_controller.StopReplay()
@classmethod
def StartBrowser(cls):
assert cls._browser_options, (
'Browser options must be set with |SetBrowserOptions| prior to '
'starting WPR')
assert not cls.browser, 'Browser is started. Must close it first'
try:
# TODO(crbug.com/803104): Note cls._browser_options actually is a
# FinderOptions object, and we need to access the real browser_option's
# contained inside.
cls._browser_to_create.SetUpEnvironment(
cls._browser_options.browser_options)
cls.browser = cls._browser_to_create.Create()
specifiers = set(cls.GetPlatformTags(cls.browser) +
cls._browser_to_create.GetTypExpectationsTags())
if cls._typ_runner.has_expectations and specifiers:
logging.info(
'The following expectations condition tags were generated %s' %
str(list(specifiers)))
cls._typ_runner.expectations.add_tags(specifiers)
except Exception:
cls._browser_to_create.CleanUpEnvironment()
raise
@classmethod
def StopBrowser(cls):
assert cls.browser, 'Browser is not started'
try:
cls.browser.Close()
cls.browser = None
finally:
cls._browser_to_create.CleanUpEnvironment()
@classmethod
def TearDownProcess(cls):
""" Tear down the testing logic after running the test cases.
This is guaranteed to be called only once for all the tests after the test
suite finishes running.
"""
if cls.platform:
cls.platform.StopAllLocalServers()
cls.platform.network_controller.Close()
cls.platform.SetPerformanceMode(android_device.NORMAL_PERFORMANCE_MODE)
if cls.browser:
cls.StopBrowser()
@classmethod
def SetStaticServerDirs(cls, dirs_path):
assert cls.platform
assert isinstance(dirs_path, list)
cls.platform.SetHTTPServerDirectories(dirs_path)
@classmethod
def UrlOfStaticFilePath(cls, file_path):
return cls.platform.http_server.UrlOf(file_path)
@classmethod
def ExpectationsFiles(cls):
"""Subclasses can override this class method to return a list of absolute
paths to the test expectations files.
Returns:
A list of test expectations file paths. The paths must be absolute.
"""
return []
@classmethod
def IgnoredTags(cls):
"""Subclasses can override this class method to return a list of typ
expectation tags to ignore for the purposes of tag validation.
This is so that expectation files don't need to get cluttered with
unnecessary tags, e.g. if two tags are functionally equivalent or if some
tag is completely unnecessary for a particular test type.
Returns:
A list of typ tags.
"""
return []
def GetExpectationsForTest(self):
"""Subclasses can override this method to return a tuple containing a set
of expected results and a flag indicating if the test has the RetryOnFailure
expectation. Tests members may want to know the test expectation in order to
modify its behavior for certain expectations. For instance GPU tests want
to avoid symbolizing any crash dumps in the case of expected test failures
or when tests are being retried because they are expected to be flaky.
Returns:
A tuple containing set of expected results for a test and a boolean value
indicating if the test contains the RetryOnFailure expectation. When there
are no expectations files passed to typ, then a tuple of
(set(['PASS']), False) should be returned from this function.
"""
exp = self.__class__._typ_runner.expectations_for(self)
return exp.results, exp.should_retry_on_failure
@classmethod
def GetPlatformTags(cls, browser):
"""This method uses the Browser instances's platform member variable to get
the operating system, operating system version and browser type tags.
Example tags for operating system are 'linux' and 'mac'. Example tags
for the operating system version are 'mojave' and 'vista'. Example tags
for browser type are 'debug' and 'release'. If a None value or empty string
is retrieved from the browser's platform member variable, then it will be
filtered out.
Args:
Browser instance returned from the possible_browser.BrowserSession() method.
Returns:
A list of tags derived from the Browser instance's platform member variable.
"""
return browser.GetTypExpectationsTags()
@staticmethod
def GetJSONResultsDelimiter():
"""This method returns the path delimiter that will be used to seperate
a test name into parts. By default, the delimiter is '.'
"""
return json_results.DEFAULT_TEST_SEPARATOR
def LoadAllTestsInModule(module):
""" Load all tests & generated browser tests in a given module.
This is supposed to be invoke in load_tests() method of your test modules that
use browser_test_runner framework to discover & generate the tests to be
picked up by the test runner. Here is the example of how your test module
should looks like:
################## my_awesome_browser_tests.py ################
import sys
from telemetry.testing import serially_executed_browser_test_case
...
class TestSimpleBrowser(
serially_executed_browser_test_case.SeriallyExecutedBrowserTestCase):
...
...
def load_tests(loader, tests, pattern):
return serially_executed_browser_test_case.LoadAllTestsInModule(
sys.modules[__name__])
#################################################################
Args:
module: the module which contains test cases classes.
Returns:
an instance of unittest.TestSuite, which contains all the tests & generated
test cases to be run.
"""
suite = unittest.TestSuite()
test_context = browser_test_context.GetCopy()
if not test_context:
return suite
for _, obj in inspect.getmembers(module):
if (inspect.isclass(obj) and
issubclass(obj, SeriallyExecutedBrowserTestCase)):
# We bail out early if this class doesn't match the targeted
# test_class in test_context to avoid calling GenerateTestCases
# for tests that we don't intend to run. This is to avoid possible errors
# in GenerateTestCases as the test class may define custom options in
# the finder_options object, and hence would raise error if they can't
# find their custom options in finder_options object.
if test_context.test_class != obj:
continue
for test in GenerateTestCases(
test_class=obj, finder_options=test_context.finder_options):
if test.id() in test_context.test_case_ids_to_run:
suite.addTest(test)
return suite
def _GenerateTestMethod(based_method, args):
return lambda self: based_method(self, *args)
_TEST_GENERATOR_PREFIX = 'GenerateTestCases_'
_INVALID_TEST_NAME_RE = re.compile(r'[^a-zA-Z0-9_\.\\\/-]')
def _ValidateTestMethodname(test_name):
assert not bool(_INVALID_TEST_NAME_RE.search(test_name))
def GenerateTestCases(test_class, finder_options):
test_cases = []
for name, method in inspect.getmembers(
test_class, predicate=inspect.ismethod):
if name.startswith('test'):
# Do not allow method names starting with "test" in these
# subclasses, to avoid collisions with Python's unit test runner.
raise Exception('Name collision with Python\'s unittest runner: %s' %
name)
elif name.startswith('Test'):
# Pass these through for the time being. We may want to rethink
# how they are handled in the future.
test_cases.append(test_class(name))
elif name.startswith(_TEST_GENERATOR_PREFIX):
based_method_name = name[len(_TEST_GENERATOR_PREFIX):]
assert hasattr(test_class, based_method_name), (
'%s is specified but based method %s does not exist' %
(name, based_method_name))
based_method = getattr(test_class, based_method_name)
for generated_test_name, args in method(finder_options):
_ValidateTestMethodname(generated_test_name)
setattr(test_class, generated_test_name, _GenerateTestMethod(
based_method, args))
test_cases.append(test_class(generated_test_name))
return test_cases