4 A smart, fast test runner. Used in a git pre-commit hook.
13 from abc import ABC, abstractmethod
14 from dataclasses import dataclass
15 from typing import Any, Dict, List, Optional
17 from overrides import overrides
24 import parallelize as par
28 logger = logging.getLogger(__name__)
29 args = config.add_commandline_args(f'({__file__})', 'Args related to __file__')
30 args.add_argument('--unittests', '-u', action='store_true', help='Run unittests.')
31 args.add_argument('--doctests', '-d', action='store_true', help='Run doctests.')
32 args.add_argument('--integration', '-i', action='store_true', help='Run integration tests.')
34 '--coverage', '-c', action='store_true', help='Run tests and capture code coverage data'
37 HOME = os.environ['HOME']
41 class TestingParameters:
43 """Should we stop as soon as one error has occurred?"""
45 halt_event: threading.Event
46 """An event that, when set, indicates to stop ASAP."""
52 """The name of this test / set of tests."""
54 tests_executed: List[str]
55 """Tests that were executed."""
57 tests_succeeded: List[str]
58 """Tests that succeeded."""
60 tests_failed: List[str]
61 """Tests that failed."""
63 tests_timed_out: List[str]
64 """Tests that timed out."""
66 def __add__(self, other):
67 self.tests_executed.extend(other.tests_executed)
68 self.tests_succeeded.extend(other.tests_succeeded)
69 self.tests_failed.extend(other.tests_failed)
70 self.tests_timed_out.extend(other.tests_timed_out)
75 def __repr__(self) -> str:
76 out = f'{self.name}: '
77 out += f'{ansi.fg("green")}'
78 out += f'{len(self.tests_succeeded)}/{len(self.tests_executed)} passed'
79 out += f'{ansi.reset()}.\n'
81 if len(self.tests_failed) > 0:
82 out += f' ..{ansi.fg("red")}'
83 out += f'{len(self.tests_failed)} tests failed'
84 out += f'{ansi.reset()}:\n'
85 for test in self.tests_failed:
89 if len(self.tests_timed_out) > 0:
90 out += f' ..{ansi.fg("yellow")}'
91 out += f'{len(self.tests_timed_out)} tests timed out'
92 out += f'{ansi.reset()}:\n'
93 for test in self.tests_failed:
99 class TestRunner(ABC, thread_utils.ThreadWithReturnValue):
100 """A Base class for something that runs a test."""
102 def __init__(self, params: TestingParameters):
103 """Create a TestRunner.
106 params: Test running paramters.
109 super().__init__(self, target=self.begin, args=[params])
111 self.test_results = TestResults(
112 name=self.get_name(),
120 def get_name(self) -> str:
121 """The name of this test collection."""
125 def begin(self, params: TestingParameters) -> TestResults:
126 """Start execution."""
130 class TemplatedTestRunner(TestRunner, ABC):
131 """A TestRunner that has a recipe for executing the tests."""
134 def identify_tests(self) -> List[str]:
135 """Return a list of tests that should be executed."""
139 def run_test(self, test: Any) -> TestResults:
140 """Run a single test and return its TestResults."""
143 def check_for_abort(self):
144 """Periodically caled to check to see if we need to stop."""
146 if self.params.halt_event.is_set():
147 logger.debug('Thread %s saw halt event; exiting.', self.get_name())
148 raise Exception("Kill myself!")
149 if self.params.halt_on_error:
150 if len(self.test_results.tests_failed) > 0:
151 logger.error('Thread %s saw abnormal results; exiting.', self.get_name())
152 raise Exception("Kill myself!")
154 def status_report(self, running: List[Any], done: List[Any]):
155 """Periodically called to report current status."""
157 total = len(running) + len(done)
159 '%s: %d/%d in flight; %d/%d completed.',
167 def persist_output(self, test_name: str, message: str, output: str) -> None:
168 """Called to save the output of a test run."""
170 basename = file_utils.without_path(test_name)
171 dest = f'{basename}-output.txt'
172 with open(f'./test_output/{dest}', 'w') as wf:
173 print(message, file=wf)
174 print('-' * len(message), file=wf)
177 def execute_commandline(
182 timeout: float = 120.0,
184 """Execute a particular commandline to run a test."""
187 logger.debug('%s: Running %s (%s)', self.get_name(), test_name, cmdline)
188 output = exec_utils.cmd(
190 timeout_seconds=timeout,
192 self.persist_output(test_name, f'{test_name} ({cmdline}) succeeded.', output)
193 logger.debug('%s (%s) succeeded', test_name, cmdline)
194 return TestResults(test_name, [test_name], [test_name], [], [])
195 except subprocess.TimeoutExpired as e:
196 msg = f'{self.get_name()}: {test_name} ({cmdline}) timed out after {e.timeout:.1f} seconds.'
199 '%s: %s output when it timed out: %s', self.get_name(), test_name, e.output
201 self.persist_output(test_name, msg, e.output)
209 except subprocess.CalledProcessError as e:
210 msg = f'{self.get_name()}: {test_name} ({cmdline}) failed; exit code {e.returncode}'
212 logger.debug('%s: %s output when it failed: %s', self.get_name(), test_name, e.output)
213 self.persist_output(test_name, msg, e.output)
223 def begin(self, params: TestingParameters) -> TestResults:
224 logger.debug('Thread %s started.', self.get_name())
225 interesting_tests = self.identify_tests()
226 running: List[Any] = []
228 for test in interesting_tests:
229 running.append(self.run_test(test))
231 while len(running) > 0:
232 self.status_report(running, done)
233 self.check_for_abort()
237 newly_finished.append(fut)
238 result = fut._resolve()
239 logger.debug('Test %s finished.', result.name)
240 self.test_results += result
242 for fut in newly_finished:
247 logger.debug('Thread %s finished.', self.get_name())
248 return self.test_results
251 class UnittestTestRunner(TemplatedTestRunner):
252 """Run all known Unittests."""
255 def get_name(self) -> str:
256 return "UnittestTestRunner"
259 def identify_tests(self) -> List[str]:
260 return list(file_utils.expand_globs('*_test.py'))
263 def run_test(self, test: Any) -> TestResults:
264 if config.config['coverage']:
265 cmdline = f'coverage run --source {HOME}/lib {test} --unittests_ignore_perf'
268 return self.execute_commandline(test, cmdline)
271 class DoctestTestRunner(TemplatedTestRunner):
272 """Run all known Doctests."""
275 def get_name(self) -> str:
276 return "DoctestTestRunner"
279 def identify_tests(self) -> List[str]:
281 out = exec_utils.cmd('grep -lR "^ *import doctest" /home/scott/lib/python_modules/*')
282 for line in out.split('\n'):
283 if re.match(r'.*\.py$', line):
284 if 'run_tests.py' not in line:
289 def run_test(self, test: Any) -> TestResults:
290 if config.config['coverage']:
291 cmdline = f'coverage run --source {HOME}/lib {test} 2>&1'
293 cmdline = f'python3 {test}'
294 return self.execute_commandline(test, cmdline)
297 class IntegrationTestRunner(TemplatedTestRunner):
298 """Run all know Integration tests."""
301 def get_name(self) -> str:
302 return "IntegrationTestRunner"
305 def identify_tests(self) -> List[str]:
306 return list(file_utils.expand_globs('*_itest.py'))
309 def run_test(self, test: Any) -> TestResults:
310 if config.config['coverage']:
311 cmdline = f'coverage run --source {HOME}/lib {test}'
314 return self.execute_commandline(test, cmdline)
317 def test_results_report(results: Dict[str, TestResults]) -> int:
318 """Give a final report about the tests that were run."""
320 for result in results.values():
321 print(result, end='')
322 total_problems += len(result.tests_failed)
323 total_problems += len(result.tests_timed_out)
325 if total_problems > 0:
326 print('Reminder: look in ./test_output to view test output logs')
327 return total_problems
330 def code_coverage_report():
331 """Give a final code coverage report."""
332 text_utils.header('Code Coverage')
333 exec_utils.cmd('coverage combine .coverage*')
334 out = exec_utils.cmd('coverage report --omit=config-3.8.py,*_test.py,*_itest.py --sort=-cover')
338 To recall this report w/o re-running the tests:
340 $ coverage report --omit=config-3.8.py,*_test.py,*_itest.py --sort=-cover
342 ...from the 'tests' directory. Note that subsequent calls to
343 run_tests.py with --coverage will klobber previous results. See:
345 https://coverage.readthedocs.io/en/6.2/
350 @bootstrap.initialize
351 def main() -> Optional[int]:
353 halt_event = threading.Event()
354 threads: List[TestRunner] = []
357 params = TestingParameters(
359 halt_event=halt_event,
362 if config.config['coverage']:
363 logger.debug('Clearing existing coverage data via "coverage erase".')
364 exec_utils.cmd('coverage erase')
366 if config.config['unittests']:
368 threads.append(UnittestTestRunner(params))
369 if config.config['doctests']:
371 threads.append(DoctestTestRunner(params))
372 if config.config['integration']:
374 threads.append(IntegrationTestRunner(params))
378 print('ERROR: one of --unittests, --doctests or --integration is required.')
381 for thread in threads:
384 results: Dict[str, TestResults] = {}
385 while len(results) != len(threads):
386 for thread in threads:
387 if not thread.is_alive():
389 if tid not in results:
390 result = thread.join()
392 results[tid] = result
393 if len(result.tests_failed) > 0:
395 'Thread %s returned abnormal results; killing the others.', tid
400 if config.config['coverage']:
401 code_coverage_report()
402 total_problems = test_results_report(results)
403 return total_problems
406 if __name__ == '__main__':