#!/usr/bin/env python3
#
# Copyright 2017 - The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import logging

from acts.libs.testtracker.testtracker_results_writer import KEY_EFFORT_NAME
from acts.libs.testtracker.testtracker_results_writer import TestTrackerError
from acts.libs.testtracker.testtracker_results_writer import TestTrackerResultsWriter
from mobly.base_test import BaseTestClass

from acts import signals


def test_info(predicate=None, **keyed_info):
    """Adds info about test.

    Extra info to include about the test. This info will be available in the
    test output. Note that if a key is given multiple times it will be added
    as a list of all values. If multiples of these are stacked their results
    will be merged.

    Example:
        # This test will have a variable my_var
        @test_info(my_var='THIS IS MY TEST')
        def my_test(self):
            return False

    Args:
        predicate: A func to call that if false will skip adding this test
                   info. Function signature is bool(test_obj, args, kwargs)
        **keyed_info: The key, value info to include in the extras for this
                      test.
    """

    def test_info_decorator(func):
        return _TestInfoDecoratorFunc(func, predicate, keyed_info)

    return test_info_decorator


def __select_last(test_signals, _):
    return test_signals[-1]


def repeated_test(num_passes, acceptable_failures=0,
                  result_selector=__select_last):
    """A decorator that runs a test case multiple times.

    This decorator can be used to run a test multiple times and aggregate the
    data into a single test result. By setting `result_selector`, the user can
    access the returned result of each run, allowing them to average results,
    return the median, or gather and return standard deviation values.

    This decorator should be used on test cases, and should not be used on
    static or class methods. The test case must take in an additional argument,
    `attempt_number`, which returns the current attempt number, starting from
    1.

    Note that any TestSignal intended to abort or skip the test will take
    abort or skip immediately.

    Args:
        num_passes: The number of times the test needs to pass to report the
            test case as passing.
        acceptable_failures: The number of failures accepted. If the failures
            exceeds this number, the test will stop repeating. The maximum
            number of runs is `num_passes + acceptable_failures`. If the test
            does fail, result_selector will still be called.
        result_selector: A lambda that takes in the list of TestSignals and
            returns the test signal to report the test case as. Note that the
            list also contains any uncaught exceptions from the test execution.
    """
    def decorator(func):
        if not func.__name__.startswith('test_'):
            raise ValueError('Tests must start with "test_".')

        def test_wrapper(self):
            num_failures = 0
            num_seen_passes = 0
            test_signals_received = []
            for i in range(num_passes + acceptable_failures):
                try:
                    func(self, i + 1)
                except (signals.TestFailure, signals.TestError,
                        AssertionError) as signal:
                    test_signals_received.append(signal)
                    num_failures += 1
                except signals.TestPass as signal:
                    test_signals_received.append(signal)
                    num_seen_passes += 1
                except (signals.TestSignal, KeyboardInterrupt):
                    raise
                except Exception as signal:
                    test_signals_received.append(signal)
                    num_failures += 1
                else:
                    num_seen_passes += 1
                    test_signals_received.append(signals.TestPass(
                        'Test iteration %s of %s passed without details.' % (
                        i, func.__name__)))

                if num_failures > acceptable_failures:
                    break
                elif num_seen_passes == num_passes:
                    break
                else:
                    self.teardown_test()
                    self.setup_test()

            raise result_selector(test_signals_received, self)

        return test_wrapper

    return decorator


def test_tracker_info(uuid, extra_environment_info=None, predicate=None):
    """Decorator for adding test tracker info to tests results.

    Will add test tracker info inside of Extras/test_tracker_info.

    Example:
        # This test will be linked to test tracker uuid abcd
        @test_tracker_info(uuid='abcd')
        def my_test(self):
            return False

    Args:
        uuid: The uuid of the test case in test tracker.
        extra_environment_info: Extra info about the test tracker environment.
        predicate: A func that if false when called will ignore this info.
    """

    def test_tracker_info_decorator(func):
        keyed_info = dict(test_tracker_uuid=uuid,
                          test_tracker_environment_info=extra_environment_info)
        return _TestTrackerInfoDecoratorFunc(func, predicate, keyed_info)

    return test_tracker_info_decorator


class _TestInfoDecoratorFunc(object):
    """Object that acts as a function decorator test info."""

    def __init__(self, func, predicate, keyed_info):
        self.func = func
        self.predicate = predicate
        self.keyed_info = keyed_info
        self.__name__ = func.__name__
        self.__doc__ = func.__doc__
        self.__module__ = func.__module__

    def __get__(self, instance, owner):
        """Called by Python to create a binding for an instance closure.

        When called by Python this object will create a special binding for
        that instance. That binding will know how to interact with this
        specific decorator.
        """
        return _TestInfoBinding(self, instance)

    def __call__(self, *args, **kwargs):
        """
        When called runs the underlying func and then attaches test info
        to a signal.
        """
        new_signal = self._get_signal_from_func_call(*args, **kwargs)
        raise new_signal

    def _get_signal_from_func_call(self, *args, **kwargs):
        """Calls the underlying func, then attaches test info to the resulting
        signal and raises the signal.
        """
        cause = None
        try:
            result = self.func(*args, **kwargs)

            if result or result is None:
                new_signal = signals.TestPass('')
            else:
                new_signal = signals.TestFailure('')
        except signals.TestSignal as signal:
            new_signal = signal
        except Exception as ex:
            cause = ex
            new_signal = signals.TestError(cause)

        if new_signal.extras is None:
            new_signal.extras = {}
        if not isinstance(new_signal.extras, dict):
            raise ValueError('test_info can only append to signal data '
                             'that has a dict as the extra value.')

        gathered_extras = self._gather_local_info(None, *args, **kwargs)
        for k, v in gathered_extras.items():
            if k not in new_signal.extras:
                new_signal.extras[k] = v
            else:
                if not isinstance(new_signal.extras[k], list):
                    new_signal.extras[k] = [new_signal.extras[k]]

                new_signal.extras[k].insert(0, v)

        raise new_signal from cause

    def gather(self, *args, **kwargs):
        """
        Gathers the info from this decorator without invoking the underlying
        function. This will also gather all child info if the underlying func
        has that ability.

        Returns: A dictionary of info.
        """
        if hasattr(self.func, 'gather'):
            extras = self.func.gather(*args, **kwargs)
        else:
            extras = {}

        self._gather_local_info(extras, *args, **kwargs)

        return extras

    def _gather_local_info(self, gather_into, *args, **kwargs):
        """Gathers info from this decorator and ignores children.

        Args:
            gather_into: Gathers into a dictionary that already exists.

        Returns: The dictionary with gathered info in it.
        """
        if gather_into is None:
            extras = {}
        else:
            extras = gather_into
        if not self.predicate or self.predicate(args, kwargs):
            for k, v in self.keyed_info.items():
                if v and k not in extras:
                    extras[k] = v
                elif v and k in extras:
                    if not isinstance(extras[k], list):
                        extras[k] = [extras[k]]
                    extras[k].insert(0, v)

        return extras


class _TestTrackerInfoDecoratorFunc(_TestInfoDecoratorFunc):
    """
    Expands on _TestInfoDecoratorFunc by writing gathered test info to a
    TestTracker proto file
    """

    def __call__(self, *args, **kwargs):
        """
        When called runs the underlying func and then attaches test info
        to a signal. It then writes the result from the signal to a TestTracker
        Result proto file.
        """
        try:
            self._get_signal_from_func_call(*args, **kwargs)
        except signals.TestSignal as new_signal:
            if not args or not isinstance(args[0], BaseTestClass):
                logging.warning('The decorated object must be an instance of'
                                'an ACTS/Mobly test class.')
            else:
                self._write_to_testtracker(args[0], new_signal)
            raise new_signal

    def _write_to_testtracker(self, test_instance, signal):
        """Write test result from given signal to a TestTracker Result proto
        file.

        Due to infra contraints on nested structures in userparams, this
        expects the test_instance to have user_params defined as follows:

            testtracker_properties: A comma-delimited list of
                'prop_name=<userparam_name>'
            <userparam_name>: testtracker property value.
        """
        tt_prop_to_param_names = test_instance.user_params.get(
            'testtracker_properties')

        if not tt_prop_to_param_names:
            return

        tt_prop_to_param_names = tt_prop_to_param_names.split(',')

        testtracker_properties = {}
        for entry in tt_prop_to_param_names:
            prop_name, param_name = entry.split('=')
            if param_name in test_instance.user_params:
                testtracker_properties[prop_name] = (
                    test_instance.user_params[param_name])

        if (hasattr(test_instance, 'android_devices') and
                KEY_EFFORT_NAME not in testtracker_properties):
            testtracker_properties[KEY_EFFORT_NAME] = (
                test_instance.android_devices[0].build_info['build_id'])

        try:
            writer = TestTrackerResultsWriter(
                test_instance.log_path, testtracker_properties)
            writer.write_results_from_test_signal(
                signal, test_instance.begin_time)
        except TestTrackerError:
            test_instance.log.exception('TestTracker Error')


class _TestInfoBinding(object):
    """
    When Python creates an instance of an object it creates a binding object
    for each closure that contains what the instance variable should be when
    called. This object is a similar binding for _TestInfoDecoratorFunc.
    When Python tries to create a binding of a _TestInfoDecoratorFunc it
    will return one of these objects to hold the instance for that closure.
    """

    def __init__(self, target, instance):
        """
        Args:
            target: The target for creating a binding to.
            instance: The instance to bind the target with.
        """
        self.target = target
        self.instance = instance
        self.__name__ = target.__name__

    def __call__(self, *args, **kwargs):
        """
        When this object is called it will call the target with the bound
        instance.
        """
        return self.target(self.instance, *args, **kwargs)

    def gather(self, *args, **kwargs):
        """
        Will gather the target with the bound instance.
        """
        return self.target.gather(self.instance, *args, **kwargs)