#!/usr/bin/env python3 """Tests for the Python check runner that reports failures as they happen. These tests are pure standard-library ``unittest`` and need no compiler, no network, and no hardware. They exercise ``tools/run_python_checks.py`` through real child processes against small temporary suites, because the property under test is about *when* bytes reach the output stream, which an in-process result double cannot show: - a passing suite keeps the plain ``Ran N tests``/``OK`` summary and exit status, - a failure, an error, and a failing subtest each write a formatted traceback block immediately, once, with the test's elapsed time, and the usual ``FAILED (...)`` summary and exit status follow, - a later test that blocks until the earlier failure has been read proves the report is written while the run is still going, not at the end, - discovery selects exactly the same tests as ``python3 -m unittest discover -s tests/python -v``, - every outcome the helper does not add a block for (a skip, an expected failure, a class-level ``setUpClass``/``tearDownClass`` error) and every one it does (an error from ``addCleanup``, an error in a subtest, an unexpected success) keeps plain ``unittest``'s counts, status words and exit status, - a timing line appears only for tests at or above the slow-test threshold, - the ``check-python`` recipe runs the helper and still runs the distribution-document check, and the helper ships in the source manifest. What a passing run does not prove: that the tracebacks of an entire real suite fit inside any particular output cap. The helper bounds what it *adds*; a suite whose failures alone exceed a caller's cap can still be truncated. """ from __future__ import annotations import json import queue import re import subprocess import sys import tempfile import textwrap import threading import time import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] TOOLS = ROOT / "tools" HELPER = TOOLS / "run_python_checks.py" MAKEFILE = ROOT / "Makefile" MANIFEST = TOOLS / "release_source_manifest.json" # Every wait in this file is bounded. The handshake test needs the child to # start Python, run one tiny test and write one block, so this is generous # rather than tight. WAIT_SECONDS = 15.0 SUBPROCESS_TIMEOUT_SECONDS = 300.0 def write_suite(directory: Path, name: str, source: str) -> Path: """Write one discovered test module into ``directory``.""" path = directory / f"{name}.py" path.write_text(textwrap.dedent(source).lstrip("\n"), encoding="utf-8") return path def run_helper(start_directory: Path, *extra: str) -> subprocess.CompletedProcess: command = [ sys.executable, str(HELPER), "--start-directory", str(start_directory), *extra, ] return subprocess.run( command, cwd=str(ROOT), capture_output=True, text=True, timeout=SUBPROCESS_TIMEOUT_SECONDS, check=False, ) def run_plain_unittest(start_directory: Path) -> subprocess.CompletedProcess: command = [ sys.executable, "-m", "unittest", "discover", "-s", str(start_directory), "-v", ] return subprocess.run( command, cwd=str(ROOT), capture_output=True, text=True, timeout=SUBPROCESS_TIMEOUT_SECONDS, check=False, ) def ran_test_count(output: str) -> int: match = re.search(r"^Ran (\d+) tests? in ", output, re.MULTILINE) if match is None: raise AssertionError(f"no unittest summary line in output:\n{output}") return int(match.group(1)) def selected_tests(output: str) -> set[tuple[str, str]]: """The ``(name, qualified)`` pairs the verbose runner names as it starts.""" return set(re.findall(r"^(\w+) \(([\w.]+)\)", output, re.MULTILINE)) MIXED_SUITE = """ import unittest class Mixed(unittest.TestCase): def test_ok(self): pass def test_fails(self): self.assertNotEqual("actual", "actual", "marker-failure") def test_errors(self): raise RuntimeError("marker-error") def test_subtests(self): for case in ("alpha", "beta"): with self.subTest(case=case): self.assertEqual(case, "alpha", f"marker-subtest-{case}") """ class ImmediateReportTests(unittest.TestCase): """Each problem is written with its traceback before the run finishes.""" def test_a_passing_suite_keeps_the_plain_summary(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-pass-") as workdir: root = Path(workdir) write_suite( root, "test_passing", """ import unittest class Passing(unittest.TestCase): def test_first(self): pass def test_second(self): pass """, ) completed = run_helper(root) self.assertEqual(completed.returncode, 0, completed.stderr) self.assertEqual(ran_test_count(completed.stderr), 2) self.assertEqual(completed.stderr.rstrip().splitlines()[-1], "OK") self.assertIn(" ... ok", completed.stderr) self.assertNotIn("[slow ", completed.stderr) def test_a_failure_reports_its_traceback_before_the_summary(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-fail-") as workdir: root = Path(workdir) write_suite(root, "test_failing", MIXED_SUITE) completed = run_helper(root) output = completed.stderr self.assertEqual(completed.returncode, 1, output) block = re.search( r"^={70}\nFAIL: test_fails \(test_failing\.Mixed\.test_fails\) after " r"\d+\.\d{2}s\n-{70}\nTraceback \(most recent call last\):", output, re.MULTILINE, ) self.assertIsNotNone(block, output) self.assertIn("AssertionError: 'actual' == 'actual' : marker-failure", output) # The immediate block comes before the summary, not after it. self.assertLess(output.index("FAIL: test_fails"), output.index("Ran 4 tests")) self.assertIn("FAILED (failures=2, errors=1)", output) def test_an_error_is_distinguished_from_a_failure(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-error-") as workdir: root = Path(workdir) write_suite( root, "test_erroring", """ import unittest class Erroring(unittest.TestCase): def test_raises(self): raise RuntimeError("marker-error") """, ) completed = run_helper(root) output = completed.stderr self.assertEqual(completed.returncode, 1, output) self.assertIn("ERROR: test_raises (test_erroring.Erroring.test_raises)", output) self.assertIn("RuntimeError: marker-error", output) self.assertNotIn("FAIL: test_raises", output) self.assertIn("FAILED (errors=1)", output) self.assertEqual(ran_test_count(output), 1) def test_a_failing_subtest_reports_its_parameters(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-subtest-") as workdir: root = Path(workdir) write_suite( root, "test_subtest", """ import unittest class Subtests(unittest.TestCase): def test_cases(self): for case in ("alpha", "beta"): with self.subTest(case=case): self.assertEqual(case, "alpha", f"marker-subtest-{case}") """, ) completed = run_helper(root) output = completed.stderr self.assertEqual(completed.returncode, 1, output) self.assertIn( "FAIL: test_cases (test_subtest.Subtests.test_cases) (case='beta')", output ) self.assertIn("AssertionError: 'beta' != 'alpha'", output) self.assertIn("marker-subtest-beta", output) # A failing subtest does not add a test to the count. self.assertEqual(ran_test_count(output), 1) self.assertIn("FAILED (failures=1)", output) self.assertNotIn("marker-subtest-alpha", output) def test_a_traceback_is_not_reprinted_after_the_run(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-once-") as workdir: root = Path(workdir) write_suite( root, "test_once", """ import unittest class Once(unittest.TestCase): def test_fails(self): self.assertNotEqual("actual", "actual", "marker-failure") """, ) completed = run_helper(root) output = completed.stderr self.assertEqual(completed.returncode, 1, output) self.assertEqual(output.count("FAIL: test_fails ("), 1, output) self.assertEqual(output.count("AssertionError"), 1, output) self.assertEqual(output.count("Traceback (most recent call last)"), 1, output) self.assertIn("(each failure above is reported once, when it happened)", output) # The summary itself is unchanged. self.assertIn("FAILED (failures=1)", output) self.assertEqual(ran_test_count(output), 1) def test_a_failure_is_written_before_a_later_test_finishes(self): """The report reaches the reader while the suite is still running. The second test blocks until this process has read the first test's traceback, then returns. If the runner deferred its tracebacks to the end of the run, the reader would see nothing, the child would exhaust its bounded wait, and that second test would fail with the marker below. """ with tempfile.TemporaryDirectory(prefix="ut-python-checks-handshake-") as workdir: root = Path(workdir) release = root / "report-seen" write_suite( root, "test_handshake", f""" import time import unittest from pathlib import Path RELEASE = Path({str(release)!r}) WAIT_SECONDS = {WAIT_SECONDS!r} UNREPORTED = "the earlier failure was not reported while the run was still going" class Handshake(unittest.TestCase): def test_a_fails_first(self): self.fail("first failure marker") def test_b_waits_for_the_report(self): deadline = time.monotonic() + WAIT_SECONDS while time.monotonic() < deadline: if RELEASE.exists(): return time.sleep(0.01) self.fail(UNREPORTED) """, ) command = [sys.executable, str(HELPER), "--start-directory", str(root)] child = subprocess.Popen( command, cwd=str(ROOT), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, ) lines: list[str] = [] sink: queue.Queue[str | None] = queue.Queue() def pump() -> None: assert child.stderr is not None for line in child.stderr: sink.put(line) sink.put(None) threading.Thread(target=pump, daemon=True).start() try: deadline = time.monotonic() + WAIT_SECONDS while True: remaining = deadline - time.monotonic() self.assertGreater( remaining, 0.0, "the runner wrote nothing while the suite was still running", ) try: line = sink.get(timeout=remaining) except queue.Empty: self.fail("the runner wrote nothing while the suite was still running") if line is None: break lines.append(line) if "first failure marker" in line: release.touch() break child.wait(timeout=WAIT_SECONDS * 4) while True: try: line = sink.get(timeout=WAIT_SECONDS) except queue.Empty: break if line is None: break lines.append(line) finally: if child.poll() is None: child.kill() child.wait(timeout=WAIT_SECONDS * 2) if child.stderr is not None: child.stderr.close() output = "".join(lines) self.assertEqual(child.returncode, 1, output) self.assertNotIn( "the earlier failure was not reported while the run was still going", output ) self.assertIn("FAILED (failures=1)", output) self.assertEqual(ran_test_count(output), 2) class DiscoveryParityTests(unittest.TestCase): """The helper selects the tests the documented command selected.""" def test_the_same_tests_are_selected_as_plain_unittest(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-parity-") as workdir: root = Path(workdir) write_suite( root, "test_alpha", """ import unittest class Alpha(unittest.TestCase): def test_one(self): pass def test_subtests(self): for case in (1, 2): with self.subTest(case=case): self.assertEqual(case, case) """, ) write_suite( root, "test_beta", """ import unittest class Beta(unittest.TestCase): def test_two(self): pass """, ) helper = run_helper(root) plain = run_plain_unittest(root) self.assertEqual(helper.returncode, 0, helper.stderr) self.assertEqual(plain.returncode, 0, plain.stderr) self.assertEqual(ran_test_count(helper.stderr), ran_test_count(plain.stderr)) self.assertEqual(ran_test_count(helper.stderr), 3) self.assertEqual(selected_tests(helper.stderr), selected_tests(plain.stderr)) self.assertEqual( selected_tests(helper.stderr), { ("test_one", "test_alpha.Alpha.test_one"), ("test_subtests", "test_alpha.Alpha.test_subtests"), ("test_two", "test_beta.Beta.test_two"), }, ) def test_the_default_start_directory_is_the_checked_in_suite(self): source = HELPER.read_text(encoding="utf-8") self.assertIn('DEFAULT_START_DIRECTORY = "tests/python"', source) self.assertIn('TEST_PATTERN = "test*.py"', source) PARITY_CASES = ( { "stem": "skipped", "source": """ import unittest class Skipped(unittest.TestCase): def test_skipped(self): self.skipTest("marker-skip-reason") """, "returncode": 0, "ran": 1, "summary": "OK (skipped=1)", "header": None, "timed": False, "marker": "... skipped 'marker-skip-reason'", "absent": ("Traceback (most recent call last)", "FAIL:", "ERROR:"), }, { "stem": "expected_failure", "source": """ import unittest class Expected(unittest.TestCase): @unittest.expectedFailure def test_expected_failure(self): self.assertEqual(1, 2, "marker-expected-failure") """, "returncode": 0, "ran": 1, "summary": "OK (expected failures=1)", "header": None, # An expected failure is muted; its assertion text must not be written. "timed": False, "marker": "... expected failure", "absent": ("Traceback (most recent call last)", "marker-expected-failure"), }, { "stem": "unexpected_success", "source": """ import unittest class Unexpected(unittest.TestCase): @unittest.expectedFailure def test_unexpected_success(self): pass """, "returncode": 1, "ran": 1, "summary": "FAILED (unexpected successes=1)", "header": None, "timed": False, # Written once, in the same header plain unittest uses, not as a FAIL. "marker": ( "UNEXPECTED SUCCESS: test_unexpected_success " "(test_parity_unexpected_success.Unexpected.test_unexpected_success)" ), "absent": ("Traceback (most recent call last)", "FAIL: test_unexpected_success"), }, { "stem": "subtest_error", "source": """ import unittest class SubtestError(unittest.TestCase): def test_cases(self): with self.subTest(case="boom"): raise RuntimeError("marker-subtest-error") """, "returncode": 1, "ran": 1, "summary": "FAILED (errors=1)", "header": ( "ERROR: test_cases " "(test_parity_subtest_error.SubtestError.test_cases) (case='boom')" ), "timed": True, "marker": "RuntimeError: marker-subtest-error", # A raised subtest is an error, not a failure, and adds no test to the count. "absent": ("FAIL: test_cases",), }, { "stem": "cleanup_error", "source": """ import unittest def boom(): raise ValueError("marker-cleanup-error") class Cleanup(unittest.TestCase): def test_cleanup(self): self.addCleanup(boom) """, "returncode": 1, "ran": 1, "summary": "FAILED (errors=1)", "header": "ERROR: test_cleanup (test_parity_cleanup_error.Cleanup.test_cleanup)", "timed": True, "marker": "ValueError: marker-cleanup-error", "absent": (), }, { "stem": "setUpClass_error", "source": """ import unittest class SetupClassErrors(unittest.TestCase): @classmethod def setUpClass(cls): raise OSError("marker-setupclass-error") def test_never_runs(self): pass """, "returncode": 5, # The only test never starts, so it is not counted and no timing exists. # With no test started and none skipped, unittest's own exit status is 5. "ran": 0, "summary": "FAILED (errors=1)", "header": "ERROR: setUpClass (test_parity_setUpClass_error.SetupClassErrors)", "timed": False, "marker": "OSError: marker-setupclass-error", "absent": ("test_never_runs",), }, { "stem": "tearDownClass_error", "source": """ import unittest class TearDownClassErrors(unittest.TestCase): @classmethod def tearDownClass(cls): raise OSError("marker-teardownclass-error") def test_passes(self): pass """, "returncode": 1, # The test itself ran and passed; only the class teardown is an error. "ran": 1, "summary": "FAILED (errors=1)", "header": ( "ERROR: tearDownClass (test_parity_tearDownClass_error.TearDownClassErrors)" ), "timed": False, "marker": "OSError: marker-teardownclass-error", "absent": (), }, { "stem": "empty_discovery", "source": """ # A module that defines no test: discovery selects nothing. """, # An empty run must not be reported as a passing one. "returncode": 5, "ran": 0, "summary": "NO TESTS RAN", "header": None, "timed": False, "marker": "NO TESTS RAN", "absent": ("Traceback (most recent call last)", "OK"), }, ) def summary_line(output: str) -> str: """The last non-empty line: ``OK ...`` or ``FAILED (...)``.""" lines = [line for line in output.splitlines() if line.strip()] if not lines: raise AssertionError("the runner wrote nothing") return lines[-1] class OutcomeParityTests(unittest.TestCase): """Every outcome is counted and worded exactly as plain unittest words it.""" def test_each_outcome_keeps_plain_unittest_counts_and_status(self): for case in PARITY_CASES: with self.subTest(case["stem"]): with tempfile.TemporaryDirectory( prefix=f"ut-python-checks-parity-{case['stem']}-" ) as workdir: root = Path(workdir) write_suite(root, f"test_parity_{case['stem']}", case["source"]) helper = run_helper(root) plain = run_plain_unittest(root) output = helper.stderr reference = plain.stderr # The report is added to; the outcome is not. self.assertEqual(helper.returncode, case["returncode"], output) self.assertEqual(helper.returncode, plain.returncode, (output, reference)) self.assertEqual(ran_test_count(output), case["ran"], output) self.assertEqual(ran_test_count(output), ran_test_count(reference)) self.assertEqual(summary_line(output), case["summary"], output) self.assertEqual(summary_line(output), summary_line(reference)) self.assertEqual(selected_tests(output), selected_tests(reference)) # Blocks already written are not printed again at the end, so the # total number of tracebacks matches the plain command's. self.assertEqual( output.count("Traceback (most recent call last)"), reference.count("Traceback (most recent call last)"), output, ) for text in case["absent"]: self.assertNotIn(text, output) self.assertEqual(output.count(case["marker"]), 1, output) if case["header"] is None: continue self.assertEqual(output.count(case["header"]), 1, output) block = re.search( r"^={70}\n" + re.escape(case["header"]) + r"( after \d+\.\d{2}s)?\n-{70}\n" r"Traceback \(most recent call last\):", output, re.MULTILINE, ) self.assertIsNotNone(block, output) self.assertEqual(block.group(1) is not None, case["timed"], output) # The block was written during the run, not after the summary. self.assertLess(output.index(case["header"]), output.index("Ran ")) class TimingLineTests(unittest.TestCase): """A timing line is printed only for tests at or above the threshold.""" def test_every_test_at_threshold_zero_gets_a_readable_line(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-timing-") as workdir: root = Path(workdir) write_suite( root, "test_timing", """ import unittest class Timing(unittest.TestCase): def test_first(self): pass def test_second(self): pass """, ) completed = run_helper(root, "--slow-seconds", "0") output = completed.stderr self.assertEqual(completed.returncode, 0, output) lines = [ line for line in output.splitlines() if line.startswith("[slow ") ] self.assertEqual(len(lines), 2, output) for line in lines: self.assertRegex(line, r"^\[slow \d+\.\d{2}s\] \w+ \([\w.]+\)$") def test_a_threshold_above_every_test_keeps_the_log_quiet(self): with tempfile.TemporaryDirectory(prefix="ut-python-checks-quiet-") as workdir: root = Path(workdir) write_suite( root, "test_quiet", """ import unittest class Quiet(unittest.TestCase): def test_only(self): pass """, ) completed = run_helper(root, "--slow-seconds", "1000000000") self.assertEqual(completed.returncode, 0, completed.stderr) self.assertNotIn("[slow ", completed.stderr) self.assertIn(" ... ok", completed.stderr) class SourceContractTests(unittest.TestCase): """The Makefile and the shipping manifest keep the helper in use.""" def check_python_recipe(self) -> list[str]: lines = MAKEFILE.read_text(encoding="utf-8").splitlines() for index, line in enumerate(lines): if line.startswith("check-python:"): recipe = [] for step in lines[index + 1 :]: if not step.startswith("\t"): break recipe.append(step.strip()) return recipe raise AssertionError("the Makefile has no check-python target") def test_the_recipe_runs_the_helper_and_keeps_the_document_check(self): recipe = self.check_python_recipe() self.assertIn("python3 tools/run_python_checks.py", recipe) self.assertIn("python3 tools/check_distribution_docs.py", recipe) self.assertFalse( [step for step in recipe if "python3 -m unittest" in step], "the recipe must not run plain unittest discovery as well", ) def test_the_helper_adds_no_failfast_retry_or_environment_input(self): source = HELPER.read_text(encoding="utf-8") self.assertIn("verbosity=2", source) self.assertIn("discover(", source) self.assertNotIn("failfast", source) self.assertNotIn("os.environ", source) self.assertNotIn("getenv", source) self.assertNotIn("time.sleep", source) def test_the_helper_ships_in_the_source_manifest(self): manifest = json.loads(MANIFEST.read_text(encoding="utf-8")) include = { entry["path"] for entry in manifest.get("source_paths", {}).get("include", []) } self.assertIn("tools/run_python_checks.py", include) # The suite the helper discovers ships as a whole directory. self.assertIn("tests/", include) if __name__ == "__main__": unittest.main()