4 A smart, fast test runner.
13 from abc import ABC, abstractmethod
14 from dataclasses import dataclass
15 from typing import Any, Dict, List, Optional
17 from overrides import overrides
23 import parallelize as par
27 logger = logging.getLogger(__name__)
28 args = config.add_commandline_args(f'({__file__})', 'Args related to __file__')
29 args.add_argument('--unittests', '-u', action='store_true', help='Run unittests.')
30 args.add_argument('--doctests', '-d', action='store_true', help='Run doctests.')
31 args.add_argument('--integration', '-i', action='store_true', help='Run integration tests.')
33 '--coverage', '-c', action='store_true', help='Run tests and capture code coverage data'
36 HOME = os.environ['HOME']
40 class TestingParameters:
42 halt_event: threading.Event
48 tests_executed: List[str]
49 tests_succeeded: List[str]
50 tests_failed: List[str]
51 tests_timed_out: List[str]
54 class TestRunner(ABC, thread_utils.ThreadWithReturnValue):
55 def __init__(self, params: TestingParameters):
56 super().__init__(self, target=self.begin, args=[params])
58 self.test_results = TestResults(
59 name=f"All {self.get_name()} tests",
66 def aggregate_test_results(self, result: TestResults):
67 self.test_results.tests_executed.extend(result.tests_executed)
68 self.test_results.tests_succeeded.extend(result.tests_succeeded)
69 self.test_results.tests_failed.extend(result.tests_failed)
70 self.test_results.tests_timed_out.extend(result.tests_timed_out)
73 def get_name(self) -> str:
77 def begin(self, params: TestingParameters) -> TestResults:
81 class TemplatedTestRunner(TestRunner, ABC):
83 def identify_tests(self) -> List[Any]:
87 def run_test(self, test: Any) -> TestResults:
90 def check_for_abort(self):
91 if self.params.halt_event.is_set():
92 logger.debug('Thread %s saw halt event; exiting.', self.get_name())
93 raise Exception("Kill myself!")
94 if self.params.halt_on_error:
95 if len(self.test_results.tests_failed) > 0:
96 logger.error('Thread %s saw abnormal results; exiting.', self.get_name())
97 raise Exception("Kill myself!")
99 def status_report(self, running: List[Any], done: List[Any]):
100 total = len(running) + len(done)
102 '%s: %d/%d in flight; %d/%d completed.',
110 def persist_output(self, test_name: str, message: str, output: str) -> None:
111 basename = file_utils.without_path(test_name)
112 dest = f'{basename}-output.txt'
113 with open(f'./test_output/{dest}', 'w') as wf:
114 print(message, file=wf)
115 print('-' * len(message), file=wf)
118 def execute_commandline(
123 timeout: float = 120.0,
127 logger.debug('%s: Running %s (%s)', self.get_name(), test_name, cmdline)
128 output = exec_utils.cmd(
130 timeout_seconds=timeout,
132 self.persist_output(test_name, f'{test_name} ({cmdline}) succeeded.', output)
133 logger.debug('%s (%s) succeeded', test_name, cmdline)
134 return TestResults(test_name, [test_name], [test_name], [], [])
135 except subprocess.TimeoutExpired as e:
136 msg = f'{self.get_name()}: {test_name} ({cmdline}) timed out after {e.timeout:.1f} seconds.'
139 '%s: %s output when it timed out: %s', self.get_name(), test_name, e.output
141 self.persist_output(test_name, msg, e.output)
149 except subprocess.CalledProcessError as e:
150 msg = f'{self.get_name()}: {test_name} ({cmdline}) failed; exit code {e.returncode}'
152 logger.debug('%s: %s output when it failed: %s', self.get_name(), test_name, e.output)
153 self.persist_output(test_name, msg, e.output)
163 def begin(self, params: TestingParameters) -> TestResults:
164 logger.debug('Thread %s started.', self.get_name())
165 interesting_tests = self.identify_tests()
166 running: List[Any] = []
168 for test in interesting_tests:
169 running.append(self.run_test(test))
171 while len(running) > 0:
172 self.status_report(running, done)
173 self.check_for_abort()
177 newly_finished.append(fut)
178 result = fut._resolve()
179 logger.debug('Test %s finished.', result.name)
180 self.aggregate_test_results(result)
182 for fut in newly_finished:
187 logger.debug('Thread %s finished.', self.get_name())
188 return self.test_results
191 class UnittestTestRunner(TemplatedTestRunner):
193 def get_name(self) -> str:
194 return "UnittestTestRunner"
197 def identify_tests(self) -> List[Any]:
198 return list(file_utils.expand_globs('*_test.py'))
201 def run_test(self, test: Any) -> TestResults:
202 if config.config['coverage']:
203 cmdline = f'coverage run --source {HOME}/lib {test} --unittests_ignore_perf'
206 return self.execute_commandline(test, cmdline)
209 class DoctestTestRunner(TemplatedTestRunner):
211 def get_name(self) -> str:
212 return "DoctestTestRunner"
215 def identify_tests(self) -> List[Any]:
217 out = exec_utils.cmd('grep -lR "^ *import doctest" /home/scott/lib/python_modules/*')
218 for line in out.split('\n'):
219 if re.match(r'.*\.py$', line):
220 if 'run_tests.py' not in line:
225 def run_test(self, test: Any) -> TestResults:
226 if config.config['coverage']:
227 cmdline = f'coverage run --source {HOME}/lib {test} 2>&1'
229 cmdline = f'python3 {test}'
230 return self.execute_commandline(test, cmdline)
233 class IntegrationTestRunner(TemplatedTestRunner):
235 def get_name(self) -> str:
236 return "IntegrationTestRunner"
239 def identify_tests(self) -> List[Any]:
240 return list(file_utils.expand_globs('*_itest.py'))
243 def run_test(self, test: Any) -> TestResults:
244 if config.config['coverage']:
245 cmdline = f'coverage run --source {HOME}/lib {test}'
248 return self.execute_commandline(test, cmdline)
251 def test_results_report(results: Dict[str, TestResults]):
252 for type, result in results.items():
253 print(text_utils.header(f'{result.name}'))
254 print(f' Ran {len(result.tests_executed)} tests.')
255 print(f' ..{len(result.tests_succeeded)} tests succeeded.')
256 if len(result.tests_failed) > 0:
257 print(f' ..{len(result.tests_failed)} tests failed:')
258 for test in result.tests_failed:
261 if len(result.tests_timed_out) > 0:
262 print(f' ..{len(result.tests_timed_out)} tests timed out:')
263 for test in result.tests_failed:
266 if len(result.tests_failed) + len(result.tests_timed_out):
267 print('Reminder: look in ./test_output to view test output logs')
270 def code_coverage_report():
271 text_utils.header('Code Coverage')
272 out = exec_utils.cmd('coverage combine .coverage*')
273 out = exec_utils.cmd('coverage report --omit=config-3.8.py,*_test.py,*_itest.py --sort=-cover')
277 To recall this report w/o re-running the tests:
279 $ coverage report --omit=config-3.8.py,*_test.py,*_itest.py --sort=-cover
281 ...from the 'tests' directory. Note that subsequent calls to
282 run_tests.py with --coverage will klobber previous results. See:
284 https://coverage.readthedocs.io/en/6.2/
290 @bootstrap.initialize
291 def main() -> Optional[int]:
293 halt_event = threading.Event()
294 threads: List[TestRunner] = []
297 params = TestingParameters(
299 halt_event=halt_event,
302 if config.config['coverage']:
303 logger.debug('Clearing existing coverage data via "coverage erase".')
304 exec_utils.cmd('coverage erase')
306 if config.config['unittests']:
308 threads.append(UnittestTestRunner(params))
309 if config.config['doctests']:
311 threads.append(DoctestTestRunner(params))
312 if config.config['integration']:
314 threads.append(IntegrationTestRunner(params))
318 print('ERROR: one of --unittests, --doctests or --integration is required.')
321 for thread in threads:
324 results: Dict[str, TestResults] = {}
325 while len(results) != len(threads):
326 for thread in threads:
327 if not thread.is_alive():
329 if tid not in results:
330 result = thread.join()
332 results[tid] = result
333 if len(result.tests_failed) > 0:
335 'Thread %s returned abnormal results; killing the others.', tid
340 test_results_report(results)
341 if config.config['coverage']:
342 code_coverage_report()
346 if __name__ == '__main__':