path: root/testing/mozbase/moztest/moztest/results.py

# 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 time

import mozinfo
import six


class TestContext(object):
    """Stores context data about the test"""

    attrs = [
        "hostname",
        "arch",
        "env",
        "os",
        "os_version",
        "tree",
        "revision",
        "product",
        "logfile",
        "testgroup",
        "harness",
        "buildtype",
    ]

    def __init__(
        self,
        hostname="localhost",
        tree="",
        revision="",
        product="",
        logfile=None,
        arch="",
        operating_system="",
        testgroup="",
        harness="moztest",
        buildtype="",
    ):
        self.hostname = hostname
        self.arch = arch or mozinfo.processor
        self.env = os.environ.copy()
        self.os = operating_system or mozinfo.os
        self.os_version = mozinfo.version
        self.tree = tree
        self.revision = revision
        self.product = product
        self.logfile = logfile
        self.testgroup = testgroup
        self.harness = harness
        self.buildtype = buildtype

    def __str__(self):
        return "%s (%s, %s)" % (self.hostname, self.os, self.arch)

    def __repr__(self):
        return "<%s>" % self.__str__()

    def __eq__(self, other):
        if not isinstance(other, TestContext):
            return False
        diffs = [a for a in self.attrs if getattr(self, a) != getattr(other, a)]
        return len(diffs) == 0

    def __hash__(self):
        def get(attr):
            value = getattr(self, attr)
            if isinstance(value, dict):
                value = frozenset(six.iteritems(value))
            return value

        return hash(frozenset([get(a) for a in self.attrs]))


class TestResult(object):
    """Stores test result data"""

    FAIL_RESULTS = [
        "UNEXPECTED-PASS",
        "UNEXPECTED-FAIL",
        "ERROR",
    ]
    COMPUTED_RESULTS = FAIL_RESULTS + [
        "PASS",
        "KNOWN-FAIL",
        "SKIPPED",
    ]
    POSSIBLE_RESULTS = [
        "PASS",
        "FAIL",
        "SKIP",
        "ERROR",
    ]

    def __init__(
        self, name, test_class="", time_start=None, context=None, result_expected="PASS"
    ):
        """Create a TestResult instance.
        name = name of the test that is running
        test_class = the class that the test belongs to
        time_start = timestamp (seconds since UNIX epoch) of when the test started
                     running; if not provided, defaults to the current time
                     ! Provide 0 if you only have the duration
        context = TestContext instance; can be None
        result_expected = string representing the expected outcome of the test"""

        msg = "Result '%s' not in possible results: %s" % (
            result_expected,
            ", ".join(self.POSSIBLE_RESULTS),
        )
        assert isinstance(name, six.string_types), "name has to be a string"
        assert result_expected in self.POSSIBLE_RESULTS, msg

        self.name = name
        self.test_class = test_class
        self.context = context
        self.time_start = time_start if time_start is not None else time.time()
        self.time_end = None
        self._result_expected = result_expected
        self._result_actual = None
        self.result = None
        self.filename = None
        self.description = None
        self.output = []
        self.reason = None

    @property
    def test_name(self):
        return "%s.py %s.%s" % (
            self.test_class.split(".")[0],
            self.test_class,
            self.name,
        )

    def __str__(self):
        return "%s | %s (%s) | %s" % (
            self.result or "PENDING",
            self.name,
            self.test_class,
            self.reason,
        )

    def __repr__(self):
        return "<%s>" % self.__str__()

    def calculate_result(self, expected, actual):
        if actual == "ERROR":
            return "ERROR"
        if actual == "SKIP":
            return "SKIPPED"

        if expected == "PASS":
            if actual == "PASS":
                return "PASS"
            if actual == "FAIL":
                return "UNEXPECTED-FAIL"

        if expected == "FAIL":
            if actual == "PASS":
                return "UNEXPECTED-PASS"
            if actual == "FAIL":
                return "KNOWN-FAIL"

        # if actual is skip or error, we return at the beginning, so if we get
        # here it is definitely some kind of error
        return "ERROR"

    def infer_results(self, computed_result):
        assert computed_result in self.COMPUTED_RESULTS
        if computed_result == "UNEXPECTED-PASS":
            expected = "FAIL"
            actual = "PASS"
        elif computed_result == "UNEXPECTED-FAIL":
            expected = "PASS"
            actual = "FAIL"
        elif computed_result == "KNOWN-FAIL":
            expected = actual = "FAIL"
        elif computed_result == "SKIPPED":
            expected = actual = "SKIP"
        else:
            return
        self._result_expected = expected
        self._result_actual = actual

    def finish(self, result, time_end=None, output=None, reason=None):
        """Marks the test as finished, storing its end time and status
        ! Provide the duration as time_end if you only have that."""

        if result in self.POSSIBLE_RESULTS:
            self._result_actual = result
            self.result = self.calculate_result(
                self._result_expected, self._result_actual
            )
        elif result in self.COMPUTED_RESULTS:
            self.infer_results(result)
            self.result = result
        else:
            valid = self.POSSIBLE_RESULTS + self.COMPUTED_RESULTS
            msg = "Result '%s' not valid. Need one of: %s" % (result, ", ".join(valid))
            raise ValueError(msg)

        # use lists instead of multiline strings
        if isinstance(output, six.string_types):
            output = output.splitlines()

        self.time_end = time_end if time_end is not None else time.time()
        self.output = output or self.output
        self.reason = reason

    @property
    def finished(self):
        """Boolean saying if the test is finished or not"""
        return self.result is not None

    @property
    def duration(self):
        """Returns the time it took for the test to finish. If the test is
        not finished, returns the elapsed time so far"""
        if self.result is not None:
            return self.time_end - self.time_start
        else:
            # returns the elapsed time
            return time.time() - self.time_start


class TestResultCollection(list):
    """Container class that stores test results"""

    resultClass = TestResult

    def __init__(self, suite_name, time_taken=0, resultClass=None):
        list.__init__(self)
        self.suite_name = suite_name
        self.time_taken = time_taken
        if resultClass is not None:
            self.resultClass = resultClass

    def __str__(self):
        return "%s (%.2fs)\n%s" % (self.suite_name, self.time_taken, list.__str__(self))

    def subset(self, predicate):
        tests = self.filter(predicate)
        duration = 0
        sub = TestResultCollection(self.suite_name)
        for t in tests:
            sub.append(t)
            duration += t.duration
        sub.time_taken = duration
        return sub

    @property
    def contexts(self):
        """List of unique contexts for the test results contained"""
        cs = [tr.context for tr in self]
        return list(set(cs))

    def filter(self, predicate):
        """Returns a generator of TestResults that satisfy a given predicate"""
        return (tr for tr in self if predicate(tr))

    def tests_with_result(self, result):
        """Returns a generator of TestResults with the given result"""
        msg = "Result '%s' not in possible results: %s" % (
            result,
            ", ".join(self.resultClass.COMPUTED_RESULTS),
        )
        assert result in self.resultClass.COMPUTED_RESULTS, msg
        return self.filter(lambda t: t.result == result)

    @property
    def tests(self):
        """Generator of all tests in the collection"""
        return (t for t in self)

    def add_result(
        self,
        test,
        result_expected="PASS",
        result_actual="PASS",
        output="",
        context=None,
    ):
        def get_class(test):
            return test.__class__.__module__ + "." + test.__class__.__name__

        t = self.resultClass(
            name=str(test).split()[0],
            test_class=get_class(test),
            time_start=0,
            result_expected=result_expected,
            context=context,
        )
        t.finish(result_actual, time_end=0, reason=relevant_line(output), output=output)
        self.append(t)

    @property
    def num_failures(self):
        fails = 0
        for t in self:
            if t.result in self.resultClass.FAIL_RESULTS:
                fails += 1
        return fails

    def add_unittest_result(self, result, context=None):
        """Adds the python unittest result provided to the collection"""
        if hasattr(result, "time_taken"):
            self.time_taken += result.time_taken

        for test, output in result.errors:
            self.add_result(test, result_actual="ERROR", output=output)

        for test, output in result.failures:
            self.add_result(test, result_actual="FAIL", output=output)

        if hasattr(result, "unexpectedSuccesses"):
            for test in result.unexpectedSuccesses:
                self.add_result(test, result_expected="FAIL", result_actual="PASS")

        if hasattr(result, "skipped"):
            for test, output in result.skipped:
                self.add_result(
                    test, result_expected="SKIP", result_actual="SKIP", output=output
                )

        if hasattr(result, "expectedFailures"):
            for test, output in result.expectedFailures:
                self.add_result(
                    test, result_expected="FAIL", result_actual="FAIL", output=output
                )

        # unittest does not store these by default
        if hasattr(result, "tests_passed"):
            for test in result.tests_passed:
                self.add_result(test)

    @classmethod
    def from_unittest_results(cls, context, *results):
        """Creates a TestResultCollection containing the given python
        unittest results"""

        if not results:
            return cls("from unittest")

        # all the TestResult instances share the same context
        context = context or TestContext()

        collection = cls("from %s" % results[0].__class__.__name__)

        for result in results:
            collection.add_unittest_result(result, context)

        return collection


# used to get exceptions/errors from tracebacks
def relevant_line(s):
    KEYWORDS = ("Error:", "Exception:", "error:", "exception:")
    lines = s.splitlines()
    for line in lines:
        for keyword in KEYWORDS:
            if keyword in line:
                return line
    return "N/A"