1#!/usr/bin/env python3
2#
3#   Copyright 2017 - 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
17import os
18import sys
19import tempfile
20import unittest
21import multiprocessing
22
23
24def run_tests(test_suite, output_file):
25    # Redirects stdout and stderr to the given output file.
26    new_stdout = open(output_file, 'w+')
27    os.dup2(new_stdout.fileno(), 1)
28    test_run = unittest.TextTestRunner(stream=new_stdout, verbosity=2).run(test_suite)
29    return test_run.wasSuccessful()
30
31
32class TestResult(object):
33    def __init__(self, process_result, output_file, test_suite):
34        self.process_result = process_result
35        self.output_file = output_file
36        self.test_suite = test_suite
37
38
39def run_all_unit_tests():
40    # Due to some incredibly powerful black magic, running this twice
41    # causes the metrics/, test_utils/ and test_runner_test.py tests to load
42    # properly. They do no load properly the first time.
43    suite = unittest.TestLoader().discover(
44        start_dir=os.path.dirname(__file__), pattern='*_test.py')
45    suite = unittest.TestLoader().discover(
46        start_dir=os.path.dirname(__file__), pattern='*_test.py')
47
48    process_pool = multiprocessing.Pool(10)
49    output_dir = tempfile.mkdtemp()
50
51    results = []
52
53    for index, test in enumerate(suite._tests):
54        output_file = os.path.join(output_dir, 'test_%s.output' % index)
55        process_result = process_pool.apply_async(run_tests,
56                                                  args=(test, output_file))
57        results.append(TestResult(process_result, output_file, test))
58
59    success = True
60    for index, result in enumerate(results):
61        try:
62            if not result.process_result.get(timeout=60):
63                success = False
64                print('Received the following test failure:')
65                with open(result.output_file, 'r') as out_file:
66                    print(out_file.read(), file=sys.stderr)
67        except multiprocessing.TimeoutError:
68            success = False
69            print('The following test timed out: %r' % result.test_suite,
70                  file=sys.stderr)
71            with open(result.output_file, 'r') as out_file:
72                print(out_file.read())
73
74    exit(not success)
75
76
77if __name__ == '__main__':
78    run_all_unit_tests()
79