Adding upstream version 115.8.0esr.upstream/115.8.0esr

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

3 files changed, 640 insertions, 0 deletions

diff --git a/testing/marionette/harness/marionette_harness/marionette_test/__init__.py b/testing/marionette/harness/marionette_harness/marionette_test/__init__.py
new file mode 100644
index 0000000000..436a282f26
--- /dev/null
+++ b/testing/marionette/harness/marionette_harness/marionette_test/__init__.py
@@ -0,0 +1,24 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+__version__ = "3.1.0"
+
+from unittest.case import SkipTest, skip
+
+from .decorators import (
+    parameterized,
+    run_if_manage_instance,
+    skip_if_chrome,
+    skip_if_desktop,
+    skip_unless_browser_pref,
+    skip_unless_protocol,
+    with_parameters,
+)
+from .testcases import (
+    CommonTestCase,
+    MarionetteTestCase,
+    MetaParameterized,
+    expectedFailure,
+    unexpectedSuccess,
+)
diff --git a/testing/marionette/harness/marionette_harness/marionette_test/decorators.py b/testing/marionette/harness/marionette_harness/marionette_test/decorators.py
new file mode 100644
index 0000000000..cc3aa091d8
--- /dev/null
+++ b/testing/marionette/harness/marionette_harness/marionette_test/decorators.py
@@ -0,0 +1,194 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+import functools
+import types
+from unittest.case import SkipTest
+
+
+def parameterized(func_suffix, *args, **kwargs):
+    r"""Decorator which generates methods given a base method and some data.
+
+    **func_suffix** is used as a suffix for the new created method and must be
+    unique given a base method. if **func_suffix** countains characters that
+    are not allowed in normal python function name, these characters will be
+    replaced with "_".
+
+    This decorator can be used more than once on a single base method. The class
+    must have a metaclass of :class:`MetaParameterized`.
+
+    Example::
+
+      # This example will generate two methods:
+      #
+      # - MyTestCase.test_it_1
+      # - MyTestCase.test_it_2
+      #
+      class MyTestCase(MarionetteTestCase):
+          @parameterized("1", 5, named='name')
+          @parameterized("2", 6, named='name2')
+          def test_it(self, value, named=None):
+              print value, named
+
+    :param func_suffix: will be used as a suffix for the new method
+    :param \*args: arguments to pass to the new method
+    :param \*\*kwargs: named arguments to pass to the new method
+    """
+
+    def wrapped(func):
+        if not hasattr(func, "metaparameters"):
+            func.metaparameters = []
+        func.metaparameters.append((func_suffix, args, kwargs))
+        return func
+
+    return wrapped
+
+
+def run_if_manage_instance(reason):
+    """Decorator which runs a test if Marionette manages the application instance."""
+
+    def decorator(test_item):
+        if not isinstance(test_item, types.FunctionType):
+            raise Exception("Decorator only supported for functions")
+
+        @functools.wraps(test_item)
+        def skip_wrapper(self, *args, **kwargs):
+            if self.marionette.instance is None:
+                raise SkipTest(reason)
+            return test_item(self, *args, **kwargs)
+
+        return skip_wrapper
+
+    return decorator
+
+
+def skip_if_chrome(reason):
+    """Decorator which skips a test if chrome context is active."""
+
+    def decorator(test_item):
+        if not isinstance(test_item, types.FunctionType):
+            raise Exception("Decorator only supported for functions")
+
+        @functools.wraps(test_item)
+        def skip_wrapper(self, *args, **kwargs):
+            if self.marionette._send_message("getContext", key="value") == "chrome":
+                raise SkipTest(reason)
+            return test_item(self, *args, **kwargs)
+
+        return skip_wrapper
+
+    return decorator
+
+
+def skip_if_desktop(reason):
+    """Decorator which skips a test if run on desktop."""
+
+    def decorator(test_item):
+        if not isinstance(test_item, types.FunctionType):
+            raise Exception("Decorator only supported for functions")
+
+        @functools.wraps(test_item)
+        def skip_wrapper(self, *args, **kwargs):
+            if self.marionette.session_capabilities.get("browserName") == "firefox":
+                raise SkipTest(reason)
+            return test_item(self, *args, **kwargs)
+
+        return skip_wrapper
+
+    return decorator
+
+
+def skip_unless_browser_pref(reason, pref, predicate=bool):
+    """Decorator which skips a test based on the value of a browser preference.
+
+    :param reason: Message describing why the test need to be skipped.
+    :param pref: the preference name
+    :param predicate: a function that should return false to skip the test.
+                      The function takes one parameter, the preference value.
+                      Defaults to the python built-in bool function.
+
+    Note that the preference must exist, else a failure is raised.
+
+    Example: ::
+
+      class TestSomething(MarionetteTestCase):
+          @skip_unless_browser_pref("Sessionstore needs to be enabled for crashes",
+                                    "browser.sessionstore.resume_from_crash",
+                                    lambda value: value is True,
+                                    )
+          def test_foo(self):
+              pass  # test implementation here
+
+    """
+
+    def decorator(test_item):
+        if not isinstance(test_item, types.FunctionType):
+            raise Exception("Decorator only supported for functions")
+        if not callable(predicate):
+            raise ValueError("predicate must be callable")
+
+        @functools.wraps(test_item)
+        def skip_wrapper(self, *args, **kwargs):
+            value = self.marionette.get_pref(pref)
+            if value is None:
+                self.fail("No such browser preference: {0!r}".format(pref))
+            if not predicate(value):
+                raise SkipTest(reason)
+            return test_item(self, *args, **kwargs)
+
+        return skip_wrapper
+
+    return decorator
+
+
+def skip_unless_protocol(reason, predicate):
+    """Decorator which skips a test if the predicate does not match the current protocol level."""
+
+    def decorator(test_item):
+        if not isinstance(test_item, types.FunctionType):
+            raise Exception("Decorator only supported for functions")
+        if not callable(predicate):
+            raise ValueError("predicate must be callable")
+
+        @functools.wraps(test_item)
+        def skip_wrapper(self, *args, **kwargs):
+            level = self.marionette.client.protocol
+            if not predicate(level):
+                raise SkipTest(reason)
+            return test_item(self, *args, **kwargs)
+
+        return skip_wrapper
+
+    return decorator
+
+
+def with_parameters(parameters):
+    """Decorator which generates methods given a base method and some data.
+
+    Acts like :func:`parameterized`, but define all methods in one call.
+
+    Example::
+
+      # This example will generate two methods:
+      #
+      # - MyTestCase.test_it_1
+      # - MyTestCase.test_it_2
+      #
+
+      DATA = [("1", [5], {'named':'name'}), ("2", [6], {'named':'name2'})]
+
+      class MyTestCase(MarionetteTestCase):
+          @with_parameters(DATA)
+          def test_it(self, value, named=None):
+              print value, named
+
+    :param parameters: list of tuples (**func_suffix**, **args**, **kwargs**)
+                       defining parameters like in :func:`todo`.
+    """
+
+    def wrapped(func):
+        func.metaparameters = parameters
+        return func
+
+    return wrapped
diff --git a/testing/marionette/harness/marionette_harness/marionette_test/testcases.py b/testing/marionette/harness/marionette_harness/marionette_test/testcases.py
new file mode 100644
index 0000000000..84a6e09665
--- /dev/null
+++ b/testing/marionette/harness/marionette_harness/marionette_test/testcases.py
@@ -0,0 +1,422 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+
+import os
+import re
+import sys
+import time
+import unittest
+import warnings
+import weakref
+from unittest.case import SkipTest
+
+import six
+from marionette_driver.errors import TimeoutException, UnresponsiveInstanceException
+from mozfile import load_source
+from mozlog import get_default_logger
+
+
+# With Python 3 both expectedFailure and unexpectedSuccess are
+# available in unittest/case.py but won't work here because both
+# do not inherit from BaseException. And that's currently needed
+# in our custom test status handling in `run()`.
+class expectedFailure(Exception):
+    """
+    Raise this when a test is expected to fail.
+
+    This is an implementation detail.
+    """
+
+    def __init__(self, exc_info):
+        super(expectedFailure, self).__init__()
+        self.exc_info = exc_info
+
+
+class unexpectedSuccess(Exception):
+    """
+    The test was supposed to fail, but it didn't!
+    """
+
+    pass
+
+
+def _wraps_parameterized(func, func_suffix, args, kwargs):
+    """Internal: Decorator used in class MetaParameterized."""
+
+    def wrapper(self):
+        return func(self, *args, **kwargs)
+
+    wrapper.__name__ = func.__name__ + "_" + str(func_suffix)
+    wrapper.__doc__ = "[{0}] {1}".format(func_suffix, func.__doc__)
+    return wrapper
+
+
+class MetaParameterized(type):
+    """
+    A metaclass that allow a class to use decorators.
+
+    It can be used like :func:`parameterized`
+    or :func:`with_parameters` to generate new methods.
+    """
+
+    RE_ESCAPE_BAD_CHARS = re.compile(r"[\.\(\) -/]")
+
+    def __new__(cls, name, bases, attrs):
+        for k, v in list(attrs.items()):
+            if callable(v) and hasattr(v, "metaparameters"):
+                for func_suffix, args, kwargs in v.metaparameters:
+                    func_suffix = cls.RE_ESCAPE_BAD_CHARS.sub("_", func_suffix)
+                    wrapper = _wraps_parameterized(v, func_suffix, args, kwargs)
+                    if wrapper.__name__ in attrs:
+                        raise KeyError(
+                            "{0} is already a defined method on {1}".format(
+                                wrapper.__name__, name
+                            )
+                        )
+                    attrs[wrapper.__name__] = wrapper
+                del attrs[k]
+
+        return type.__new__(cls, name, bases, attrs)
+
+
+@six.add_metaclass(MetaParameterized)
+class CommonTestCase(unittest.TestCase):
+
+    match_re = None
+    failureException = AssertionError
+    pydebugger = None
+
+    def __init__(self, methodName, marionette_weakref, fixtures, **kwargs):
+        super(CommonTestCase, self).__init__(methodName)
+        self.methodName = methodName
+
+        self._marionette_weakref = marionette_weakref
+        self.fixtures = fixtures
+
+        self.duration = 0
+        self.start_time = 0
+        self.expected = kwargs.pop("expected", "pass")
+        self.logger = get_default_logger()
+
+    def _enter_pm(self):
+        if self.pydebugger:
+            self.pydebugger.post_mortem(sys.exc_info()[2])
+
+    def _addSkip(self, result, reason):
+        addSkip = getattr(result, "addSkip", None)
+        if addSkip is not None:
+            addSkip(self, reason)
+        else:
+            warnings.warn(
+                "TestResult has no addSkip method, skips not reported",
+                RuntimeWarning,
+                2,
+            )
+            result.addSuccess(self)
+
+    def assertRaisesRegxp(
+        self, expected_exception, expected_regexp, callable_obj=None, *args, **kwargs
+    ):
+        return six.assertRaisesRegex(
+            self,
+            expected_exception,
+            expected_regexp,
+            callable_obj=None,
+            *args,
+            **kwargs
+        )
+
+    def run(self, result=None):
+        # Bug 967566 suggests refactoring run, which would hopefully
+        # mean getting rid of this inner function, which only sits
+        # here to reduce code duplication:
+        def expected_failure(result, exc_info):
+            addExpectedFailure = getattr(result, "addExpectedFailure", None)
+            if addExpectedFailure is not None:
+                addExpectedFailure(self, exc_info)
+            else:
+                warnings.warn(
+                    "TestResult has no addExpectedFailure method, "
+                    "reporting as passes",
+                    RuntimeWarning,
+                )
+                result.addSuccess(self)
+
+        self.start_time = time.time()
+        orig_result = result
+        if result is None:
+            result = self.defaultTestResult()
+            startTestRun = getattr(result, "startTestRun", None)
+            if startTestRun is not None:
+                startTestRun()
+
+        result.startTest(self)
+
+        testMethod = getattr(self, self._testMethodName)
+        if getattr(self.__class__, "__unittest_skip__", False) or getattr(
+            testMethod, "__unittest_skip__", False
+        ):
+            # If the class or method was skipped.
+            try:
+                skip_why = getattr(
+                    self.__class__, "__unittest_skip_why__", ""
+                ) or getattr(testMethod, "__unittest_skip_why__", "")
+                self._addSkip(result, skip_why)
+            finally:
+                result.stopTest(self)
+            self.stop_time = time.time()
+            return
+        try:
+            success = False
+            try:
+                if self.expected == "fail":
+                    try:
+                        self.setUp()
+                    except Exception:
+                        raise expectedFailure(sys.exc_info())
+                else:
+                    self.setUp()
+            except SkipTest as e:
+                self._addSkip(result, str(e))
+            except (KeyboardInterrupt, UnresponsiveInstanceException):
+                raise
+            except expectedFailure as e:
+                expected_failure(result, e.exc_info)
+            except Exception:
+                self._enter_pm()
+                result.addError(self, sys.exc_info())
+            else:
+                try:
+                    if self.expected == "fail":
+                        try:
+                            testMethod()
+                        except Exception:
+                            raise expectedFailure(sys.exc_info())
+                        raise unexpectedSuccess
+                    else:
+                        testMethod()
+                except self.failureException:
+                    self._enter_pm()
+                    result.addFailure(self, sys.exc_info())
+                except (KeyboardInterrupt, UnresponsiveInstanceException):
+                    raise
+                except expectedFailure as e:
+                    expected_failure(result, e.exc_info)
+                except unexpectedSuccess:
+                    addUnexpectedSuccess = getattr(result, "addUnexpectedSuccess", None)
+                    if addUnexpectedSuccess is not None:
+                        addUnexpectedSuccess(self)
+                    else:
+                        warnings.warn(
+                            "TestResult has no addUnexpectedSuccess method, "
+                            "reporting as failures",
+                            RuntimeWarning,
+                        )
+                        result.addFailure(self, sys.exc_info())
+                except SkipTest as e:
+                    self._addSkip(result, str(e))
+                except Exception:
+                    self._enter_pm()
+                    result.addError(self, sys.exc_info())
+                else:
+                    success = True
+                try:
+                    if self.expected == "fail":
+                        try:
+                            self.tearDown()
+                        except Exception:
+                            raise expectedFailure(sys.exc_info())
+                    else:
+                        self.tearDown()
+                except (KeyboardInterrupt, UnresponsiveInstanceException):
+                    raise
+                except expectedFailure as e:
+                    expected_failure(result, e.exc_info)
+                except Exception:
+                    self._enter_pm()
+                    result.addError(self, sys.exc_info())
+                    success = False
+            # Here we could handle doCleanups() instead of calling cleanTest directly
+            self.cleanTest()
+
+            if success:
+                result.addSuccess(self)
+
+        finally:
+            result.stopTest(self)
+            if orig_result is None:
+                stopTestRun = getattr(result, "stopTestRun", None)
+                if stopTestRun is not None:
+                    stopTestRun()
+
+    @classmethod
+    def match(cls, filename):
+        """Determine if the specified filename should be handled by this test class.
+
+        This is done by looking for a match for the filename using cls.match_re.
+        """
+        if not cls.match_re:
+            return False
+        m = cls.match_re.match(filename)
+        return m is not None
+
+    @classmethod
+    def add_tests_to_suite(
+        cls,
+        mod_name,
+        filepath,
+        suite,
+        testloader,
+        marionette,
+        fixtures,
+        testvars,
+        **kwargs
+    ):
+        """Add all the tests in the specified file to the specified suite."""
+        raise NotImplementedError
+
+    @property
+    def test_name(self):
+        rel_path = None
+        if os.path.exists(self.filepath):
+            rel_path = self._fix_test_path(self.filepath)
+
+        return "{0} {1}.{2}".format(
+            rel_path, self.__class__.__name__, self._testMethodName
+        )
+
+    def id(self):
+        # TBPL starring requires that the "test name" field of a failure message
+        # not differ over time. The test name to be used is passed to
+        # mozlog via the test id, so this is overriden to maintain
+        # consistency.
+        return self.test_name
+
+    def setUp(self):
+        # Convert the marionette weakref to an object, just for the
+        # duration of the test; this is deleted in tearDown() to prevent
+        # a persistent circular reference which in turn would prevent
+        # proper garbage collection.
+        self.start_time = time.time()
+        self.marionette = self._marionette_weakref()
+        if self.marionette.session is None:
+            self.marionette.start_session()
+        self.marionette.timeout.reset()
+
+        super(CommonTestCase, self).setUp()
+
+    def cleanTest(self):
+        self._delete_session()
+
+    def _delete_session(self):
+        if hasattr(self, "start_time"):
+            self.duration = time.time() - self.start_time
+        if self.marionette.session is not None:
+            try:
+                self.marionette.delete_session()
+            except IOError:
+                # Gecko has crashed?
+                pass
+        self.marionette = None
+
+    def _fix_test_path(self, path):
+        """Normalize a logged test path from the test package."""
+        test_path_prefixes = [
+            "tests{}".format(os.path.sep),
+        ]
+
+        path = os.path.relpath(path)
+        for prefix in test_path_prefixes:
+            if path.startswith(prefix):
+                path = path[len(prefix) :]
+                break
+        path = path.replace("\\", "/")
+
+        return path
+
+
+class MarionetteTestCase(CommonTestCase):
+
+    match_re = re.compile(r"test_(.*)\.py$")
+
+    def __init__(
+        self, marionette_weakref, fixtures, methodName="runTest", filepath="", **kwargs
+    ):
+        self.filepath = filepath
+        self.testvars = kwargs.pop("testvars", None)
+
+        super(MarionetteTestCase, self).__init__(
+            methodName,
+            marionette_weakref=marionette_weakref,
+            fixtures=fixtures,
+            **kwargs
+        )
+
+    @classmethod
+    def add_tests_to_suite(
+        cls,
+        mod_name,
+        filepath,
+        suite,
+        testloader,
+        marionette,
+        fixtures,
+        testvars,
+        **kwargs
+    ):
+        # since load_source caches modules, if a module is loaded with the same
+        # name as another one the module would just be reloaded.
+        #
+        # We may end up by finding too many test in a module then since reload()
+        # only update the module dict (so old keys are still there!) see
+        # https://docs.python.org/2/library/functions.html#reload
+        #
+        # we get rid of that by removing the module from sys.modules, so we
+        # ensure that it will be fully loaded by the imp.load_source call.
+
+        if mod_name in sys.modules:
+            del sys.modules[mod_name]
+
+        test_mod = load_source(mod_name, filepath)
+
+        for name in dir(test_mod):
+            obj = getattr(test_mod, name)
+            if isinstance(obj, six.class_types) and issubclass(obj, unittest.TestCase):
+                testnames = testloader.getTestCaseNames(obj)
+                for testname in testnames:
+                    suite.addTest(
+                        obj(
+                            weakref.ref(marionette),
+                            fixtures,
+                            methodName=testname,
+                            filepath=filepath,
+                            testvars=testvars,
+                            **kwargs
+                        )
+                    )
+
+    def setUp(self):
+        super(MarionetteTestCase, self).setUp()
+        self.marionette.test_name = self.test_name
+
+    def tearDown(self):
+        # In the case no session is active (eg. the application was quit), start
+        # a new session for clean-up steps.
+        if not self.marionette.session:
+            self.marionette.start_session()
+
+        self.marionette.test_name = None
+
+        super(MarionetteTestCase, self).tearDown()
+
+    def wait_for_condition(self, method, timeout=30):
+        timeout = float(timeout) + time.time()
+        while time.time() < timeout:
+            value = method(self.marionette)
+            if value:
+                return value
+            time.sleep(0.5)
+        else:
+            raise TimeoutException("wait_for_condition timed out")