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