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 num_tests_executed: int
49 num_tests_succeeded: int
55 class TestRunner(ABC, thread_utils.ThreadWithReturnValue):
56 def __init__(self, params: TestingParameters):
57 super().__init__(self, target=self.begin, args=[params])
59 self.test_results = TestResults(
60 name=f"All {self.get_name()} tests",
62 num_tests_succeeded=0,
68 def aggregate_test_results(self, result: TestResults):
69 self.test_results.num_tests_executed += result.num_tests_executed
70 self.test_results.num_tests_succeeded += result.num_tests_succeeded
71 self.test_results.num_tests_failed += result.num_tests_failed
72 self.test_results.normal_exit = self.test_results.normal_exit and result.normal_exit
73 self.test_results.output += "\n\n\n" + result.output
76 def get_name(self) -> str:
80 def begin(self, params: TestingParameters) -> TestResults:
84 class TemplatedTestRunner(TestRunner, ABC):
86 def identify_tests(self) -> List[Any]:
90 def run_test(self, test: Any) -> TestResults:
93 def check_for_abort(self):
94 if self.params.halt_event.is_set():
95 logger.debug('Thread %s saw halt event; exiting.', self.get_name())
96 raise Exception("Kill myself!")
97 if not self.test_results.normal_exit:
98 if self.params.halt_on_error:
99 logger.error('Thread %s saw abnormal results; exiting.', self.get_name())
100 raise Exception("Kill myself!")
102 def status_report(self, running: List[Any], done: List[Any]):
103 total = len(running) + len(done)
105 '%s: %d/%d in flight; %d/%d completed.',
114 def begin(self, params: TestingParameters) -> TestResults:
115 logger.debug('Thread %s started.', self.get_name())
116 interesting_tests = self.identify_tests()
117 running: List[Any] = []
119 for test in interesting_tests:
120 running.append(self.run_test(test))
122 while len(running) > 0:
123 self.status_report(running, done)
124 self.check_for_abort()
128 newly_finished.append(fut)
129 result = fut._resolve()
130 logger.debug('Test %s finished.', result.name)
131 self.aggregate_test_results(result)
133 for fut in newly_finished:
138 logger.debug('Thread %s finished.', self.get_name())
139 return self.test_results
142 class UnittestTestRunner(TemplatedTestRunner):
144 def get_name(self) -> str:
145 return "UnittestTestRunner"
148 def identify_tests(self) -> List[Any]:
149 return list(file_utils.expand_globs('*_test.py'))
152 def run_test(self, test: Any) -> TestResults:
153 if config.config['coverage']:
154 cmdline = f'coverage run --source {HOME}/lib --append {test} --unittests_ignore_perf'
159 logger.debug('Running unittest %s (%s)', test, cmdline)
160 output = exec_utils.cmd(
162 timeout_seconds=120.0,
165 logger.error('Unittest %s timed out; ran for > 120.0 seconds', test)
172 f"Unittest {test} timed out.",
174 except subprocess.CalledProcessError:
175 logger.error('Unittest %s failed.', test)
182 f"Unittest {test} failed.",
184 return TestResults(test, 1, 1, 0, True, output)
187 class DoctestTestRunner(TemplatedTestRunner):
189 def get_name(self) -> str:
190 return "DoctestTestRunner"
193 def identify_tests(self) -> List[Any]:
195 out = exec_utils.cmd('grep -lR "^ *import doctest" /home/scott/lib/python_modules/*')
196 for line in out.split('\n'):
197 if re.match(r'.*\.py$', line):
198 if 'run_tests.py' not in line:
203 def run_test(self, test: Any) -> TestResults:
204 if config.config['coverage']:
205 cmdline = f'coverage run --source {HOME}/lib --append {test} 2>&1'
207 cmdline = f'python3 {test}'
209 logger.debug('Running doctest %s (%s).', test, cmdline)
210 output = exec_utils.cmd(
212 timeout_seconds=120.0,
215 logger.error('Doctest %s timed out; ran for > 120.0 seconds', test)
222 f"Doctest {test} timed out.",
224 except subprocess.CalledProcessError:
225 logger.error('Doctest %s failed.', test)
232 f"Docttest {test} failed.",
244 class IntegrationTestRunner(TemplatedTestRunner):
246 def get_name(self) -> str:
247 return "IntegrationTestRunner"
250 def identify_tests(self) -> List[Any]:
251 return list(file_utils.expand_globs('*_itest.py'))
254 def run_test(self, test: Any) -> TestResults:
255 if config.config['coverage']:
256 cmdline = f'coverage run --source {HOME}/lib --append {test}'
260 logger.debug('Running integration test %s (%s).', test, cmdline)
261 output = exec_utils.cmd(
263 timeout_seconds=240.0,
266 logger.error('Integration Test %s timed out; ran for > 240.0 seconds', test)
273 f"Integration Test {test} timed out.",
275 except subprocess.CalledProcessError:
276 logger.error('Integration Test %s failed.', test)
283 f"Integration Test {test} failed.",
295 def test_results_report(results: Dict[str, TestResults]):
299 def code_coverage_report():
300 text_utils.header('Code Coverage')
301 out = exec_utils.cmd('coverage report --omit=config-3.8.py,*_test.py,*_itest.py --sort=-cover')
305 To recall this report w/o re-running the tests:
307 $ coverage report --omit=config-3.8.py,*_test.py,*_itest.py --sort=-cover
309 ...from the 'tests' directory. Note that subsequent calls to
310 run_tests.py with --coverage will klobber previous results. See:
312 https://coverage.readthedocs.io/en/6.2/
318 @bootstrap.initialize
319 def main() -> Optional[int]:
321 halt_event = threading.Event()
322 threads: List[TestRunner] = []
325 params = TestingParameters(
327 halt_event=halt_event,
330 if config.config['coverage']:
331 logger.debug('Clearing existing coverage data via "coverage erase".')
332 exec_utils.cmd('coverage erase')
334 if config.config['unittests']:
336 threads.append(UnittestTestRunner(params))
337 if config.config['doctests']:
339 threads.append(DoctestTestRunner(params))
340 if config.config['integration']:
342 threads.append(IntegrationTestRunner(params))
346 print('ERROR: one of --unittests, --doctests or --integration is required.')
349 for thread in threads:
352 results: Dict[str, TestResults] = {}
353 while len(results) != len(threads):
354 for thread in threads:
355 if not thread.is_alive():
357 if tid not in results:
358 result = thread.join()
360 results[tid] = result
361 if not result.normal_exit:
365 test_results_report(results)
366 if config.config['coverage']:
367 code_coverage_report()
371 if __name__ == '__main__':