Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
28 kB · 740 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741#!/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, nonetwork, and no hardware. They exercise ``tools/run_python_checks.py`` throughreal child processes against small temporary suites, because the property undertest is about *when* bytes reach the output stream, which an in-process resultdouble 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 suitefit inside any particular output cap. The helper bounds what it *adds*; a suitewhose failures alone exceed a caller's cap can still be truncated."""
from __future__ import annotations
import jsonimport queueimport reimport subprocessimport sysimport tempfileimport textwrapimport threadingimport timeimport unittestfrom 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.0SUBPROCESS_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()