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