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 os
20import time
21import traceback
22from typing import Any, Dict, List, Set
23
24from atest import result_reporter
25from atest.atest_enum import ExitCode
26from atest.metrics import metrics
27from atest.metrics import metrics_utils
28from atest.test_finders import test_info
29from atest.test_runners import test_runner_base
30from atest.test_runners.event_handler import EventHandleError
31
32# Look for this in tradefed log messages.
33TRADEFED_EARLY_EXIT_LOG_SIGNAL = (
34    'INSTRUMENTATION_RESULT: shortMsg=Process crashed'
35)
36
37# Print this to user.
38TRADEFED_EARLY_EXIT_ATEST_MSG = (
39    'Test failed because instrumentation process died.'
40    ' Please check your device logs.'
41)
42
43
44class TestRunnerInvocation:
45  """An invocation executing tests based on given arguments."""
46
47  def __init__(
48      self,
49      *,
50      test_runner: test_runner_base.TestRunnerBase,
51      extra_args: Dict[str, Any],
52      test_infos: List[test_info.TestInfo],
53  ):
54    self._extra_args = extra_args
55    self._test_infos = test_infos
56    self._test_runner = test_runner
57
58  @property
59  def test_infos(self):
60    return self._test_infos
61
62  def __eq__(self, other):
63    return self.__dict__ == other.__dict__
64
65  def requires_device_update(self):
66    """Checks whether this invocation requires device update."""
67    return self._test_runner.requires_device_update(self._test_infos)
68
69  def get_test_runner_reqs(self) -> Set[str]:
70    """Returns the required build targets for this test runner invocation."""
71    return self._test_runner.get_test_runner_build_reqs(self._test_infos)
72
73  # pylint: disable=too-many-locals
74  def run_all_tests(self, reporter: result_reporter.ResultReporter) -> ExitCode:
75    """Runs all tests."""
76
77    test_start = time.time()
78    is_success = True
79    err_msg = None
80    try:
81      tests_ret_code = self._test_runner.run_tests(
82          self._test_infos, self._extra_args, reporter
83      )
84    except EventHandleError:
85      is_success = False
86      if self.log_shows_early_exit():
87        err_msg = TRADEFED_EARLY_EXIT_ATEST_MSG
88      else:
89        err_msg = traceback.format_exc()
90
91    except Exception:  # pylint: disable=broad-except
92      is_success = False
93      err_msg = traceback.format_exc()
94
95    if not is_success:
96      reporter.runner_failure(self._test_runner.NAME, err_msg)
97      tests_ret_code = ExitCode.TEST_FAILURE
98
99    run_time = metrics_utils.convert_duration(time.time() - test_start)
100    tests = []
101    for test in reporter.get_test_results_by_runner(self._test_runner.NAME):
102      # group_name is module name with abi(for example,
103      # 'x86_64 CtsSampleDeviceTestCases').
104      # Filtering abi in group_name.
105      test_group = test.group_name
106      # Withdraw module name only when the test result has reported.
107      module_name = test_group
108      if test_group and ' ' in test_group:
109        _, module_name = test_group.split()
110      testcase_name = '%s:%s' % (module_name, test.test_name)
111      result = test_runner_base.RESULT_CODE[test.status]
112      tests.append(
113          {'name': testcase_name, 'result': result, 'stacktrace': test.details}
114      )
115    metrics.RunnerFinishEvent(
116        duration=run_time,
117        success=is_success,
118        runner_name=self._test_runner.NAME,
119        test=tests,
120    )
121
122    return tests_ret_code
123
124  def log_shows_early_exit(self) -> bool:
125    """Grep the log file for TF process crashed message."""
126    # Ensure file exists and is readable.
127    if not os.access(self._test_runner.test_log_file.name, os.R_OK):
128      return False
129
130    with open(self._test_runner.test_log_file.name, 'r') as log_file:
131      for line in log_file:
132        if TRADEFED_EARLY_EXIT_LOG_SIGNAL in line:
133          return True
134
135    return False
136