# Copyright 2013 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Module containing base test results classes."""

import functools
import re
import sys
import threading

from lib.results import result_types

# This must match the source adding the suffix: bit.ly/3Zmwwyx
MULTIPROCESS_SUFFIX = '__multiprocess_mode'

# This must match the source adding the suffix: bit.ly/3Qt0Ww4
_NULL_MUTATION_SUFFIX = '__null_'
_MUTATION_SUFFIX_PATTERN = re.compile(r'^(.*)__([a-zA-Z]+)\.\.([a-zA-Z]+)_$')


class ResultType:
    """Class enumerating test types.

    Wraps the results defined in //build/util/lib/results/.
    """

    PASS = result_types.PASS
    SKIP = result_types.SKIP
    FAIL = result_types.FAIL
    CRASH = result_types.CRASH
    TIMEOUT = result_types.TIMEOUT
    UNKNOWN = result_types.UNKNOWN
    NOTRUN = result_types.NOTRUN

    @staticmethod
    def GetTypes():
        """Get a list of all test types."""
        return [
            ResultType.PASS,
            ResultType.SKIP,
            ResultType.FAIL,
            ResultType.CRASH,
            ResultType.TIMEOUT,
            ResultType.UNKNOWN,
            ResultType.NOTRUN,
        ]


@functools.total_ordering
class BaseTestResult:
    """Base class for a single test result."""

    def __init__(
        self,
        name,
        test_type,
        duration=0,
        log='',
        failure_reason=None,
        test_file=None,
    ):
        """Construct a BaseTestResult.

        Args:
          name: Name of the test which defines uniqueness.
          test_type: Type of the test result as defined in ResultType.
          duration: Time it took for the test to run in milliseconds.
          log: An optional string listing any errors.
          failure_reason: An optional string listing why the test failed.
          test_file: An optional string listing the file location of the test.
        """
        assert name
        assert test_type in ResultType.GetTypes()
        self._name = name
        self._test_type = test_type
        self._duration = duration
        self._log = log
        self._failure_reason = failure_reason
        self._test_file = test_file
        self._links = {}
        self._webview_multiprocess_mode = MULTIPROCESS_SUFFIX in name

    def __str__(self):
        return self._name

    def __repr__(self):
        return self._name

    def __eq__(self, other):
        return self.GetName() == other.GetName()

    def __lt__(self, other):
        return self.GetName() < other.GetName()

    def __hash__(self):
        return hash(self._name)

    def SetName(self, name):
        """Set the test name.

        Because we're putting this into a set, this should only be used if moving
        this test result into another set.
        """
        self._name = name

    def GetName(self):
        """Get the test name."""
        return self._name

    def GetNameForResultSink(self):
        """Get the test name to be reported to resultsink."""
        raw_name = self.GetName()

        # The name can include suffixes encoding Webview variant data:
        # a Webview multiprocess mode suffix and an AwSettings mutation suffix.
        # If both are present, the mutation suffix will come after the multiprocess
        # suffix. The mutation suffix can either be "__null_" or "__{key}..{value}_"
        #
        # Examples:
        # (...)AwSettingsTest#testAssetUrl__multiprocess_mode__allMutations..true_
        # (...)AwSettingsTest#testAssetUrl__multiprocess_mode__null_
        # (...)AwSettingsTest#testAssetUrl__allMutations..true_
        # org.chromium.android_webview.test.AwSettingsTest#testAssetUrl__null_

        # first, strip any AwSettings mutation parameter information
        # from the RHS of the raw_name
        if raw_name.endswith(_NULL_MUTATION_SUFFIX):
            raw_name = raw_name[: -len(_NULL_MUTATION_SUFFIX)]
        elif match := _MUTATION_SUFFIX_PATTERN.search(raw_name):
            raw_name = match.group(1)

        # At this stage, the name will only have the multiprocess suffix appended,
        # if applicable.
        #
        # Examples:
        # (...)AwSettingsTest#testAssetUrl__multiprocess_mode
        # org.chromium.android_webview.test.AwSettingsTest#testAssetUrl

        # then check for multiprocess mode suffix and strip it, if present
        if self._webview_multiprocess_mode:
            assert raw_name.endswith(MULTIPROCESS_SUFFIX), (
                'multiprocess mode test raw name should have the corresponding suffix'
            )
            return raw_name[: -len(MULTIPROCESS_SUFFIX)]
        return raw_name

    def SetType(self, test_type):
        """Set the test result type."""
        assert test_type in ResultType.GetTypes()
        self._test_type = test_type

    def GetType(self):
        """Get the test result type."""
        return self._test_type

    def GetDuration(self):
        """Get the test duration."""
        return self._duration

    def SetTestFile(self, test_file):
        """Set the test file location."""
        self._test_file = test_file

    def GetTestFile(self):
        """Get the test file location."""
        return self._test_file

    def SetLog(self, log):
        """Set the test log."""
        self._log = log

    def GetLog(self):
        """Get the test log."""
        return self._log

    def SetFailureReason(self, failure_reason):
        """Set the reason the test failed.

        This should be the first failure the test encounters and exclude any stack
        trace.
        """
        self._failure_reason = failure_reason

    def GetFailureReason(self):
        """Get the reason the test failed.

        Returns None if the test did not fail or if the reason the test failed is
        unknown.
        """
        return self._failure_reason

    def SetLink(self, name, link_url):
        """Set link with test result data."""
        self._links[name] = link_url

    def GetLinks(self):
        """Get dict containing links to test result data."""
        return self._links

    def GetVariantForResultSink(self):
        """Get the variant dict to be reported to result sink."""
        variants = {}
        if match := _MUTATION_SUFFIX_PATTERN.search(self.GetName()):
            # variant keys need to be lowercase
            variants[match.group(2).lower()] = match.group(3)
        if self._webview_multiprocess_mode:
            variants['webview_multiprocess_mode'] = 'Yes'
        return variants or None


class TestRunResults:
    """Set of results for a test run."""

    def __init__(self):
        self._links = {}
        self._results = set()
        self._results_lock = threading.RLock()

    def SetLink(self, name, link_url):
        """Add link with test run results data."""
        self._links[name] = link_url

    def GetLinks(self):
        """Get dict containing links to test run result data."""
        return self._links

    def GetLogs(self):
        """Get the string representation of all test logs."""
        with self._results_lock:
            s = []
            for test_type in ResultType.GetTypes():
                if test_type != ResultType.PASS:
                    for t in sorted(self._GetType(test_type)):
                        log = t.GetLog()
                        if log:
                            s.append('[%s] %s:' % (test_type, t))
                            s.append(log)
            if sys.version_info.major == 2:
                decoded = [
                    u.decode(encoding='utf-8', errors='ignore') for u in s
                ]
                return '\n'.join(decoded)
            return '\n'.join(s)

    def GetGtestForm(self):
        """Get the gtest string representation of this object."""
        with self._results_lock:
            s = []
            plural = lambda n, s, p: '%d %s' % (n, p if n != 1 else s)
            tests = lambda n: plural(n, 'test', 'tests')

            s.append('[==========] %s ran.' % (tests(len(self.GetAll()))))
            s.append('[  PASSED  ] %s.' % (tests(len(self.GetPass()))))

            skipped = self.GetSkip()
            if skipped:
                s.append(
                    '[  SKIPPED ] Skipped %s, listed below:'
                    % tests(len(skipped))
                )
                for t in sorted(skipped):
                    s.append('[  SKIPPED ] %s' % str(t))

            all_failures = self.GetFail().union(
                self.GetCrash(), self.GetTimeout(), self.GetUnknown()
            )
            if all_failures:
                s.append(
                    '[  FAILED  ] %s, listed below:' % tests(len(all_failures))
                )
                for t in sorted(self.GetFail()):
                    s.append('[  FAILED  ] %s' % str(t))
                for t in sorted(self.GetCrash()):
                    s.append('[  FAILED  ] %s (CRASHED)' % str(t))
                for t in sorted(self.GetTimeout()):
                    s.append('[  FAILED  ] %s (TIMEOUT)' % str(t))
                for t in sorted(self.GetUnknown()):
                    s.append('[  FAILED  ] %s (UNKNOWN)' % str(t))
                s.append('')
                s.append(
                    plural(len(all_failures), 'FAILED TEST', 'FAILED TESTS')
                )
            return '\n'.join(s)

    def GetShortForm(self):
        """Get the short string representation of this object."""
        with self._results_lock:
            s = []
            s.append('ALL: %d' % len(self._results))
            for test_type in ResultType.GetTypes():
                s.append('%s: %d' % (test_type, len(self._GetType(test_type))))
            return ''.join([x.ljust(15) for x in s])

    def __str__(self):
        return self.GetGtestForm()

    def AddResult(self, result):
        """Add |result| to the set.

        Args:
          result: An instance of BaseTestResult.
        """
        assert isinstance(result, BaseTestResult)
        with self._results_lock:
            self._results.discard(result)
            self._results.add(result)

    def AddResults(self, results):
        """Add |results| to the set.

        Args:
          results: An iterable of BaseTestResult objects.
        """
        with self._results_lock:
            for t in results:
                self.AddResult(t)

    def AddTestRunResults(self, results):
        """Add the set of test results from |results|.

        Args:
          results: An instance of TestRunResults.
        """
        assert isinstance(results, TestRunResults), (
            'Expected TestRunResult object: %s' % type(results)
        )
        with self._results_lock:
            # pylint: disable=W0212
            self._results.update(results._results)

    def GetAll(self):
        """Get the set of all test results."""
        with self._results_lock:
            return self._results.copy()

    def _GetType(self, test_type):
        """Get the set of test results with the given test type."""
        with self._results_lock:
            return set(t for t in self._results if t.GetType() == test_type)

    def GetPass(self):
        """Get the set of all passed test results."""
        return self._GetType(ResultType.PASS)

    def GetSkip(self):
        """Get the set of all skipped test results."""
        return self._GetType(ResultType.SKIP)

    def GetFail(self):
        """Get the set of all failed test results."""
        return self._GetType(ResultType.FAIL)

    def GetCrash(self):
        """Get the set of all crashed test results."""
        return self._GetType(ResultType.CRASH)

    def GetTimeout(self):
        """Get the set of all timed out test results."""
        return self._GetType(ResultType.TIMEOUT)

    def GetUnknown(self):
        """Get the set of all unknown test results."""
        return self._GetType(ResultType.UNKNOWN)

    def GetNotPass(self):
        """Get the set of all non-passed test results."""
        return self.GetAll() - self.GetPass()

    def DidRunPass(self):
        """Return whether the test run was successful."""
        return not self.GetNotPass() - self.GetSkip()