xref: /aosp_15_r20/tools/asuite/atest/atest_integration_tests.py (revision c2e18aaa1096c836b086f94603d04f4eb9cf37f5) 
1#!/usr/bin/env python3
2#
3# Copyright 2018, 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
17"""ATest Integration Test Class.
18
19The purpose is to prevent potential side-effects from breaking ATest at the
20early stage while landing CLs with potential side-effects.
21
22It forks a subprocess with ATest commands to validate if it can pass all the
23finding, running logic of the python code, and waiting for TF to exit properly.
24    - When running with ROBOLECTRIC tests, it runs without TF, and will exit
25    the subprocess with the message "All tests passed"
26    - If FAIL, it means something breaks ATest unexpectedly!
27"""
28
29from __future__ import print_function
30
31import os
32import subprocess
33import sys
34import tempfile
35import time
36import unittest
37
38
39_TEST_RUN_DIR_PREFIX = 'atest_integration_tests_%s_'
40_LOG_FILE = 'integration_tests.log'
41_FAILED_LINE_LIMIT = 50
42_EXIT_TEST_FAILED = 1
43_ALTERNATIVES = {'-dev'}
44_INTEGRATION_TESTS = [
45    os.path.join(
46        os.environ.get('ANDROID_BUILD_TOP', os.getcwd()),
47        'tools/asuite/atest/test_plans/INTEGRATION_TESTS',
48    )
49]
50
51
52class ATestIntegrationTest(unittest.TestCase):
53  """ATest Integration Test Class."""
54
55  NAME = 'ATestIntegrationTest'
56  EXECUTABLE = 'atest'
57  OPTIONS = ''
58  _RUN_CMD = '{exe} {options} {test}'
59  _PASSED_CRITERIA = ['will be rescheduled', 'All tests passed']
60
61  def setUp(self):
62    """Set up stuff for testing."""
63    self.full_env_vars = os.environ.copy()
64    self.test_passed = False
65    self.log = []
66
67  def run_test(self, testcase):
68    """Create a subprocess to execute the test command.
69
70    Strategy:
71        Fork a subprocess to wait for TF exit properly, and log the error
72        if the exit code isn't 0.
73
74    Args:
75        testcase: A string of testcase name.
76    """
77    run_cmd_dict = {
78        'exe': self.EXECUTABLE,
79        'options': self.OPTIONS,
80        'test': testcase,
81    }
82    run_command = self._RUN_CMD.format(**run_cmd_dict)
83    try:
84      subprocess.check_output(
85          run_command,
86          stderr=subprocess.PIPE,
87          env=self.full_env_vars,
88          shell=True,
89      )
90    except subprocess.CalledProcessError as e:
91      self.log.append(e.output.decode())
92      return False
93    return True
94
95  def get_failed_log(self):
96    """Get a trimmed failed log.
97
98    Strategy:
99        In order not to show the unnecessary log such as build log,
100        it's better to get a trimmed failed log that contains the
101        most important information.
102
103    Returns:
104        A trimmed failed log.
105    """
106    failed_log = '\n'.join(filter(None, self.log[-_FAILED_LINE_LIMIT:]))
107    return failed_log
108
109
110def create_test_method(testcase, log_path):
111  """Create a test method according to the testcase.
112
113  Args:
114      testcase: A testcase name.
115      log_path: A file path for storing the test result.
116
117  Returns:
118      A created test method, and a test function name.
119  """
120  test_function_name = 'test_%s' % testcase.replace(' ', '_')
121
122  # pylint: disable=missing-docstring
123  def template_test_method(self):
124    self.test_passed = self.run_test(testcase)
125    open(log_path, 'a').write('\n'.join(self.log))
126    failed_message = 'Running command: %s failed.\n' % testcase
127    failed_message += '' if self.test_passed else self.get_failed_log()
128    self.assertTrue(self.test_passed, failed_message)
129
130  return test_function_name, template_test_method
131
132
133def create_test_run_dir():
134  """Create the test run directory in tmp.
135
136  Returns:
137      A string of the directory path.
138  """
139  utc_epoch_time = int(time.time())
140  prefix = _TEST_RUN_DIR_PREFIX % utc_epoch_time
141  return tempfile.mkdtemp(prefix=prefix)
142
143
144if __name__ == '__main__':
145  # TODO(b/129029189) Implement detail comparison check for dry-run mode.
146  ARGS = sys.argv[1:]
147  if ARGS:
148    for exe in _ALTERNATIVES:
149      if exe in ARGS:
150        ARGS.remove(exe)
151        ATestIntegrationTest.EXECUTABLE += exe
152    ATestIntegrationTest.OPTIONS = ' '.join(ARGS)
153  print('Running tests with {}\n'.format(ATestIntegrationTest.EXECUTABLE))
154  try:
155    LOG_PATH = os.path.join(create_test_run_dir(), _LOG_FILE)
156    for TEST_PLANS in _INTEGRATION_TESTS:
157      with open(TEST_PLANS) as test_plans:
158        for test in test_plans:
159          # Skip test when the line startswith #.
160          if not test.strip() or test.strip().startswith('#'):
161            continue
162          test_func_name, test_func = create_test_method(test.strip(), LOG_PATH)
163          setattr(ATestIntegrationTest, test_func_name, test_func)
164    SUITE = unittest.TestLoader().loadTestsFromTestCase(ATestIntegrationTest)
165    RESULTS = unittest.TextTestRunner(verbosity=2).run(SUITE)
166  finally:
167    if RESULTS.failures:
168      print('Full test log is saved to %s' % LOG_PATH)
169      sys.exit(_EXIT_TEST_FAILED)
170    else:
171      os.remove(LOG_PATH)
172