1#!/usr/bin/env python3
2#
3#   Copyright 2020 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import statistics
18
19from acts.base_test import BaseTestClass
20from acts.test_decorators import repeated_test
21from acts import signals
22
23
24def get_median_current(test_results):
25    """Returns the median current, or a failure if the test failed."""
26    # If the last run was not a pass signal, the test exceeded
27    # acceptable_failures.
28    if not isinstance(test_results[-1], signals.TestPass):
29        return test_results[-1]
30
31    # Only look at results within the good range (i.e., passing results).
32    valid_results = filter(lambda result: isinstance(result, signals.TestPass),
33                           test_results)
34
35    # Gather the current measurements and return the median.
36    median_current = statistics.median(
37        map(lambda result: result.extras['current'], valid_results))
38    return signals.TestPass('Pass msg! Current: %s' % median_current,
39                            extras={'current': median_current})
40
41
42class RepeatedTest(BaseTestClass):
43    def __init__(self, controllers):
44        super().__init__(controllers)
45        self.count_for_passing_testcase = 0
46        self.count_for_failing_testcase = 0
47
48    @repeated_test(num_passes=3, acceptable_failures=0)
49    def test_repeated_case(self):
50        self.log.info('This logic executes three times.')
51
52    @repeated_test(num_passes=3, acceptable_failures=2,
53                   result_selector=get_median_current)
54    def test_repeated_case_pass(self):
55        """The end result of this test is a pass with current=3.5"""
56        returned_results = [
57            signals.TestPass('0Pass msg!', extras={'current': 3.5}),
58            signals.TestFailure('Fail msg!', extras={'current': 100.0}),
59            signals.TestPass('1Pass msg!', extras={'current': 3.2}),
60            signals.TestPass('2Pass msg!', extras={'current': 3.6})
61        ]
62        # Every time this function runs, we return a different signal.
63        self.count_for_passing_testcase += 1
64        raise returned_results[self.count_for_passing_testcase - 1]
65
66    @repeated_test(num_passes=3, acceptable_failures=2,
67                   result_selector=get_median_current)
68    def test_repeated_case_with_failures(self):
69        """The end result of this test is the last failure to occur."""
70        returned_results = [
71            signals.TestPass('Pass msg!', extras={'current': 3.5}),
72            signals.TestFailure('Fail msg!', extras={'current': 100.0}),
73            signals.TestFailure('Fail msg!', extras={'current': 58.1}),
74            signals.TestFailure('Fail msg!', extras={'current': 74.2}),
75        ]
76        # Every time this function runs, we return a different signal.
77        self.count_for_failing_testcase += 1
78        raise returned_results[(self.count_for_failing_testcase - 1) % 4]
79