diff --git a/makefile b/makefile index ae04b1b..401da25 100644 --- a/makefile +++ b/makefile @@ -21,7 +21,7 @@ venv: .venv echo 'run `source .venv/bin/activate` to activate virtualenv' test: - python -m unittest_ft -s $(PKG).tests + python -m unittest_ft -rs $(PKG).tests python -m mypy -p $(PKG) lint: diff --git a/unittest_ft/cli.py b/unittest_ft/cli.py index 0ad6fa3..f31fafb 100644 --- a/unittest_ft/cli.py +++ b/unittest_ft/cli.py @@ -13,7 +13,9 @@ @click.command() -@click.option("--debug/--quiet", default=None, help="Increase or decrease logging") +@click.option("--debug", default=None, help="Enable debug logging") +@click.option("--verbose", "-v", is_flag=True, help="Verbose output") +@click.option("--quiet", "-q", is_flag=True, help="Quiet output") @click.option("--stress-test", "-s", is_flag=True, help="Run every test 10 times") @click.option("--randomize", "-r", is_flag=True, help="Randomize test order") @click.option( @@ -26,24 +28,24 @@ ) @click.argument("module") def main( - debug: bool | None, + debug: bool, + verbose: bool, + quiet: bool, module: str, randomize: bool, stress_test: bool, threads: int, ) -> NoReturn: logging.basicConfig( - level=( - logging.DEBUG - if debug - else (logging.WARNING if debug is None else logging.ERROR) - ), + level=(logging.DEBUG if debug else logging.WARNING), stream=sys.stderr, ) + verbosity = 2 if verbose else 0 if quiet else 1 result = run( module, randomize=randomize, stress_test=stress_test, threads=threads, + verbosity=verbosity, ) sys.exit(0 if result.wasSuccessful() else 1) diff --git a/unittest_ft/core.py b/unittest_ft/core.py index d3cdc79..526b72e 100644 --- a/unittest_ft/core.py +++ b/unittest_ft/core.py @@ -7,12 +7,13 @@ import logging import os import random +import sys import time -from concurrent.futures import as_completed, ThreadPoolExecutor -from typing import Generator +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait +from typing import Any, Generator, TextIO from unittest import TestCase, TestLoader, TestResult, TestSuite -from rich import print +# from rich import print from typing_extensions import Self LOG = logging.getLogger(__name__) @@ -21,17 +22,58 @@ class FTTestResult(TestResult): + def __init__( + self, + stream: TextIO | None = None, + descriptions: bool | None = None, + verbosity: int | None = None, + ) -> None: + super().__init__(stream=stream, descriptions=descriptions, verbosity=verbosity) + self.verbosity = verbosity or 1 + self.before = time.monotonic_ns() + self.duration = 0 + self.collected_duration = 0 + + def stopTest(self, test: Any) -> None: + super().stopTest(test) + self.duration = time.monotonic_ns() - self.before + + def stopTestRun(self) -> None: + super().stopTestRun() + self.duration = time.monotonic_ns() - self.before + def __str__(self) -> str: items = [(f"ERROR: {test_case}", trace) for test_case, trace in self.errors] items += [(f"FAIL: {test_case}", trace) for test_case, trace in self.failures] longest = max(len(label) for label, _ in items) if items else 70 - msg = "\n".join( + msg = "\n" + msg += "\n".join( f"{'=' * longest}\n{label}\n{'-' * longest}\n{trace}" for label, trace in items ) msg += "-" * longest + msg += f"\nRan {self.testsRun} tests in {format_ns(self.duration)}" + + saved = self.collected_duration - self.duration + if saved > 0 and (saved / self.duration) > 0.10: + msg += f" (saved {format_ns(self.collected_duration - self.duration)})" + msg += "\n\n" + + if self.wasSuccessful(): + msg += "OK" + else: + parts = [] + if self.errors: + parts += [f"errors={len(self.errors)}"] + if self.failures: + parts += [f"failures={len(self.failures)}"] + if self.skipped: + parts += [f"skipped={len(self.skipped)}"] + if self.expectedFailures: + parts += [f"expected failures={len(self.expectedFailures)}"] + msg += f"FAILED ({', '.join(parts)})" return msg @@ -48,6 +90,8 @@ def __add__(self, other: object) -> FTTestResult: result.unexpectedSuccesses = ( self.unexpectedSuccesses + other.unexpectedSuccesses ) + if isinstance(other, FTTestResult): + result.collected_duration = self.duration + other.duration return result def __iadd__(self, other: object) -> Self: @@ -60,6 +104,8 @@ def __iadd__(self, other: object) -> Self: self.skipped += other.skipped self.testsRun += other.testsRun self.unexpectedSuccesses += other.unexpectedSuccesses + if isinstance(other, FTTestResult): + self.collected_duration += other.duration return self @@ -71,17 +117,15 @@ def get_individual_tests(suite: TestSuite) -> Generator[TestCase, None, None]: yield test -def run_single_test(test_id: str) -> tuple[str, TestResult, int]: +def run_single_test(test_id: str) -> tuple[str, FTTestResult]: LOG.debug("Loading test %s", test_id) loader = TestLoader() - result = FTTestResult(descriptions=True, verbosity=2) suite = loader.loadTestsFromName(test_id) LOG.debug("Running test %s", test_id) - before = time.monotonic_ns() + result = FTTestResult(descriptions=True, verbosity=2) suite.run(result) - duration = time.monotonic_ns() - before LOG.debug("Finished test %s", test_id) - return (test_id, result, duration) + return (test_id, result) def format_ns(duration: int) -> str: @@ -97,13 +141,13 @@ def run( randomize: bool = False, stress_test: bool = False, threads: int = DEFAULT_THREADS, + verbosity: int = 1, ) -> TestResult: loaded_module = importlib.import_module(module) loader = TestLoader() suite = loader.loadTestsFromModule(loaded_module) LOG.debug("loaded %d test cases from %s", suite.countTestCases(), module) - before = time.monotonic_ns() test_ids = [test.id() for test in get_individual_tests(suite)] if stress_test: test_ids = test_ids * 10 @@ -114,33 +158,25 @@ def run( LOG.debug("ready to run %d tests:\n %s", len(test_ids), "\n ".join(test_ids)) pool = ThreadPoolExecutor(max_workers=threads) - futs = [pool.submit(run_single_test, test_id) for test_id in test_ids] + futs = {pool.submit(run_single_test, test_id) for test_id in test_ids} - test_duration = 0 + stream = sys.stdout result = FTTestResult() + while futs: + done, futs = wait(futs, timeout=0.1, return_when=FIRST_COMPLETED) + for fut in done: + test_id, test_result = fut.result() + if verbosity == 2: + stream.write( + f"{test_id} ... {'OK' if test_result.wasSuccessful() else 'FAIL'} " + f" {format_ns(test_result.duration)}\n" + ) + elif verbosity == 1: + stream.write("." if test_result.wasSuccessful() else "F") + stream.flush() + result += test_result + result.stopTestRun() - for fut in as_completed(futs): - test_id, test_result, duration = fut.result() - print( - f"{test_id} ... {'OK' if test_result.wasSuccessful() else 'FAIL'} " - f" {format_ns(duration)}" - ) - - test_duration += duration - result += test_result - - runner_duration = time.monotonic_ns() - before - - print() print(result) - print( - f"Ran {len(test_ids)} tests in {format_ns(runner_duration)} " - f"(saved {format_ns(test_duration - runner_duration)})\n" - ) - - if result.wasSuccessful(): - print("OK") - else: - print("FAILED") return result