#!/usr/bin/env python3 """check_coverage.py -- a lane module no collected cell reads is invisible (Q85). THE DEFECT THIS GRADES `beam/CHECK` runs every `test_*.py` in this lane through `run_lane_tests.py`. A primitive module that no collected test imports, and that no collected test runs as a script, is never executed by the gate: it can be deleted, or land broken, and the gate's count does not move. Q85 measured the shape three times over (`083981e` added `zeller_dow.py`, 38 lines, and `CHECK` read `Ran 214` before and after). Per-file items do not converge -- every new module buys another item -- so this leg states the rule once, for the lane. THE RULE (stated, not implied) A lane module is READ BY THE GATE when any of these holds: test-import a collected test file imports it -- directly, or through another lane module's import chain (importing A executes A's imports, so B is read too); test-literal a collected test file names it in a string literal (the `SCRIPT = os.path.join(HERE, "shadow_lint.py")` and `from csv_rfc4180 import ...` cases: a module used as a subprocess, or under a different test file's name); gate-invoked `CHECK` itself runs it (`python3 .py`), i.e. the gate executes it on every tick. "No file named `test_.py`" is NOT the fault -- `csv_rfc4180` has no such file and is covered. "Read by nothing the gate runs" is. Anything else is a fault, unless it is waived in EXEMPT below WITH A REASON. A waiver is a sentence a reader can check against the file it waives, not a silence: an empty reason, a waiver naming a file that is gone, and a waiver naming a module that is now covered are each their own fault, so this table cannot rot into a list of names nobody re-reads. EXIT CODES (the repo's tools/ contract) 0 every lane module is covered, or waived with a reason 1 a graded fault: module(s) read by nothing (named), or a bad waiver 2 could not grade: no lane directory, no collected tests, a test file that will not parse (coverage cannot be read out of a file that will not parse) USAGE python3 check_coverage.py [--lane DIR] [--json OUT] [--list] [--selftest] """ import argparse import ast import glob import json import os import re import sys DEFAULT_LANE = os.path.dirname(os.path.abspath(__file__)) # Modules read by nothing the gate runs, each with the reason. Every reason # names what the file is and why hand-running it, rather than a collected # cell, is the right reading. Keep this list short: it is a record of # exceptions, not a coverage plan. EXEMPT = { "verify_toposort_none.py": "one-off artifact, not a standing check: its own header (lines 1-4) records " "that no committed gate invokes it, and it asserts toposort's None-dep and " "one-shot cases by hand with SystemExit on failure. Its subject, toposort.py, " "is covered by test_toposort.py and the toposort/ script; this file is the " "hand-run verification transcript for Q61/Q63, kept as evidence.", "verify_toposort_gen.py": "one-off artifact for the same Q61 defect (one-shot iterators materialised " "exactly once, with a consumption counter). Hand-run against toposort.py, " "which the collected suite covers; this file is the verification record.", "verify_wordcount_test.py": "one-off artifact for wordcount.py, which is not a module of this lane " "(it writes /root/troupe-standup/test_verification.txt from a __main__ and " "compares a subprocess run against `wc`). It is a host-path transcript, not " "a primitive; nothing in the lane imports it.", "CHECK_toposort.py": "hand-run check for toposort_kahn.py, named CHECK_ so the collector does not " "take it (it is not unittest or bare-test shaped for the runner either: its " "three graphs are `def test_dag` style cells with no runner). Its subject is " "covered from this commit by test_toposort_kahn.py, which ports these three " "graphs and adds the cycle cases; kept as the original reading.", "_run_ng.py": "five-line hand runner: it imports test_next_greater and calls every test_* " "in it, printing PASS per cell. Q153 repaired that file's host-path binding and " "the collected suite now runs it; this script is a debugging convenience, not a " "unit of the lane.", "conftest.py": "pytest collection hook (two lines: put the lane on sys.path), read by the " "collector rather than imported by a test. The lane's gate does not use pytest " "at all (Q128), so no cell can import it.", } def module_stem(path): return os.path.splitext(os.path.basename(path))[0] def _parse(path): """ast.parse one file, or None when it will not parse.""" try: with open(path, "r", encoding="utf-8", errors="replace") as fh: return ast.parse(fh.read(), filename=path) except (OSError, SyntaxError): return None def imported_names(tree): """First components of every import in one parsed file.""" names = set() for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: names.add(alias.name.split(".")[0]) elif isinstance(node, ast.ImportFrom): if node.level == 0 and node.module: names.add(node.module.split(".")[0]) return names def string_literals(tree): """Every string constant in one parsed file.""" return { node.value for node in ast.walk(tree) if isinstance(node, ast.Constant) and isinstance(node.value, str) } def grade(lane, exempt=None): """Grade one lane directory. Returns (rc, rows, notes). rows: list of dicts {module, state, why} -- state is one of covered-test-import, covered-transitive, covered-literal, covered-gate, exempt, UNCOVERED, WAIVER-MISSING, WAIVER-STALE, WAIVER-NO-REASON. """ exempt = dict(EXEMPT if exempt is None else exempt) # Waivers are keyed by filename ("probe.py"); a bare stem is accepted and # normalised here so the table and the census speak the same keys. waivers = {} for key, reason in exempt.items(): fname = key if key.endswith(".py") else key + ".py" waivers[fname] = reason notes = [] rows = [] if not os.path.isdir(lane): return 2, rows, ["no lane directory: %s" % lane] lane_py = sorted(glob.glob(os.path.join(lane, "*.py"))) tests = [p for p in lane_py if os.path.basename(p).startswith("test_")] modules = [p for p in lane_py if not os.path.basename(p).startswith("test_")] if not tests: return 2, rows, ["no collected tests (test_*.py) in %s: nothing can be read" % lane] if not modules: return 2, rows, ["no lane modules (*.py minus test_*) in %s: nothing to grade" % lane] lane_names = {module_stem(p) for p in modules} files = {module_stem(p): p for p in modules} # --- parse the collected tests; a broken test file is ungradable ------- test_trees = {} for t in tests: tree = _parse(t) if tree is None: return 2, rows, ["collected test file will not parse: %s" % os.path.basename(t)] test_trees[t] = tree # --- direct imports and literals from the collected tests -------------- seeds, seed_why, literals = set(), {}, set() for t, tree in test_trees.items(): for name in imported_names(tree) & lane_names: seeds.add(name) seed_why.setdefault(name, os.path.basename(t)) literals |= string_literals(tree) covered, why = {}, {} def cover(name, state, reason): if name not in covered: covered[name] = state why[name] = reason for name in sorted(seeds): cover(name, "covered-test-import", "imported by %s" % seed_why[name]) for name in sorted(lane_names): if name in covered: continue # A literal counts only when it IS a path to the file: the bare # filename (`"shadow_lint.py"`) or a path ending in it # (`os.path.join(HERE, "shadow_lint.py")`, `"/root/.../sha256_manifest.py"`). # A mention inside prose -- a docstring saying "test_sha256_manifest.py # grades ..." -- is not a read, and counting prose is exactly the # substring-grep defect Q85's own census had (it reported four modules # invisible while a sixth was read by a path constant only). def _is_path_literal(lit): return (os.path.basename(lit.strip()) == name + ".py" and not any(ch.isspace() for ch in lit)) if any(_is_path_literal(lit) for lit in literals): cover(name, "covered-literal", "the test names the file's path in a literal") # --- transitive closure over the lane's own import graph --------------- edges = {} for name, path in files.items(): tree = _parse(path) edges[name] = (imported_names(tree) if tree is not None else set()) & lane_names if tree is None: notes.append("lane module will not parse (its imports are unread): %s.py" % name) queue = [n for n in covered if covered[n] == "covered-test-import"] while queue: cur = queue.pop() for nxt in sorted(edges.get(cur, ())): if nxt not in covered: cover(nxt, "covered-transitive", "imported through %s.py" % cur) queue.append(nxt) # --- what the gate itself runs ---------------------------------------- check_path = os.path.join(lane, "CHECK") gate_text = "" if os.path.isfile(check_path): with open(check_path, "r", encoding="utf-8", errors="replace") as fh: gate_text = fh.read() else: notes.append("no CHECK in %s: gate-invoked coverage is unread" % lane) for name in sorted(lane_names): if name in covered: continue if re.search(r"python3\s+(?:\./)?%s\.py\b" % re.escape(name), gate_text): cover(name, "covered-gate", "run by CHECK (python3 %s.py)" % name) # --- every module accounted for --------------------------------------- for name in sorted(lane_names): if name in covered: rows.append({"module": name, "state": covered[name], "why": why[name]}) elif (name + ".py") in waivers: reason = (waivers[name + ".py"] or "").strip() if not reason: rows.append({"module": name, "state": "WAIVER-NO-REASON", "why": "waived in EXEMPT with no reason: a waiver needs a sentence"}) else: rows.append({"module": name, "state": "exempt", "why": reason}) else: rows.append({"module": name, "state": "UNCOVERED", "why": "imported by no collected test, named by no literal, run by no CHECK step"}) # --- the waiver table is itself graded -------------------------------- for fname in sorted(waivers): stem = module_stem(fname) if os.path.basename(fname) != fname: rows.append({"module": fname, "state": "WAIVER-MISSING", "why": "EXEMPT names a path, not a lane module: %r" % fname}) elif fname not in {os.path.basename(p) for p in modules}: rows.append({"module": stem, "state": "WAIVER-MISSING", "why": "EXEMPT waives %r, which is not a module in this lane" % fname}) elif stem in covered: rows.append({"module": stem, "state": "WAIVER-STALE", "why": "%s is now covered (%s): delete this waiver" % (fname, covered[stem])}) faults = [r for r in rows if r["state"] in ("UNCOVERED", "WAIVER-MISSING", "WAIVER-STALE", "WAIVER-NO-REASON")] return (1 if faults else 0), rows, notes def report(lane, rc, rows, notes, out=sys.stdout): states = {} for r in rows: states[r["state"]] = states.get(r["state"], 0) + 1 out.write("coverage: %s\n" % lane) for r in rows: if r["state"] not in ("covered-test-import", "covered-transitive", "covered-gate"): out.write(" %-14s %-26s %s\n" % (r["state"], r["module"], r["why"])) for note in notes: out.write(" note %s\n" % note) out.write( "coverage: %d module(s): %d covered by a collected test, %d through an import " "chain, %d by literal, %d run by CHECK, %d exempt, %d fault(s)\n" % (len(rows), states.get("covered-test-import", 0), states.get("covered-transitive", 0), states.get("covered-literal", 0), states.get("covered-gate", 0), states.get("exempt", 0), sum(states.get(s, 0) for s in ("UNCOVERED", "WAIVER-MISSING", "WAIVER-STALE", "WAIVER-NO-REASON")))) out.write("coverage: %s\n" % {0: "PASS", 1: "FAULT", 2: "UNGRADABLE"}[rc]) # -------------------------------------------------------------------------- # selftest: fixture lanes built in a throwaway directory, each one cell. # -------------------------------------------------------------------------- def _fixture(root, name, files, exempt=None, check_text=None): """Build one fixture lane; returns (lane_dir, exempt_dict).""" lane = os.path.join(root, name) os.makedirs(lane, exist_ok=True) for rel, body in files.items(): path = os.path.join(lane, rel) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as fh: fh.write(body) if check_text is not None: with open(os.path.join(lane, "CHECK"), "w", encoding="utf-8") as fh: fh.write(check_text) return lane, (exempt or {}) def selftest(tmp): cases = [] def cell(name, lane, exempt, want_rc, want_states, want_named=None): rc, rows, notes = grade(lane, exempt) got = {r["module"]: r["state"] for r in rows} ok = rc == want_rc detail = "rc=%d want %d" % (rc, want_rc) for mod, state in want_states.items(): if got.get(mod) != state: ok = False detail += "; %s=%r want %r" % (mod, got.get(mod), state) if want_named: text = "\n".join("%s %s" % (r["state"], r["module"]) for r in rows) for needle in want_named: if needle not in text: ok = False detail += "; %r not in the report" % needle cases.append({"case": name, "ok": ok, "detail": detail}) return rc, rows, notes cover_test = "from helper import value\nassert value() == 1\n" chain_test = "from first import one\nassert one() == 1\n" literal_test = 'import os, subprocess\nSCRIPT = os.path.join(os.path.dirname(__file__), "helper.py")\n' # 1 a collected test imports the module lane, ex = _fixture(tmp, "import-covered", { "helper.py": "def value():\n return 1\n", "test_helper.py": cover_test, }) cell("import-covered", lane, ex, 0, {"helper": "covered-test-import"}) # 2 the import chain covers the second module lane, ex = _fixture(tmp, "transitive-covered", { "first.py": "from second import two\ndef one():\n return two()\n", "second.py": "def two():\n return 1\n", "test_first.py": chain_test, }) cell("transitive-covered", lane, ex, 0, {"first": "covered-test-import", "second": "covered-transitive"}) # 3 a filename literal covers a module the test runs, not imports lane, ex = _fixture(tmp, "literal-covered", { "helper.py": "print('hi')\n", "test_run.py": literal_test, }) cell("literal-covered", lane, ex, 0, {"helper": "covered-literal"}) # 4 nothing reads it -> red, and the report names it lane, ex = _fixture(tmp, "uncovered", { "orphan.py": "def f():\n return 1\n", "test_other.py": "def test_x():\n assert True\n", }) cell("uncovered-is-red", lane, ex, 1, {"orphan": "UNCOVERED"}, want_named=["UNCOVERED orphan"]) # 5 a waiver with a reason is a pass, and the reason is printed lane, ex = _fixture(tmp, "exempt-reason", { "probe.py": "print('hand run')\n", "test_other.py": "def test_x():\n assert True\n", }, exempt={"probe.py": "one-off probe, hand-run against a subject the suite covers"}) rc, rows, _ = cell("exempt-with-reason", lane, ex, 0, {"probe": "exempt"}) if not any("one-off probe" in r["why"] for r in rows): cases[-1]["ok"] = False cases[-1]["detail"] += "; reason not printed" # 6 a waiver without a reason is a fault, not a silence lane, ex = _fixture(tmp, "exempt-silent", { "probe.py": "print('hand run')\n", "test_other.py": "def test_x():\n assert True\n", }, exempt={"probe.py": " "}) cell("waiver-needs-a-reason", lane, ex, 1, {"probe": "WAIVER-NO-REASON"}) # 7 a waiver naming a file that is gone is a fault lane, ex = _fixture(tmp, "exempt-missing", { "kept.py": "def f():\n return 1\n", "test_kept.py": "from kept import f\nassert f() == 1\n", }, exempt={"deleted_last_year.py": "was a probe"}) cell("waiver-names-a-missing-file", lane, ex, 1, {"deleted_last_year": "WAIVER-MISSING"}) # 8 a waiver left standing over a now-covered module is a fault lane, ex = _fixture(tmp, "exempt-stale", { "helper.py": "def value():\n return 1\n", "test_helper.py": cover_test, }, exempt={"helper.py": "used to be uncovered"}) cell("waiver-now-covered", lane, ex, 1, {"helper": "WAIVER-STALE"}) # 9 a module CHECK runs is covered by the gate itself lane, ex = _fixture(tmp, "gate-invoked", { "gate_script.py": "print('gate')\n", "test_other.py": "def test_x():\n assert True\n", }, check_text="#!/bin/bash\npython3 gate_script.py\n") cell("gate-invoked", lane, ex, 0, {"gate_script": "covered-gate"}) # 10 no collected tests -> cannot grade, rather than "all covered" lane, ex = _fixture(tmp, "no-tests", {"orphan.py": "x = 1\n"}) cell("no-tests-is-ungradable", lane, ex, 2, {}) # 11 a test file that will not parse -> cannot grade lane, ex = _fixture(tmp, "bad-test", { "orphan.py": "x = 1\n", "test_broken.py": "def test_x(:\n pass\n", }) cell("unparseable-test-is-ungradable", lane, ex, 2, {}) # 12 the falsifiability cell: delete the one test file, the fixture reds lane, ex = _fixture(tmp, "mutation", { "helper.py": "def value():\n return 1\n", "test_helper.py": cover_test, }) first = grade(lane, ex)[0] os.remove(os.path.join(lane, "test_helper.py")) with open(os.path.join(lane, "test_other.py"), "w", encoding="utf-8") as fh: fh.write("def test_x():\n assert True\n") second = grade(lane, ex)[0] ok = first == 0 and second == 1 cases.append({"case": "removing-the-cell-reds-it", "ok": ok, "detail": "rc %d -> %d (want 0 -> 1)" % (first, second)}) # 13 no lane directory cases.append({"case": "missing-lane-is-ungradable", "ok": grade(os.path.join(tmp, "does-not-exist"), {})[0] == 2, "detail": "rc=%d" % grade(os.path.join(tmp, "does-not-exist"), {})[0]}) bad = [c for c in cases if not c["ok"]] for c in cases: print("%-4s %-32s %s" % ("ok" if c["ok"] else "FAIL", c["case"], c["detail"])) print("selftest: %d cell(s), %d failed" % (len(cases), len(bad))) return 1 if bad else 0 def main(argv=None): ap = argparse.ArgumentParser(description="grade lane-module coverage for one lane") ap.add_argument("--lane", default=DEFAULT_LANE) ap.add_argument("--json", default=None, help="write the row table here") ap.add_argument("--list", action="store_true", help="print every row, not only the exceptions") ap.add_argument("--selftest", action="store_true") args = ap.parse_args(argv) if args.selftest: import tempfile with tempfile.TemporaryDirectory(prefix="coverage-selftest.") as tmp: return selftest(tmp) rc, rows, notes = grade(args.lane) if args.list: for r in rows: print("%-20s %-26s %s" % (r["state"], r["module"], r["why"])) report(args.lane, rc, rows, notes) if args.json: with open(args.json, "w", encoding="utf-8") as fh: json.dump({"lane": args.lane, "rc": rc, "rows": rows, "notes": notes}, fh, indent=1) return rc if __name__ == "__main__": sys.exit(main())