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