1# Copyright 2024, The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15"""Test runner invocation class.""" 16 17from __future__ import annotations 18 19import time 20import traceback 21from typing import Any, Dict, List, Set 22 23from atest import result_reporter 24from atest.atest_enum import ExitCode 25from atest.metrics import metrics 26from atest.metrics import metrics_utils 27from atest.test_finders import test_info 28from atest.test_runners import test_runner_base 29 30 31class TestRunnerInvocation: 32 """An invocation executing tests based on given arguments.""" 33 34 def __init__( 35 self, 36 *, 37 test_runner: test_runner_base.TestRunnerBase, 38 extra_args: Dict[str, Any], 39 test_infos: List[test_info.TestInfo], 40 ): 41 self._extra_args = extra_args 42 self._test_infos = test_infos 43 self._test_runner = test_runner 44 45 @property 46 def test_infos(self): 47 return self._test_infos 48 49 def __eq__(self, other): 50 return self.__dict__ == other.__dict__ 51 52 def requires_device_update(self): 53 """Checks whether this invocation requires device update.""" 54 return self._test_runner.requires_device_update(self._test_infos) 55 56 def get_test_runner_reqs(self) -> Set[str]: 57 """Returns the required build targets for this test runner invocation.""" 58 return self._test_runner.get_test_runner_build_reqs(self._test_infos) 59 60 # pylint: disable=too-many-locals 61 def run_all_tests(self, reporter: result_reporter.ResultReporter) -> ExitCode: 62 """Runs all tests.""" 63 64 test_start = time.time() 65 is_success = True 66 err_msg = None 67 try: 68 tests_ret_code = self._test_runner.run_tests( 69 self._test_infos, self._extra_args, reporter 70 ) 71 except Exception: # pylint: disable=broad-except 72 is_success = False 73 err_msg = traceback.format_exc() 74 75 if not is_success: 76 reporter.runner_failure(self._test_runner.NAME, err_msg) 77 tests_ret_code = ExitCode.TEST_FAILURE 78 79 run_time = metrics_utils.convert_duration(time.time() - test_start) 80 tests = [] 81 for test in reporter.get_test_results_by_runner(self._test_runner.NAME): 82 # group_name is module name with abi(for example, 83 # 'x86_64 CtsSampleDeviceTestCases'). 84 # Filtering abi in group_name. 85 test_group = test.group_name 86 # Withdraw module name only when the test result has reported. 87 module_name = test_group 88 if test_group and ' ' in test_group: 89 _, module_name = test_group.split() 90 testcase_name = '%s:%s' % (module_name, test.test_name) 91 result = test_runner_base.RESULT_CODE[test.status] 92 tests.append( 93 {'name': testcase_name, 'result': result, 'stacktrace': test.details} 94 ) 95 metrics.RunnerFinishEvent( 96 duration=run_time, 97 success=is_success, 98 runner_name=self._test_runner.NAME, 99 test=tests, 100 ) 101 102 return tests_ret_code 103