"""Invariants of the code-health ratchet (scripts/code_health), on a real git repo + pinned ruff.""" from __future__ import annotations import json import shutil import subprocess from pathlib import Path import pytest from scripts.code_health import gitio, replay from scripts.code_health.cli import run from scripts.code_health.compare import compare from scripts.code_health.measure import Measurer from scripts.code_health.report import apply_allows from scripts.code_health.ruff_runner import resolve_ruff REPO = Path(__file__).resolve().parents[2] _LEGACY = "def legacy(x):\n" + "".join(f" if x == {i}:\n return {i}\n" for i in range(21)) _SWALLOW = "def other():\n try:\n pass\n except Exception:\n pass\n" def _git(repo: Path, *args: str) -> str: return subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True, text=True, encoding="utf-8", timeout=60).stdout.strip() def _commit(repo: Path, files: dict[str, str | None]) -> str: for rel, text in files.items(): path = repo / rel if text is None: path.unlink() else: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") _git(repo, "add", "-A") _git(repo, "commit", "-q", "-m", "step") return _git(repo, "rev-parse", "HEAD") def _repo(tmp_path: Path) -> tuple[Path, str]: repo = tmp_path / "repo" (repo / "scripts" / "ci").mkdir(parents=True) shutil.copy(REPO / "pyproject.toml", repo / "pyproject.toml") shutil.copy(REPO / "scripts/ci/profile_scope_patterns.json", repo / "scripts/ci/") _git(repo, "init", "-q", "-b", "main") _git(repo, "config", "user.email", "t@example.com") _git(repo, "config", "user.name", "t") return repo, _commit(repo, {"pkg/a.py": _LEGACY + "\n\n" + _SWALLOW}) def _verdict(repo: Path, base: str, files: dict[str, str | None], capsys) -> tuple[int, str]: head = _commit(repo, files) code = run(repo, base, head) out = capsys.readouterr().out _git(repo, "reset", "-q", "--hard", base) return code, out def test_ratchet_blocks_growth_new_and_swapped_violations(tmp_path, capsys): repo, base = _repo(tmp_path) legacy_plus = _LEGACY + " if x == 99:\n return 99\n" ok = _LEGACY + "\n\n" + _SWALLOW + "\n\ndef added():\n return 1\n" grown = legacy_plus + "\n\n" + _SWALLOW fresh = _SWALLOW.replace("pass\n", "return 2\n", 1).replace("other", "fresh") swapped = _LEGACY + "\n\ndef other():\n return 1\n\n\n" + fresh assert _verdict(repo, base, {"pkg/a.py": ok}, capsys)[0] == 0 # legacy debt never blocks code, out = _verdict(repo, base, {"pkg/a.py": grown}, capsys) assert code == 1 and "legacy" in out and "CC 23 > 22" in out code, out = _verdict(repo, base, {"pkg/a.py": swapped}, capsys) assert code == 1 and "BLE001" in out and "fresh" in out # fixing one doesn't buy another traded = {"pkg/a.py": _LEGACY, "pkg/b.py": fresh} code, out = _verdict(repo, base, traded, capsys) assert code == 1 and "fresh" in out # deleting a violation never pays for an unrelated one def test_enforcement_switch_is_read_from_the_base(tmp_path, capsys): repo, base = _repo(tmp_path) grown = _LEGACY + " if x == 99:\n return 99\n\n\n" + _SWALLOW switch = "scripts/code_health/config.py" # A PR cannot relax its own check: the switch it commits is not the one that applies. assert _verdict(repo, base, {"pkg/a.py": grown, switch: 'ENFORCEMENT = "off"\n'}, capsys)[0] == 1 for mode, expected in (("advisory", 0), ("off", 0), ("blocking", 1)): main_tip = _commit(repo, {switch: f'ENFORCEMENT = "{mode}"\n'}) code, out = _verdict(repo, main_tip, {"pkg/a.py": grown}, capsys) assert code == expected, (mode, out) assert ("CC 23 > 22" in out) == (mode != "off"), (mode, out) def test_moved_code_keeps_its_cap(tmp_path, capsys): repo, base = _repo(tmp_path) for split in ({"pkg/a.py": _SWALLOW, "pkg/a_legacy.py": _LEGACY}, {"pkg/a.py": _LEGACY, "pkg/a_swallow.py": _SWALLOW}): # hits move with their unit code, out = _verdict(repo, base, dict(split), capsys) assert code == 0, out renamed: dict[str, str | None] = {"pkg/a.py": None, "pkg/b.py": _LEGACY + "\n\n" + _SWALLOW} code, out = _verdict(repo, base, renamed, capsys) assert code == 0, out _RUN = "import subprocess\n\n\ndef f(cmd):\n return subprocess.run(cmd{})\n" _PROC = "import asyncio\n\n\nasync def f(cmd):\n proc = await asyncio.create_subprocess_exec(*cmd)\n{}\n" _TWO_RUNS = "import subprocess\n\n\ndef f(a, b):\n subprocess.run(a){}\n subprocess.run(b){}\n" _ENV = "import os\n\n{}\n" _EXCEPT = " try:\n pass\n except Exception:\n pass\n" _STUB = {"pkg/b.py": "def legacy(x):\n return x\n"} @pytest.mark.parametrize("extra_base, files, blocks", [ # one base unit is credit for one head unit: a copy of unchanged debt is new debt ({}, {"pkg/a.py": _LEGACY + "\n\n" + _SWALLOW + "\n\n" + _LEGACY.replace("legacy", "copied")}, True), # a file rename re-keys each old hit once, so a second swallow in that function is new ({}, {"pkg/a.py": None, "pkg/b.py": _LEGACY + "\n\n" + _SWALLOW + _EXCEPT}, True), # a function moved over a same-name stub keeps its own cap, not the stub's (_STUB, {"pkg/a.py": _SWALLOW, "pkg/b.py": _LEGACY}, False), ({}, {"pkg/p.py": _RUN.format(", timeout=None")}, True), # a disabled deadline is no deadline ({}, {"pkg/p.py": _RUN.format(", **{}")}, True), ({}, {"pkg/p.py": _PROC.format(" return await asyncio.wait_for(proc.communicate(), None)")}, True), ({}, {"pkg/p.py": _PROC.format(" async with asyncio.timeout(1):\n return await proc.communicate()")}, False), # an allow directive inside a string literal waives nothing ({}, {"pkg/p.py": _RUN.format("").replace(" return", " print('health: allow HX006 -- doc')\n return")}, True), ({}, {"pkg/c.py": _ENV.format("if __name__ != '__main__':\n CACHED = os.getenv('PATH')")}, True), ({}, {"pkg/c.py": _ENV.format("for _ in range(1):\n CACHED = os.getenv('PATH')")}, True), ({}, {"pkg/c.py": _ENV.format("current = lambda: os.getenv('PATH')")}, False), # deferred read ({}, {"pkg/c.py": _ENV.format("current = lambda p=os.getenv('PATH'): p")}, True), # eager default ({}, {"pkg/c.py": _ENV.format("with open(os.getenv('PATH', 'x'), encoding='utf-8') as fh:\n pass")}, True), ({}, {"pkg/c.py": _ENV.format("current = (os.getenv('PATH') for _ in range(1))")}, False), # a coroutine defined inside the deadline runs after it, so it is not bounded by it ({}, {"pkg/p.py": _PROC.format(" async with asyncio.timeout(1):\n async def later():\n" " return await proc.communicate()\n return later")}, True), ({}, {"pkg/p.py": _PROC.format(" return await asyncio.wait_for(fut=proc.communicate(), timeout=1)")}, False), # a BOM reads the same from git and from disk, so a new BOM file's debt is still measured ({}, {"pkg/bom.py": "\ufeff" + _LEGACY}, True), # after kill(), a sync wait() is bounded by SIGKILL; communicate() and asyncio's wait() block # while a grandchild holds the pipe (measured), so those stay flagged *(({}, {"pkg/k.py": "import subprocess\n\n\ndef f(c):\n proc = subprocess.Popen(c)\n" f" try:\n proc.wait(timeout=1)\n except subprocess.TimeoutExpired:\n" f" proc.kill()\n {tail}\n"}, blocked) for tail, blocked in (("proc.wait()", False), ("proc.communicate()", True))), ({}, {"pkg/k.py": _PROC.format(" proc.kill()\n await proc.wait()")}, True), # rules see through import aliases and require the real asyncio deadline API ({}, {"pkg/c.py": "from subprocess import run as execute\n\n\ndef f(c):\n return execute(c)\n"}, True), ({}, {"pkg/c.py": "def run(c):\n return c\n\n\ndef f(c):\n return run(c)\n"}, False), ({}, {"pkg/p.py": "async def wait_for(aw, t):\n return await aw\n\n\n" + _PROC.format(" return await wait_for(proc.communicate(), 1)")}, True), ({}, {"pkg/c.py": "def f(client):\n return client.communicate()\n"}, False), ({}, {"pkg/p.py": _PROC.format(" return await asyncio.wait_for(proc.wait(), float('inf'))")}, True), # writes introduce a HERMES_* name too; bool() of a raw env string is HX010 ({}, {"pkg/c.py": _ENV.format("os.environ['HERMES_BRAND_NEW'] = '1'")}, True), ({}, {"pkg/c.py": _ENV.format("FLAG = bool(os.environ['PATH'])")}, True), # an inline allow covers its own line only; removing an allow keeps the debt existing ({}, {"pkg/p.py": _TWO_RUNS.format(" # health: allow HX006 -- x", "")}, True), ({"pkg/p.py": _TWO_RUNS.format(" # health: allow HX006 -- x", " # health: allow HX006 -- y")}, {"pkg/p.py": _TWO_RUNS.format("", "")}, False), # an identical violation moved past unrelated code is a new occurrence ({"pkg/a.py": _LEGACY + "\n\n" + _SWALLOW.replace("pass\n except", "a()\n except") + " b()\n c()\n d()\n"}, {"pkg/a.py": _LEGACY + "\n\n" + "def other():\n a()\n b()\n c()\n d()\n" " try:\n pass\n except Exception:\n pass\n"}, True), # a recursive function renamed with its self-calls keeps its cap ({}, {"pkg/a.py": (_LEGACY + " return legacy(x - 1)\n").replace("legacy", "walk") + "\n\n" + _SWALLOW}, True), ({"pkg/a.py": _LEGACY + " return legacy(x - 1)\n\n\n" + _SWALLOW}, {"pkg/a.py": (_LEGACY + " return legacy(x - 1)\n").replace("legacy", "walk") + "\n\n" + _SWALLOW}, False), # an over-cap file may not grow, and an unparseable file never passes ({"pkg/big.py": "V = 0\n" * 2001}, {"pkg/big.py": "V = 0\n" * 2002}, True), ({}, {"pkg/c.py": "def f(:\n pass\n"}, True), ]) def test_verdicts_follow_ownership_deadlines_and_import_execution(tmp_path, capsys, extra_base, files, blocks): repo, base = _repo(tmp_path) if extra_base: base = _commit(repo, extra_base) code, out = _verdict(repo, base, files, capsys) assert code == (1 if blocks else 0), out # --- replay: a range it could not measure is reported and fails the run, never counted clean --- def test_replay_reports_unmeasured_prs_and_fails(tmp_path, monkeypatch, capsys): repo, base = _repo(tmp_path) head = _commit(repo, {"pkg/b.py": _SWALLOW.replace("other", "fresh")}) def pr(number: int, oid: str) -> dict: return {"number": number, "title": f"pr {number}", "mergeCommit": {"oid": oid}, "commits": {"totalCount": 1, "nodes": [{"commit": {"messageHeadline": "elsewhere"}}]}} # #2's merge commit is not in the clone: building the manifest cannot resolve its range monkeypatch.setattr(replay, "merged_prs", lambda *_: [pr(1, head), pr(2, "1" * 40)]) monkeypatch.chdir(repo) out_dir = tmp_path / "out" assert replay.main(["--merged", "2026-09-01..2026-09-30", "--out", str(out_dir)]) == 1 out = capsys.readouterr().out assert "1 of 1 measured PRs had at least one blocking finding" in out, out assert "1 of 2 PRs could not be measured: #2" in out, out # a frozen manifest whose range no longer exists cannot be replayed either manifest = json.loads((out_dir / "manifest.json").read_text(encoding="utf-8")) manifest.append({"number": 3, "title": "pr 3", "base": "0" * 40, "head": "1" * 40}) (tmp_path / "frozen.json").write_text(json.dumps(manifest), encoding="utf-8") assert replay.main(["--manifest", str(tmp_path / "frozen.json"), "--out", str(out_dir)]) == 1 out = capsys.readouterr().out assert "1 of 1 measured PRs had at least one blocking finding" in out, out assert "2 of 3 PRs could not be measured: #2, #3" in out, out # --- a blob's line endings never move a hit, its scope or its waiver --- def _windows_write_text(self, data, encoding=None, errors=None, newline=None): """``Path.write_text`` as on Windows, where ``newline=None`` turns each "\n" into "\r\n".""" if newline is None: data = data.replace("\n", "\r\n") with open(self, "w", encoding=encoding, errors=errors, newline="") as fh: return fh.write(data) def test_crlf_blob_measures_like_lf(tmp_path, monkeypatch): waived = _SWALLOW.replace("except Exception:", "except Exception: # health: allow BLE001 S110 -- boundary") results = {} for name, eol in (("lf", "\n"), ("crlf", "\r\n")): repo, base = _repo(tmp_path / name) _git(repo, "config", "core.autocrlf", "false") (repo / "pkg/b.py").write_bytes(waived.replace("\n", eol).encode("utf-8")) head = _commit(repo, {}) assert (eol == "\r\n") == (b"\r\n" in subprocess.run( ["git", "show", f"{head}:pkg/b.py"], cwd=repo, capture_output=True, timeout=60, check=True).stdout) with monkeypatch.context() as patch: patch.setattr(Path, "write_text", _windows_write_text) measurer = Measurer(repo, resolve_ruff(repo), known_env=set()) base_m, head_m = measurer.measure(base, []), measurer.measure(head, ["pkg/b.py"]) findings = compare(base_m, head_m, gitio.changed_files(repo, base, head)) apply_allows(findings, head_m) results[name] = sorted((f.rule, f.scope, f.line, f.allowed_reason) for f in findings) assert results["lf"] == [("BLE001", "other", 4, "boundary"), ("S110", "other", 4, "boundary")] assert results["crlf"] == results["lf"]