diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 01bc2d96..40ad08a7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -77,7 +77,9 @@ jobs: uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: coverage-report - path: implementations/python/coverage.xml + path: | + implementations/python/coverage.xml + implementations/python/coverage.json fuzz: runs-on: ubuntu-latest diff --git a/docs/decisions/issue-1104-asr-505-coverage-ratchet-preflight.md b/docs/decisions/issue-1104-asr-505-coverage-ratchet-preflight.md new file mode 100644 index 00000000..36370cd8 --- /dev/null +++ b/docs/decisions/issue-1104-asr-505-coverage-ratchet-preflight.md @@ -0,0 +1,88 @@ +# Issue 1104 Coverage Ratchet Preflight + +Date: 2026-08-11 + +Issue: #1104. Requirement: ASR-505. + +## Decision + +Canonical verification measures branch as well as statement coverage. Every +executable production or repository-tooling line added or modified relative to +the exact verified base revision must execute, and every branch whose source is +added or modified must take all of its exits. A changed measured Python file +that is absent from coverage data is a failure rather than an implicit skip. + +The changed-code rule is 100%. The legacy aggregate is not: current OpenRAE has +thousands of uncovered statements and no historical branch data. The repository +therefore records an honest whole-tree baseline and enforces it as a monotonic +ratchet. Raising that baseline is welcome; lowering it or adding exclusions to +hide difficult code is not. A later issue may set 100% aggregate only after the +measured corpus actually reaches it. + +When the ratchet already exists at the exact base revision, the gate also reads +that historical file and rejects any lower line or branch floor. Editing the +checked-in threshold downward therefore fails in the same job that evaluates +the coverage report. + +The initial canonical full unit-plus-integration inventory on CPython 3.12.3 +and Linux x86_64 contains 82,536 executable statements and 25,836 branches. It +covers 74,067 statements (89.739023%) and 19,542 branches (75.638644%); the +checked-in floors round down to 89.739% and 75.638%. Subsequent updates may only +raise them. + +Measured scope includes the twelve shipped package roots, repository-owned +`tools`, `noxfile.py`, and the Hatch build hook. Tests, generated environments, +acquired tool caches, and external dependencies are not production denominator +padding. The checked-in Coverage.py configuration is itself validated: source +roots and non-source omissions are fixed, same-checkout absolute data paths are +required, path aliases are forbidden, and no custom partial-branch or exclusion +rule may suppress the changed-code gate. Absolute data paths let Coverage.py +emit one repository source and repository-relative XML filenames for project, +tooling, and Nox code without merging aliases across machines. Existing explicit +exclusions retain their separately documented integration rationale. An inline +coverage pragma is rejected whenever its semantic +statement owns changed code, even when the pragma's physical line is unchanged. +Coverage.py's structural `TYPE_CHECKING` and `Protocol` exclusions are accepted +only for canonical, unshadowed imports from `typing`; runtime-evaluated defaults, +class bases, and class/function decorators remain coverage obligations. Unsafe +namespace access, escaped namespace factories, dynamic execution or attribute +mutation, and wildcard imports invalidate that trust. Every annotation +expression span is an obligation when Python evaluates annotations eagerly; the +same syntax remains structural when `from __future__ import annotations` +postpones it. + +## Diff And Branch Semantics + +The gate consumes Coverage.py JSON produced from the combined unit and +integration data and a Git commit accepted by `git merge-base --is-ancestor`. +It parses zero-context added-line ranges from that exact base through `HEAD`. +Comments and other non-executable changed lines are ignored. An executable +changed line is covered only when Coverage.py reports it executed. Physical +changes inside a multiline expression are mapped to the owning Coverage.py +statement; multiline branch headers are also mapped to their branch source. +This prevents a continuation-line edit from disappearing between Git's physical +line model and Coverage.py's executable-line model. A deletion-only hunk that +removes semantic content is anchored to its surviving destination neighbors, +so removing one line from a multiline condition cannot produce an empty change +set. When a changed executable line is a branch source, no missing destination +from that source is allowed. + +Files outside the measured OpenRAE source/tool roots do not enter the changed +coverage gate. New and renamed files use their destination path; a rename is +checked conservatively as a new destination file. Deleted files have no +remaining executable obligation. The aggregate ratchet stays at the canonical +`tools/coverage_ratchet.json` path. Once that path has appeared in base history, +its deletion or relocation is an error rather than a new first adoption. + +## Verification + +Unit tests cover diff ranges, new and renamed files, path normalization, +comments, multiline statements and branch headers, missing source records, +missing lines, missing branches, configuration suppression attempts, ratchet +relocation/deletion, acquired-cache isolation, branch-data absence, and invalid +base revisions. Canonical verification must produce distinct, non-empty unit +and integration data files before combining them, publish XML and JSON, enforce +the aggregate ratchet, and run the changed-code gate with the same base SHA used +by repository policy. Both report commands receive canonical absolute output +paths under `implementations/python`, so Coverage.py configuration cannot +redirect either report away from the artifact paths uploaded by CI. diff --git a/docs/requirements/ASR-505/requirement.md b/docs/requirements/ASR-505/requirement.md index e41342cc..5f1794f7 100644 --- a/docs/requirements/ASR-505/requirement.md +++ b/docs/requirements/ASR-505/requirement.md @@ -35,3 +35,9 @@ Requirement inventory phase. Status audit deferred until the full canonical grap - IMPLEMENTS → SPEC `specs/formal/assurance-fulfillment.yaml` (Per-subsystem assurance fulfillment map (delivered/waived artifacts per classified formal domain)) - TESTS → TEST `implementations/python/tests/test_participant_runtime_invariants.py` (Participant runtime invariant oracle property tests) - DOCUMENTS → SPEC `specs/formal/participant-runtime/README.md` (Participant runtime invariant oracle predicate mapping) +- DOCUMENTS → GITHUB_ISSUE `1104` (Branch and changed-code coverage ratchet) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1104-asr-505-coverage-ratchet-preflight.md` (Coverage scope, diff semantics, and legacy-debt nonclaim) +- IMPLEMENTS → POLICY `tools/check_changed_coverage.py` (Exact-base 100% changed executable-line and branch gate) +- IMPLEMENTS → CONFIG `implementations/python/pyproject.toml` (Branch-aware measured source configuration) +- IMPLEMENTS → CONFIG `noxfile.py` (Combined coverage ratchet and changed-code enforcement) +- TESTS → TEST `implementations/python/tests/test_changed_coverage.py` (Diff, path, missing-line, and missing-branch policy tests) diff --git a/implementations/python/pyproject.toml b/implementations/python/pyproject.toml index 788f05f5..bd13d5e6 100644 --- a/implementations/python/pyproject.toml +++ b/implementations/python/pyproject.toml @@ -123,29 +123,19 @@ addopts = "-m 'not fuzz and not integration and not docker'" timeout = 120 [tool.coverage.run] -source = [ - "raes_contracts", - "raes_backend_protocols", - "raes_backend_stubs", - "raes_backend_libvirt", - "raes_operations", - "raes_reference_backend", - "raes_cli", - "raes_conformance", - "raes_mcp", - "raes_processor", - "raes_runtime", - "raes", - # Repo CI tooling (issue #54): the `tools/` tree lives at the repo root and - # is imported as the top-level `tools` package via pytest's `pythonpath` - # (`../..`). Coverage matches source packages by module name, so its - # `tools.*` modules are measured even though pytest runs from - # `implementations/python/`. - "tools", +branch = true +relative_files = false +source = ["../.."] +omit = [ + "*/.cache/*", + "*/implementations/python/tests/*", + "*/implementations/python/.venv/*", + "*/docs/*", ] [tool.coverage.report] show_missing = true +include_namespace_packages = true exclude_also = [ # The reference backend's OCI subprocess leaf (RUN-314): real container # IO, exercised only by the opt-in docker integration tests, never in the diff --git a/implementations/python/tests/test_changed_coverage.py b/implementations/python/tests/test_changed_coverage.py new file mode 100644 index 00000000..3933a4db --- /dev/null +++ b/implementations/python/tests/test_changed_coverage.py @@ -0,0 +1,1694 @@ +"""Branch-aware changed-code and aggregate coverage policy tests.""" + +from __future__ import annotations + +import json +import runpy +import shutil +import subprocess +import sys +from copy import deepcopy +from pathlib import Path + +import pytest +from coverage import Coverage +from tools import check_changed_coverage as coverage_policy + +REPO_ROOT = Path(__file__).resolve().parents[3] +PROJECT_CONFIG = REPO_ROOT / "implementations" / "python" / "pyproject.toml" + + +def _git(repo: Path, *args: str) -> str: + return subprocess.run( + ("git", *args), + cwd=repo, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def _repository(tmp_path: Path) -> tuple[Path, str]: + repo = tmp_path / "repo" + source = repo / "implementations" / "python" / "packages" / "demo.py" + source.parent.mkdir(parents=True) + source.write_text("value = 1\n", encoding="utf-8") + (repo / "tools").mkdir() + (repo / "noxfile.py").write_text("VALUE = 1\n", encoding="utf-8") + _git(repo, "init") + _git(repo, "config", "user.email", "coverage@example.invalid") + _git(repo, "config", "user.name", "Coverage Test") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "base") + return repo, _git(repo, "rev-parse", "HEAD") + + +def _report(*, line_covered: int = 1, branch_covered: int = 1) -> dict[str, object]: + return { + "meta": {"branch_coverage": True}, + "files": {}, + "totals": { + "covered_lines": line_covered, + "num_statements": 1, + "covered_branches": branch_covered, + "num_branches": 1, + }, + } + + +def _ratchet(*, line: float = 100.0, branch: float = 100.0) -> dict[str, object]: + return {"minimum_line_percent": line, "minimum_branch_percent": branch} + + +def _install_coverage_config(project: Path) -> Path: + project.mkdir(parents=True, exist_ok=True) + config_path = project / "pyproject.toml" + shutil.copyfile(PROJECT_CONFIG, config_path) + return config_path + + +def _main_paths(repo: Path) -> tuple[Path, Path]: + project = repo / "implementations" / "python" + config_path = _install_coverage_config(project) + ratchet_path = repo / "tools" / "coverage_ratchet.json" + ratchet_path.parent.mkdir(parents=True, exist_ok=True) + return config_path, ratchet_path + + +@pytest.mark.parametrize( + ("path", "expected"), + [ + ("implementations/python/packages/raes/core.py", True), + ("tools/check.py", True), + ("noxfile.py", True), + ("implementations/python/hatch_build.py", True), + ("implementations/python/tests/test_core.py", False), + ("tools/data.json", False), + ], +) +def test_measured_python_path_policy(path: str, expected: bool) -> None: + assert coverage_policy.is_measured_python_path(path) is expected + + +def test_added_lines_from_zero_context_patch() -> None: + patch = """@@ -1 +1,3 @@ +-old ++one ++two ++three +@@ -9,0 +12 @@ ++last +@@ -20,2 +30,0 @@ +-gone +""" + assert coverage_policy.added_lines_from_patch(patch) == {1, 2, 3, 12} + assert coverage_policy.added_lines_from_patch("not a patch") == set() + + +def test_deletion_anchors_ignore_nonsemantic_base_lines() -> None: + patch = """@@ -4 +3,0 @@ +- and second +@@ -8 +6,0 @@ +- # explanation +""" + assert coverage_policy.deletion_anchor_lines_from_patch( + patch, + base_semantic_lines={4}, + current_semantic_lines={1, 2, 3, 4, 5, 6}, + ) == {3, 4} + + +@pytest.mark.parametrize( + ("patch", "current_lines", "expected"), + [ + ("@@ -1 +0,0 @@\n-old\n", {2, 3}, {2}), + ("@@ -9 +8,0 @@\n-old\n", {1, 2}, {2}), + ], +) +def test_deletion_anchors_use_the_available_surviving_neighbor( + patch: str, + current_lines: set[int], + expected: set[int], +) -> None: + assert ( + coverage_policy.deletion_anchor_lines_from_patch( + patch, + base_semantic_lines={1, 9}, + current_semantic_lines=current_lines, + ) + == expected + ) + + +def test_changed_python_lines_reads_committed_and_worktree_changes(tmp_path: Path) -> None: + repo, base = _repository(tmp_path) + source = repo / "implementations" / "python" / "packages" / "demo.py" + source.write_text("value = 1\nadded = 2\n", encoding="utf-8") + (repo / "tools" / "new_check.py").write_text("ENABLED = True\n", encoding="utf-8") + tests = repo / "implementations" / "python" / "tests" + tests.mkdir() + (tests / "test_demo.py").write_text("assert True\n", encoding="utf-8") + + assert coverage_policy.changed_python_lines(repo, base) == { + "implementations/python/packages/demo.py": {2}, + "tools/new_check.py": {1}, + } + + +def test_changed_python_lines_checks_every_line_of_renamed_destinations(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + source = repo / "implementations" / "python" / "packages" / "demo.py" + original = "".join(f"value_{line} = {line}\n" for line in range(1, 6)) + source.write_text(original, encoding="utf-8") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "expand source") + base = _git(repo, "rev-parse", "HEAD") + renamed = source.with_name("renamed.py") + _git(repo, "mv", source.relative_to(repo).as_posix(), renamed.relative_to(repo).as_posix()) + renamed.write_text(f"{original}added = 6\n", encoding="utf-8") + + assert coverage_policy.changed_python_lines(repo, base) == { + "implementations/python/packages/renamed.py": {1, 2, 3, 4, 5, 6} + } + + +def test_changed_python_lines_maps_deletion_only_multiline_semantics(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + source = repo / "implementations" / "python" / "packages" / "demo.py" + source.write_text( + """def choose(first: bool, second: bool) -> int: + if ( + first + and second + ): + return 1 + return 0 +""", + encoding="utf-8", + ) + _git(repo, "add", ".") + _git(repo, "commit", "-m", "add multiline branch") + base = _git(repo, "rev-parse", "HEAD") + source.write_text( + """def choose(first: bool, second: bool) -> int: + if ( + first + ): + return 1 + return 0 +""", + encoding="utf-8", + ) + + path = "implementations/python/packages/demo.py" + changed = coverage_policy.changed_python_lines(repo, base) + assert changed == {path: {3, 4}} + assert coverage_policy.changed_coverage_failures( + changed, + { + path: { + "executed_lines": [1, 2, 3, 5], + "missing_lines": [6], + "executed_branches": [[2, 5]], + "missing_branches": [[2, 6]], + } + }, + repo_root=repo, + ) == [f"{path}:2->6: changed branch exit is not covered"] + + +def test_changed_python_lines_ignores_deletion_only_comments(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + source = repo / "implementations" / "python" / "packages" / "demo.py" + source.write_text("value = (\n # explanation\n 1\n)\n", encoding="utf-8") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "add comment") + base = _git(repo, "rev-parse", "HEAD") + source.write_text("value = (\n 1\n)\n", encoding="utf-8") + + assert coverage_policy.changed_python_lines(repo, base) == {"implementations/python/packages/demo.py": set()} + + +def test_changed_python_lines_rejects_undecodable_current_source(tmp_path: Path) -> None: + repo, base = _repository(tmp_path) + source = repo / "implementations" / "python" / "packages" / "demo.py" + source.write_bytes(b"\xff") + + with pytest.raises(coverage_policy.CoveragePolicyError, match="could not inspect changed Python source"): + coverage_policy.changed_python_lines(repo, base) + + +def test_changed_python_lines_rejects_unknown_and_nonancestor_bases(tmp_path: Path) -> None: + repo, base = _repository(tmp_path) + with pytest.raises(coverage_policy.CoveragePolicyError, match="does not resolve"): + coverage_policy.changed_python_lines(repo, "does-not-exist") + + (repo / "noxfile.py").write_text("VALUE = 2\n", encoding="utf-8") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "second") + descendant = _git(repo, "rev-parse", "HEAD") + _git(repo, "checkout", base) + with pytest.raises(coverage_policy.CoveragePolicyError, match="is not an ancestor"): + coverage_policy.changed_python_lines(repo, descendant) + + +def test_git_failure_without_stderr_uses_stable_fallback(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr( + coverage_policy.subprocess, + "run", + lambda *args, **kwargs: subprocess.CompletedProcess(args=args, returncode=1, stdout=b"", stderr=b""), + ) + with pytest.raises(coverage_policy.CoveragePolicyError, match="git status failed"): + coverage_policy._git(tmp_path, "status") + + +def test_normalized_file_records_resolves_project_repo_and_absolute_paths(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + project = repo / "implementations" / "python" + tool = repo / "tools" / "check.py" + tool.write_text("pass\n", encoding="utf-8") + absolute_package = project / "packages" / "absolute.py" + absolute_package.write_text("pass\n", encoding="utf-8") + absolute_tool = repo / "tools" / "absolute.py" + absolute_tool.write_text("pass\n", encoding="utf-8") + report = { + "files": { + "packages/demo.py": {"executed_lines": [1]}, + "tools/check.py": {"executed_lines": [1]}, + str(absolute_package): {"executed_lines": [1]}, + str(absolute_tool): {"executed_lines": [1]}, + str(repo / "noxfile.py"): {"executed_lines": [1]}, + "tests/test_demo.py": {"executed_lines": [1]}, + } + } + + assert coverage_policy.normalized_file_records(report, repo_root=repo, project_root=project) == { + "implementations/python/packages/demo.py": {"executed_lines": [1]}, + "implementations/python/packages/absolute.py": {"executed_lines": [1]}, + "tools/absolute.py": {"executed_lines": [1]}, + "tools/check.py": {"executed_lines": [1]}, + "noxfile.py": {"executed_lines": [1]}, + } + + +@pytest.mark.parametrize( + "report", + [ + {}, + {"files": []}, + {"files": {1: {}}}, + {"files": {"noxfile.py": []}}, + ], +) +def test_normalized_file_records_rejects_malformed_data(tmp_path: Path, report: dict[str, object]) -> None: + repo, _ = _repository(tmp_path) + with pytest.raises(coverage_policy.CoveragePolicyError): + coverage_policy.normalized_file_records( + report, + repo_root=repo, + project_root=repo / "implementations" / "python", + ) + + +def test_normalized_file_records_rejects_paths_outside_repository(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + outside = tmp_path / "outside.py" + outside.write_text("pass\n", encoding="utf-8") + with pytest.raises(coverage_policy.CoveragePolicyError, match="escapes the repository"): + coverage_policy.normalized_file_records( + {"files": {str(outside): {}}}, + repo_root=repo, + project_root=repo / "implementations" / "python", + ) + + +def test_canonical_coverage_config_omits_acquired_cache_but_keeps_repo_tools( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repo = tmp_path / "repo" + project = repo / "implementations" / "python" + config_path = _install_coverage_config(project) + owned_tool = repo / "tools" / "owned.py" + owned_tool.parent.mkdir(parents=True) + owned_tool.write_text("OWNED = True\n", encoding="utf-8") + acquired = repo / ".cache" / "raes-sdl" / "tooling" / "isabelle" / "appendix_gen.py" + acquired.parent.mkdir(parents=True) + acquired.write_text('print "Python 2 syntax"\n', encoding="utf-8") + monkeypatch.chdir(project) + + measured = Coverage(config_file=str(config_path), data_file=str(tmp_path / ".coverage")) + measured.start() + runpy.run_path(str(owned_tool)) + measured.stop() + measured.save() + report_path = tmp_path / "coverage.xml" + measured.xml_report(outfile=str(report_path)) + + report = report_path.read_text(encoding="utf-8") + assert "tools/owned.py" in report + assert "appendix_gen.py" not in report + + +def test_canonical_coverage_config_passes_policy_validation() -> None: + coverage_policy.validate_coverage_config(coverage_policy.load_toml(PROJECT_CONFIG, trusted_root=REPO_ROOT)) + + +@pytest.mark.parametrize( + ("section", "key", "value", "match"), + [ + ("run", "branch", False, "enable branch data"), + ("run", "relative_files", True, "same-checkout absolute source paths"), + ("run", "source", ["packages"], "canonical repository source root"), + ("run", "omit", ["*/.cache/*"], "canonical non-source paths"), + ("run", "plugins", ["weakener"], "can narrow measurement"), + ("report", "include_namespace_packages", False, "namespace-package"), + ("report", "exclude_also", [".*"], "governed legacy exclusion"), + ("report", "partial_branches", [".*"], "can suppress measurement"), + ("report", "partial_also", [".*"], "can suppress measurement"), + ("report", "exclude_lines", [".*"], "can suppress measurement"), + ], +) +def test_coverage_config_rejects_scope_and_suppression_weakening( + section: str, + key: str, + value: object, + match: str, +) -> None: + config = deepcopy(coverage_policy.load_toml(PROJECT_CONFIG, trusted_root=REPO_ROOT)) + config["tool"]["coverage"][section][key] = value + + with pytest.raises(coverage_policy.CoveragePolicyError, match=match): + coverage_policy.validate_coverage_config(config) + + +@pytest.mark.parametrize("missing", ["tool", "coverage", "run", "report"]) +def test_coverage_config_requires_governed_tables(missing: str) -> None: + config = deepcopy(coverage_policy.load_toml(PROJECT_CONFIG, trusted_root=REPO_ROOT)) + if missing == "tool": + config.pop("tool") + elif missing == "coverage": + config["tool"].pop("coverage") + else: + config["tool"]["coverage"].pop(missing) + + with pytest.raises(coverage_policy.CoveragePolicyError, match="must contain"): + coverage_policy.validate_coverage_config(config) + + +def test_coverage_config_rejects_path_aliases() -> None: + config = deepcopy(coverage_policy.load_toml(PROJECT_CONFIG, trusted_root=REPO_ROOT)) + config["tool"]["coverage"]["paths"] = {"source": ["implementations/python/packages", "elsewhere"]} + + with pytest.raises(coverage_policy.CoveragePolicyError, match="path aliases"): + coverage_policy.validate_coverage_config(config) + + +def test_changed_coverage_reports_missing_files_lines_and_branch_exits(tmp_path: Path) -> None: + changed = { + "implementations/python/packages/absent.py": {1}, + "implementations/python/packages/demo.py": {1, 2, 3, 4}, + } + packages = tmp_path / "implementations" / "python" / "packages" + packages.mkdir(parents=True) + (packages / "absent.py").write_text("ABSENT = True\n", encoding="utf-8") + (packages / "demo.py").write_text("\n".join(f"VALUE_{line} = {line}" for line in range(1, 5)), encoding="utf-8") + records = { + "implementations/python/packages/demo.py": { + "executed_lines": [1], + "missing_lines": [2], + "excluded_lines": [4], + "missing_branches": [[1, 2], [3, -1], [4], 5], + } + } + + assert coverage_policy.changed_coverage_failures(changed, records, repo_root=tmp_path) == [ + "implementations/python/packages/absent.py: changed measured Python file is absent from coverage data", + "implementations/python/packages/demo.py:2: changed executable line is not covered", + "implementations/python/packages/demo.py:4: changed line is excluded from coverage", + "implementations/python/packages/demo.py:1->2: changed branch exit is not covered", + "implementations/python/packages/demo.py:3->-1: changed branch exit is not covered", + ] + (tmp_path / "noxfile.py").write_text("\n" * 9, encoding="utf-8") + assert ( + coverage_policy.changed_coverage_failures( + {"noxfile.py": {9}}, + {"noxfile.py": {}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_ignores_changed_comments_and_blank_lines(tmp_path: Path) -> None: + path = "tools/comments.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text("# changed comment\n\n", encoding="utf-8") + + assert ( + coverage_policy.changed_coverage_failures( + {path: {1, 2}}, + {path: {}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_maps_multiline_condition_changes_to_branch_owner(tmp_path: Path) -> None: + path = "tools/multiline_branch.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """def choose(flag: bool) -> int: + if ( + flag + ): + return 1 + return 0 +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {3}}, + { + path: { + "executed_lines": [1, 2, 5], + "missing_lines": [6], + "excluded_lines": [], + "executed_branches": [[2, 5]], + "missing_branches": [[2, 6]], + } + }, + repo_root=tmp_path, + ) == [f"{path}:2->6: changed branch exit is not covered"] + + +def test_changed_coverage_keeps_branch_owner_when_condition_continuation_is_reported(tmp_path: Path) -> None: + path = "tools/reported_condition.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """def choose(flag: bool) -> int: + if ( + check(flag) + ): + return 1 + return 0 +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {3}}, + { + path: { + "executed_lines": [1, 2, 3, 5], + "missing_lines": [6], + "missing_branches": [[2, 6]], + } + }, + repo_root=tmp_path, + ) == [f"{path}:2->6: changed branch exit is not covered"] + + +def test_changed_coverage_maps_multiline_loop_iterable_to_branch_owner(tmp_path: Path) -> None: + path = "tools/multiline_loop.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """for item in ( + items +): + consume(item) +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {2}}, + { + path: { + "executed_lines": [1, 2, 4], + "missing_lines": [], + "missing_branches": [[1, -1]], + } + }, + repo_root=tmp_path, + ) == [f"{path}:1->-1: changed branch exit is not covered"] + + +def test_changed_coverage_maps_multiline_match_subject_to_case_branches(tmp_path: Path) -> None: + path = "tools/multiline_match.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """match ( + value +): + case 1: + consume_one() + case _: + consume_other() +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {2}}, + { + path: { + "executed_lines": [1, 4, 5], + "missing_lines": [6, 7], + "executed_branches": [[4, 5]], + "missing_branches": [[4, 6]], + } + }, + repo_root=tmp_path, + ) == [f"{path}:4->6: changed branch exit is not covered"] + + +def test_changed_coverage_maps_multiline_match_guard_to_case_branch(tmp_path: Path) -> None: + path = "tools/multiline_match_guard.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """match value: + case int() if ( + positive(value) + ): + consume_positive() + case _: + consume_other() +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {3}}, + { + path: { + "executed_lines": [1, 2, 5], + "missing_lines": [6, 7], + "executed_branches": [[2, 5]], + "missing_branches": [[2, 6]], + } + }, + repo_root=tmp_path, + ) == [f"{path}:2->6: changed branch exit is not covered"] + + +def test_changed_coverage_maps_comprehension_changes_when_reported_as_branches(tmp_path: Path) -> None: + path = "tools/multiline_comprehension.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """result = [ + normalize(value) + for value in values +] +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {2}}, + { + path: { + "executed_lines": [1, 2, 3], + "missing_lines": [], + "executed_branches": [[3, 2]], + "missing_branches": [[3, -1]], + } + }, + repo_root=tmp_path, + ) == [f"{path}:3->-1: changed branch exit is not covered"] + + +def test_changed_coverage_does_not_map_unrelated_code_to_comprehension_branch(tmp_path: Path) -> None: + path = "tools/unrelated_comprehension.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """result = [ + normalize(value) + for value in values +] +unrelated = updated_value() +""", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {5}}, + { + path: { + "executed_lines": [1, 2, 3, 5], + "missing_lines": [], + "executed_branches": [[3, 2]], + "missing_branches": [[3, -1]], + } + }, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_maps_multiline_expression_changes_to_statement_owner(tmp_path: Path) -> None: + path = "tools/multiline_expression.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """value = ( + build_value() +) +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {2}}, + {path: {"executed_lines": [], "missing_lines": [1], "missing_branches": []}}, + repo_root=tmp_path, + ) == [f"{path}:1: changed executable line is not covered"] + + +def test_changed_coverage_does_not_map_comment_only_continuation_lines(tmp_path: Path) -> None: + path = "tools/multiline_comment.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """if ( + # explanation only + enabled +): + run() +""", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {2}}, + {path: {"executed_lines": [1, 5], "missing_lines": [], "missing_branches": [[1, -1]]}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_fails_closed_when_code_has_no_reported_owner(tmp_path: Path) -> None: + path = "tools/unmapped.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text("VALUE = 1\n", encoding="utf-8") + + assert coverage_policy.changed_coverage_failures( + {path: {1}}, + {path: {}}, + repo_root=tmp_path, + ) == [f"{path}:1: changed code is absent from the coverage line mapping"] + + +def test_changed_coverage_ignores_docstring_only_changes(tmp_path: Path) -> None: + path = "tools/docstrings.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + '''"""Module +documentation. +""" + +class Example: + """Class documentation.""" + + def sync(self) -> None: + """Sync documentation.""" + + async def asynchronous(self) -> None: + """Async documentation.""" +''', + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {2, 6, 9, 12}}, + {path: {"executed_lines": [5, 8, 11], "missing_lines": [], "missing_branches": []}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_allows_only_structural_default_exclusions(tmp_path: Path) -> None: + path = "implementations/python/packages/declarations.py" + source = tmp_path / path + source.parent.mkdir(parents=True) + source.write_text( + """from __future__ import annotations +from typing import TYPE_CHECKING, Protocol + +if TYPE_CHECKING: + from demo import Model + +class Store(Protocol): + def load(self) -> Model: ... + + def save( + self, + value: Model, + ) -> None: ... + +class Concrete: + def omitted(self) -> None: ... +""", + encoding="utf-8", + ) + changed_lines = {4, 5, 8, 9, 10, 11, 12, 13, 14, 16} + records = { + path: { + "executed_lines": [], + "missing_lines": [], + "excluded_lines": sorted(changed_lines), + "missing_branches": [], + } + } + + assert coverage_policy.changed_coverage_failures( + {path: changed_lines}, + records, + repo_root=tmp_path, + ) == [f"{path}:16: changed line is excluded from coverage"] + + +def test_changed_coverage_rejects_untrusted_type_checking_and_recognizes_generic_protocols(tmp_path: Path) -> None: + path = "implementations/python/packages/qualified_declarations.py" + source = tmp_path / path + source.parent.mkdir(parents=True) + source.write_text( + """from __future__ import annotations + +import typing +from typing import Protocol + +if typing.TYPE_CHECKING: + from demo import Model + +class Store(Protocol[Model]): + @property + def item(self) -> Model: + \"\"\"Return the current model.\"\"\" + ... + + def implemented(self) -> None: + return None + +class Derived(Store, typing.Protocol): + async def save(self) -> None: ... + +class Generated(factory()): + pass +""", + encoding="utf-8", + ) + structural_lines = {6, 7, 10, 11, 12, 13, 19} + + assert coverage_policy.changed_coverage_failures( + {path: structural_lines}, + { + path: { + "executed_lines": [], + "missing_lines": [], + "excluded_lines": sorted(structural_lines), + "missing_branches": [], + } + }, + repo_root=tmp_path, + ) == [ + f"{path}:6: changed line is excluded from coverage", + f"{path}:7: changed line is excluded from coverage", + f"{path}:10: changed line is excluded from coverage", + f"{path}:19: changed line is excluded from coverage", + ] + + +@pytest.mark.parametrize( + "prelude", + [ + "TYPE_CHECKING = True\n", + "from other_module import TYPE_CHECKING\n", + "from typing import TYPE_CHECKING as TYPE_CHECKING\n", + "from typing import TYPE_CHECKING\nTYPE_CHECKING = runtime_flag\n", + "from typing import TYPE_CHECKING\nmodule.TYPE_CHECKING = runtime_flag\n", + "from typing import TYPE_CHECKING\nmatch {}:\n case {**TYPE_CHECKING}:\n pass\n", + 'from typing import TYPE_CHECKING\nglobals()["TYPE_CHECKING"] = True\n', + "from typing import TYPE_CHECKING\nglobals().update(TYPE_CHECKING=True)\n", + "from typing import TYPE_CHECKING\nglobals().update(dict(TYPE_CHECKING=True))\n", + "from typing import TYPE_CHECKING\nglobals().update(**{'TYPE_CHECKING': True})\n", + 'from typing import TYPE_CHECKING\nsetattr(module, "TYPE_CHECKING", True)\n', + ], +) +def test_changed_coverage_rejects_spoofed_or_rebound_type_checking( + tmp_path: Path, + prelude: str, +) -> None: + path = "tools/spoofed_type_checking.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text(f"{prelude}if TYPE_CHECKING:\n hidden_runtime = 1\n", encoding="utf-8") + if_line = len(prelude.splitlines()) + 1 + + assert coverage_policy.changed_coverage_failures( + {path: {if_line, if_line + 1}}, + { + path: { + "executed_lines": [], + "missing_lines": [], + "excluded_lines": [if_line, if_line + 1], + "missing_branches": [], + } + }, + repo_root=tmp_path, + ) == [ + f"{path}:{if_line}: changed line is excluded from coverage", + f"{path}:{if_line + 1}: changed line is excluded from coverage", + ] + + +def test_changed_coverage_allows_read_only_globals_lookup_of_type_checking(tmp_path: Path) -> None: + path = "tools/read_type_checking.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + 'from typing import TYPE_CHECKING\nglobals().get("TYPE_CHECKING")\nif TYPE_CHECKING:\n from demo import Model\n', + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {3, 4}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [3, 4]}}, + repo_root=tmp_path, + ) + == [] + ) + + +@pytest.mark.parametrize( + "prelude", + [ + "Protocol = runtime_base\n", + "from other_module import Protocol\n", + "from typing import Protocol as Protocol\n", + 'from typing import Protocol\nglobals()["Protocol"] = runtime_base\n', + "from typing import Protocol\nglobals().update(dict(Protocol=runtime_base))\n", + "from typing import Protocol\nglobals().update(**{'Protocol': runtime_base})\n", + ], +) +def test_changed_coverage_rejects_spoofed_or_rebound_protocol(tmp_path: Path, prelude: str) -> None: + path = "tools/spoofed_protocol.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text(f"{prelude}class Hidden(Protocol):\n def runtime(self): ...\n", encoding="utf-8") + method_line = len(prelude.splitlines()) + 2 + + assert coverage_policy.changed_coverage_failures( + {path: {method_line}}, + { + path: { + "executed_lines": [], + "missing_lines": [], + "excluded_lines": [method_line], + "missing_branches": [], + } + }, + repo_root=tmp_path, + ) == [f"{path}:{method_line}: changed line is excluded from coverage"] + + +@pytest.mark.parametrize( + "mutation", + [ + pytest.param('ns = globals()\nns["$SYMBOL"] = $VALUE', id="mapping-alias"), + pytest.param('ns = locals()\nns["$SYMBOL"] = $VALUE', id="locals-mapping-alias"), + pytest.param('ns = vars()\nns["$SYMBOL"] = $VALUE', id="vars-mapping-alias"), + pytest.param('namespace = globals\nnamespace()["$SYMBOL"] = $VALUE', id="factory-alias"), + pytest.param('exec("$SYMBOL = $VALUE")', id="exec"), + pytest.param('exec("$SYMBOL = $VALUE", globals())', id="exec-namespace"), + pytest.param('run = exec\nrun("$SYMBOL = $VALUE")', id="exec-alias"), + pytest.param( + 'run = eval\nrun("globals().__setitem__(\\"$SYMBOL\\", $VALUE)")', + id="eval-alias", + ), + pytest.param( + 'import sys\nvars(sys.modules[__name__])["$SYMBOL"] = $VALUE', + id="module-vars", + ), + pytest.param( + 'import sys\nsys.modules[__name__].__dict__["$SYMBOL"] = $VALUE', + id="module-dict", + ), + pytest.param( + 'import sys\ngetattr(sys.modules[__name__], "__dict__")["$SYMBOL"] = $VALUE', + id="module-dict-getattr", + ), + pytest.param("dict.update(globals(), $SYMBOL=$VALUE)", id="dict-update-descriptor"), + pytest.param( + 'dict.__setitem__(globals(), "$SYMBOL", $VALUE)', + id="dict-setitem-descriptor", + ), + pytest.param('mutate(globals(), "$SYMBOL", $VALUE)', id="unknown-mutator"), + pytest.param("globals().__init__($SYMBOL=$VALUE)", id="mapping-init"), + pytest.param( + 'import builtins, sys\nbuiltins.setattr(sys.modules[__name__], "$SYMBOL", $VALUE)', + id="builtins-setattr", + ), + pytest.param( + 'from builtins import setattr as assign\nimport sys\nassign(sys.modules[__name__], "$SYMBOL", $VALUE)', + id="imported-setattr", + ), + pytest.param( + 'import sys\nassign = setattr\nassign(sys.modules[__name__], "$SYMBOL", $VALUE)', + id="setattr-alias", + ), + pytest.param( + 'import sys\nremove = delattr\nremove(sys.modules[__name__], "$SYMBOL")', + id="delattr-alias", + ), + pytest.param("reader = globals().get", id="read-method-escape"), + pytest.param("from runtime_symbols import *", id="star-import"), + pytest.param( + 'from builtins import globals as ns\nns()["$SYMBOL"] = $VALUE', + id="imported-builtins-globals", + ), + pytest.param( + 'import builtins\nbuiltins.globals()["$SYMBOL"] = $VALUE', + id="attribute-builtins-globals", + ), + pytest.param( + 'import builtins\ngetattr(builtins, "globals")()["$SYMBOL"] = $VALUE', + id="getattr-builtins-globals", + ), + pytest.param( + 'import builtins\nbuiltins.getattr(builtins, "globals")()["$SYMBOL"] = $VALUE', + id="attribute-getattr-builtins-globals", + ), + pytest.param( + "from builtins import getattr as lookup\nimport builtins\n" + 'lookup(builtins, "globals")()["$SYMBOL"] = $VALUE', + id="imported-getattr-builtins-globals", + ), + pytest.param( + 'import builtins\nlookup = getattr\nlookup(builtins, "globals")()["$SYMBOL"] = $VALUE', + id="aliased-getattr-builtins-globals", + ), + pytest.param( + 'from builtins import vars as ns\nns()["$SYMBOL"] = $VALUE', + id="imported-builtins-vars", + ), + pytest.param('__builtins__["exec"]("$SYMBOL = $VALUE")', id="builtins-mapping-exec"), + pytest.param( + 'import inspect\ninspect.currentframe().f_globals["$SYMBOL"] = $VALUE', + id="inspect-frame-globals", + ), + pytest.param( + 'import sys\nsys._getframe().f_globals["$SYMBOL"] = $VALUE', + id="sys-frame-globals", + ), + pytest.param( + 'import sys\nsys._getframe().f_locals["$SYMBOL"] = $VALUE', + id="sys-frame-locals", + ), + ], +) +@pytest.mark.parametrize(("symbol", "value"), [("TYPE_CHECKING", "True"), ("Protocol", "runtime_base")]) +def test_structural_typing_rejects_namespace_escape_and_dynamic_rebinding( + tmp_path: Path, + mutation: str, + symbol: str, + value: str, +) -> None: + path = f"tools/unsafe_{symbol.lower()}.py" + rendered_mutation = mutation.replace("$SYMBOL", symbol).replace("$VALUE", value) + prelude = f"from typing import {symbol}\n{rendered_mutation}\n" + if symbol == "TYPE_CHECKING": + construct = "if TYPE_CHECKING:\n hidden_runtime = 1\n" + else: + construct = "class Hidden(Protocol):\n def runtime(self): ...\n" + source = tmp_path / path + source.parent.mkdir() + source.write_text(f"{prelude}{construct}", encoding="utf-8") + first_line = len(prelude.splitlines()) + 1 + changed_lines = {first_line, first_line + 1} if symbol == "TYPE_CHECKING" else {first_line + 1} + + assert coverage_policy.changed_coverage_failures( + {path: changed_lines}, + { + path: { + "executed_lines": [], + "missing_lines": [], + "excluded_lines": sorted(changed_lines), + "missing_branches": [], + } + }, + repo_root=tmp_path, + ) == [f"{path}:{line}: changed line is excluded from coverage" for line in sorted(changed_lines)] + + +def test_changed_coverage_allows_proven_read_only_namespace_access_for_protocol(tmp_path: Path) -> None: + path = "tools/read_protocol_namespace.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + 'from typing import Protocol\nglobals().get("unrelated")\nglobals()["__name__"]\n' + "class Reader(Protocol):\n def read(self): ...\n", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {5}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [5]}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_rejects_runtime_protocol_default_expression(tmp_path: Path) -> None: + path = "tools/protocol_default.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from typing import Protocol + +class Store(Protocol): + def load( + self, + default: int = runtime_default(), + ) -> int: ... +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {6}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [3, 4, 5, 6, 7]}}, + repo_root=tmp_path, + ) == [f"{path}:6: changed line is excluded from coverage"] + + +def test_changed_coverage_rejects_runtime_protocol_annotation_expression(tmp_path: Path) -> None: + path = "tools/protocol_annotation.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from typing import Protocol + +class Store(Protocol): + def load( + self, + *args: runtime_args(), + key: runtime_annotation(), + **kwargs: runtime_kwargs(), + ) -> str: ... + + def close(self): ... +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {6, 7, 8}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [3, 4, 5, 6, 7, 8, 9, 11]}}, + repo_root=tmp_path, + ) == [ + f"{path}:6: changed line is excluded from coverage", + f"{path}:7: changed line is excluded from coverage", + f"{path}:8: changed line is excluded from coverage", + ] + + +def test_changed_coverage_rejects_every_eager_protocol_annotation_span(tmp_path: Path) -> None: + path = "tools/eager_protocol_annotations.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from typing import Protocol + +class Store(Protocol): + def load( + self, + key: runtime_types.Key, + value: runtime_types["Value"], + ) -> Left | Right: ... +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {6, 7, 8}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [3, 4, 5, 6, 7, 8]}}, + repo_root=tmp_path, + ) == [ + f"{path}:6: changed line is excluded from coverage", + f"{path}:7: changed line is excluded from coverage", + f"{path}:8: changed line is excluded from coverage", + ] + + +def test_changed_coverage_allows_postponed_protocol_annotation_expression(tmp_path: Path) -> None: + path = "tools/postponed_protocol_annotation.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from __future__ import annotations +from typing import Protocol + +class Store(Protocol): + def load( + self, + key: runtime_annotation(), + ) -> str: ... +""", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {7}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [4, 5, 6, 7, 8]}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_rejects_runtime_protocol_decorator_and_class_base(tmp_path: Path) -> None: + path = "tools/protocol_runtime_headers.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from typing import Protocol + +class Store(runtime_base(), Protocol): + @runtime_decorator() + def load(self) -> int: ... +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {3, 4}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [3, 4, 5]}}, + repo_root=tmp_path, + ) == [ + f"{path}:3: changed line is excluded from coverage", + f"{path}:4: changed line is excluded from coverage", + ] + + +def test_changed_coverage_allows_literal_protocol_default_declaration(tmp_path: Path) -> None: + path = "tools/protocol_literal_default.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from __future__ import annotations +from typing import Protocol + +class Reader(Protocol): + def read(self, size: int = ...) -> bytes: ... +""", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {5}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [4, 5]}}, + repo_root=tmp_path, + ) + == [] + ) + + +def test_changed_coverage_allows_protocol_class_docstring(tmp_path: Path) -> None: + path = "tools/documented_protocol.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """from __future__ import annotations +from typing import Protocol + +class Reader(Protocol): + \"\"\"Read bytes.\"\"\" + + def read(self) -> bytes: ... +""", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {7}}, + {path: {"executed_lines": [], "missing_lines": [], "excluded_lines": [4, 5, 7]}}, + repo_root=tmp_path, + ) + == [] + ) + + +@pytest.mark.parametrize( + "pragma", + [ + "# pragma: no cover", + "# pragma no branch", + "# pragmano branch", + "# PRAGMA : NO branch - rationale", + ], +) +def test_changed_coverage_forbids_explicit_pragmas_on_changed_lines(tmp_path: Path, pragma: str) -> None: + path = "tools/example.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text(f'notice = "# pragma: no cover"\nif enabled: {pragma}\n run()\n', encoding="utf-8") + + assert coverage_policy.changed_coverage_failures( + {path: {1, 2}}, + { + path: { + "executed_lines": [1, 2], + "missing_lines": [], + "excluded_lines": [2], + "missing_branches": [], + } + }, + repo_root=tmp_path, + ) == [f"{path}:2: coverage exclusion pragma governs changed executable code"] + + +def test_changed_coverage_rejects_unchanged_pragma_that_governs_changed_header(tmp_path: Path) -> None: + path = "tools/suppressed_branch.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """if ( + enabled +): # pragma: no branch + run() +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {2}}, + {path: {"executed_lines": [1, 4], "missing_lines": [], "missing_branches": []}}, + repo_root=tmp_path, + ) == [f"{path}:3: coverage exclusion pragma governs changed executable code"] + + +def test_changed_coverage_rejects_pragma_that_governs_changed_match_guard(tmp_path: Path) -> None: + path = "tools/suppressed_match.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """match value: + case int() if ( + positive(value) + ): # pragma: no branch + consume_positive() + case _: + consume_other() +""", + encoding="utf-8", + ) + + assert coverage_policy.changed_coverage_failures( + {path: {3}}, + {path: {"executed_lines": [1, 2, 5], "missing_lines": [6, 7], "missing_branches": []}}, + repo_root=tmp_path, + ) == [f"{path}:4: coverage exclusion pragma governs changed executable code"] + + +def test_changed_coverage_allows_unrelated_change_beside_legacy_pragma(tmp_path: Path) -> None: + path = "tools/legacy_pragma.py" + source = tmp_path / path + source.parent.mkdir() + source.write_text( + """def guarded() -> None: + try: + optional_import() + except ImportError: # pragma: no cover + explain_missing_dependency() + +unrelated = updated_value() +""", + encoding="utf-8", + ) + + assert ( + coverage_policy.changed_coverage_failures( + {path: {7}}, + { + path: { + "executed_lines": [1, 2, 3, 7], + "missing_lines": [], + "excluded_lines": [4, 5], + "missing_branches": [], + } + }, + repo_root=tmp_path, + ) + == [] + ) + + +@pytest.mark.parametrize( + ("path", "contents", "match"), + [ + ("tools/missing.py", None, "could not inspect changed Python source"), + ("tools/invalid.py", "if True\n", "could not parse changed Python source"), + ], +) +def test_changed_coverage_rejects_uninspectable_source( + tmp_path: Path, + path: str, + contents: str | None, + match: str, +) -> None: + if contents is not None: + source = tmp_path / path + source.parent.mkdir() + source.write_text(contents, encoding="utf-8") + + with pytest.raises(coverage_policy.CoveragePolicyError, match=match): + coverage_policy.changed_coverage_failures({path: {1}}, {}, repo_root=tmp_path) + + +def test_aggregate_coverage_ratchet_passes_and_reports_both_regressions() -> None: + report = { + "meta": {"branch_coverage": True}, + "totals": { + "covered_lines": 9, + "num_statements": 10, + "covered_branches": 7, + "num_branches": 10, + }, + } + assert coverage_policy.aggregate_coverage_failures(report, _ratchet(line=90, branch=70)) == [] + assert coverage_policy.aggregate_coverage_failures(report, _ratchet(line=91, branch=71)) == [ + "aggregate line coverage 90.000% is below 91.000%", + "aggregate branch coverage 70.000% is below 71.000%", + ] + + +def test_aggregate_coverage_handles_empty_denominators() -> None: + report = _report() + report["totals"] = { + "covered_lines": 0, + "num_statements": 0, + "covered_branches": 0, + "num_branches": 0, + } + assert coverage_policy.aggregate_coverage_failures(report, _ratchet()) == [] + + +@pytest.mark.parametrize( + "report", + [ + {}, + {"meta": {}, "totals": {}}, + {"meta": {"branch_coverage": False}, "totals": {}}, + {"meta": {"branch_coverage": True}}, + ], +) +def test_aggregate_coverage_requires_branch_metadata_and_totals(report: dict[str, object]) -> None: + ratchet = _ratchet() + with pytest.raises(coverage_policy.CoveragePolicyError): + coverage_policy.aggregate_coverage_failures(report, ratchet) + + +@pytest.mark.parametrize( + ("ratchet", "match"), + [ + ({"minimum_branch_percent": 90}, "must be a number"), + ({"minimum_line_percent": "invalid", "minimum_branch_percent": 90}, "must be a number"), + ({"minimum_line_percent": float("inf"), "minimum_branch_percent": 90}, "between 0 and 100"), + ({"minimum_line_percent": -1, "minimum_branch_percent": 90}, "between 0 and 100"), + ({"minimum_line_percent": 90, "minimum_branch_percent": 101}, "between 0 and 100"), + ], +) +def test_aggregate_coverage_rejects_invalid_ratchet_floors( + ratchet: dict[str, object], + match: str, +) -> None: + report = _report() + with pytest.raises(coverage_policy.CoveragePolicyError, match=match): + coverage_policy.aggregate_coverage_failures(report, ratchet) + + +def test_ratchet_history_is_monotonic_after_initial_adoption(tmp_path: Path) -> None: + repo, initial_base = _repository(tmp_path) + ratchet_path = repo / "tools" / "coverage_ratchet.json" + + assert coverage_policy.base_ratchet(repo, initial_base, ratchet_path) is None + + prior = _ratchet(line=80, branch=70) + ratchet_path.write_text(json.dumps(prior), encoding="utf-8") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "add ratchet") + ratchet_base = _git(repo, "rev-parse", "HEAD") + + assert coverage_policy.base_ratchet(repo, ratchet_base, ratchet_path) == prior + assert coverage_policy.ratchet_regression_failures(_ratchet(line=81, branch=70), prior) == [] + assert coverage_policy.ratchet_regression_failures(_ratchet(line=79, branch=69), prior) == [ + "aggregate line ratchet 79.000% is below base 80.000%", + "aggregate branch ratchet 69.000% is below base 70.000%", + ] + assert coverage_policy.ratchet_regression_failures(_ratchet(), None) == [] + + +def test_ratchet_cannot_move_away_from_its_canonical_history(tmp_path: Path) -> None: + repo, base = _repository(tmp_path) + renamed = repo / "tools" / "coverage_floor.json" + renamed.write_text(json.dumps(_ratchet(line=1, branch=1)), encoding="utf-8") + + with pytest.raises(coverage_policy.CoveragePolicyError, match="must remain at the canonical path"): + coverage_policy.base_ratchet(repo, base, renamed) + + +def test_missing_base_ratchet_fails_after_prior_canonical_adoption(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + ratchet_path = repo / "tools" / "coverage_ratchet.json" + ratchet_path.write_text(json.dumps(_ratchet(line=80, branch=70)), encoding="utf-8") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "adopt ratchet") + ratchet_path.unlink() + _git(repo, "add", "-u") + _git(repo, "commit", "-m", "remove ratchet") + base_without_ratchet = _git(repo, "rev-parse", "HEAD") + + with pytest.raises(coverage_policy.CoveragePolicyError, match="missing at base.*after its canonical adoption"): + coverage_policy.base_ratchet(repo, base_without_ratchet, ratchet_path) + + +def test_base_ratchet_rejects_outside_and_malformed_files(tmp_path: Path) -> None: + repo, _ = _repository(tmp_path) + outside = tmp_path / "outside.json" + with pytest.raises(coverage_policy.CoveragePolicyError, match="escapes the repository"): + coverage_policy.base_ratchet(repo, "HEAD", outside) + + ratchet_path = repo / "tools" / "coverage_ratchet.json" + ratchet_path.write_bytes(b"\xff") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "invalid unicode ratchet") + with pytest.raises(coverage_policy.CoveragePolicyError, match="is not valid JSON"): + coverage_policy.base_ratchet(repo, "HEAD", ratchet_path) + + ratchet_path.write_text("[]", encoding="utf-8") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "non-object ratchet") + with pytest.raises(coverage_policy.CoveragePolicyError, match="must contain a JSON object"): + coverage_policy.base_ratchet(repo, "HEAD", ratchet_path) + + +def test_load_json_accepts_objects_and_rejects_bad_inputs(tmp_path: Path) -> None: + path = tmp_path / "value.json" + path.write_text('{"value": 1}', encoding="utf-8") + assert coverage_policy.load_json(path, trusted_root=tmp_path) == {"value": 1} + + path.write_text("[]", encoding="utf-8") + with pytest.raises(coverage_policy.CoveragePolicyError, match="JSON object"): + coverage_policy.load_json(path, trusted_root=tmp_path) + path.write_text("{", encoding="utf-8") + with pytest.raises(coverage_policy.CoveragePolicyError, match="could not read"): + coverage_policy.load_json(path, trusted_root=tmp_path) + missing_path = tmp_path / "missing.json" + with pytest.raises(coverage_policy.CoveragePolicyError, match="could not read"): + coverage_policy.load_json(missing_path, trusted_root=tmp_path) + outside_path = tmp_path.parent / f"{tmp_path.name}-outside.json" + with pytest.raises(coverage_policy.CoveragePolicyError, match="escapes the trusted root"): + coverage_policy.load_json(outside_path, trusted_root=tmp_path) + + +def test_load_toml_accepts_objects_and_rejects_bad_inputs(tmp_path: Path) -> None: + path = tmp_path / "value.toml" + path.write_text("value = 1\n", encoding="utf-8") + assert coverage_policy.load_toml(path, trusted_root=tmp_path) == {"value": 1} + + path.write_text("value = [\n", encoding="utf-8") + with pytest.raises(coverage_policy.CoveragePolicyError, match="could not read"): + coverage_policy.load_toml(path, trusted_root=tmp_path) + missing_path = tmp_path / "missing.toml" + with pytest.raises(coverage_policy.CoveragePolicyError, match="could not read"): + coverage_policy.load_toml(missing_path, trusted_root=tmp_path) + outside_path = tmp_path.parent / f"{tmp_path.name}-outside.toml" + with pytest.raises(coverage_policy.CoveragePolicyError, match="escapes the trusted root"): + coverage_policy.load_toml(outside_path, trusted_root=tmp_path) + + +def test_main_pass_failure_and_policy_error(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + repo, base = _repository(tmp_path) + project = repo / "implementations" / "python" + source = project / "packages" / "demo.py" + source.write_text("value = 2\n", encoding="utf-8") + report_path = project / "coverage.json" + config_path, ratchet_path = _main_paths(repo) + report = _report() + report["files"] = {str(source): {"executed_lines": [1], "missing_lines": [], "missing_branches": []}} + report_path.write_text(json.dumps(report), encoding="utf-8") + ratchet_path.write_text(json.dumps(_ratchet()), encoding="utf-8") + args = [ + "--coverage-json", + str(report_path), + "--coverage-config", + str(config_path), + "--ratchet", + str(ratchet_path), + "--repo-root", + str(repo), + "--project-root", + str(project), + "--base-rev", + base, + ] + + assert coverage_policy.main(args) == 0 + assert "COVERAGE_POLICY_PASS" in capsys.readouterr().out + + report["files"] = {str(source): {"executed_lines": [], "missing_lines": [1], "missing_branches": []}} + report_path.write_text(json.dumps(report), encoding="utf-8") + assert coverage_policy.main(args) == 1 + assert "COVERAGE_POLICY_FAILURE" in capsys.readouterr().out + + assert coverage_policy.main([*args[:-1], "unknown"]) == 2 + assert "COVERAGE_POLICY_ERROR" in capsys.readouterr().out + + outside_report = tmp_path / "coverage.json" + outside_report.write_text(json.dumps(report), encoding="utf-8") + escaped_args = [*args] + escaped_args[1] = str(outside_report) + assert coverage_policy.main(escaped_args) == 2 + assert "coverage JSON path escapes the repository" in capsys.readouterr().out + + +def test_main_rejects_noncanonical_report_and_project_roots( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + repo, _base = _repository(tmp_path) + project = repo / "implementations" / "python" + config_path, ratchet_path = _main_paths(repo) + report_path = project / "coverage.json" + report_path.write_text(json.dumps(_report()), encoding="utf-8") + ratchet_path.write_text(json.dumps(_ratchet()), encoding="utf-8") + args = [ + "--coverage-json", + str(report_path), + "--coverage-config", + str(config_path), + "--ratchet", + str(ratchet_path), + "--repo-root", + str(repo), + "--project-root", + str(project), + ] + + in_repo_report = repo / "coverage.json" + in_repo_report.write_text(json.dumps(_report()), encoding="utf-8") + noncanonical_report_args = [*args] + noncanonical_report_args[1] = str(in_repo_report) + assert coverage_policy.main(noncanonical_report_args) == 2 + assert "coverage JSON must remain at the canonical path" in capsys.readouterr().out + + noncanonical_project = repo / "project" + noncanonical_project.mkdir() + noncanonical_project_args = [*args] + noncanonical_project_args[9] = str(noncanonical_project) + assert coverage_policy.main(noncanonical_project_args) == 2 + assert "project root must remain at the canonical path" in capsys.readouterr().out + + escaping_project_args = [*args] + escaping_project_args[9] = str(tmp_path) + assert coverage_policy.main(escaping_project_args) == 2 + assert "project root path escapes the repository" in capsys.readouterr().out + + +def test_main_can_check_only_the_aggregate(tmp_path: Path) -> None: + config_path, ratchet_path = _main_paths(tmp_path) + project = config_path.parent + report_path = project / "coverage.json" + report_path.write_text(json.dumps(_report()), encoding="utf-8") + ratchet_path.write_text(json.dumps(_ratchet()), encoding="utf-8") + + assert ( + coverage_policy.main( + [ + "--coverage-json", + str(report_path), + "--coverage-config", + str(config_path), + "--ratchet", + str(ratchet_path), + "--repo-root", + str(tmp_path), + "--project-root", + str(project), + ] + ) + == 0 + ) + + +def test_script_entrypoint_delegates_to_main(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + config_path, ratchet_path = _main_paths(tmp_path) + project = config_path.parent + report_path = project / "coverage.json" + report_path.write_text(json.dumps(_report()), encoding="utf-8") + ratchet_path.write_text(json.dumps(_ratchet()), encoding="utf-8") + monkeypatch.setattr( + sys, + "argv", + [ + "check_changed_coverage.py", + "--coverage-json", + str(report_path), + "--coverage-config", + str(config_path), + "--ratchet", + str(ratchet_path), + "--repo-root", + str(tmp_path), + "--project-root", + str(project), + ], + ) + + with pytest.raises(SystemExit, match="0"): + runpy.run_path(str(Path(coverage_policy.__file__)), run_name="__main__") diff --git a/implementations/python/tests/test_repo_policy_tools.py b/implementations/python/tests/test_repo_policy_tools.py index e1c40e0d..44b0ed20 100644 --- a/implementations/python/tests/test_repo_policy_tools.py +++ b/implementations/python/tests/test_repo_policy_tools.py @@ -10,10 +10,12 @@ import threading import tomllib import types -from contextlib import nullcontext +from contextlib import contextmanager, nullcontext from pathlib import Path from typing import Any +from defusedxml import ElementTree as ET + REPO_ROOT = Path(__file__).resolve().parents[3] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) @@ -143,9 +145,17 @@ def chdir(self, _path: Path): for command, options in session.commands if command[:4] == ("uv", "run", "--frozen", "coverage") ] - assert [command[4] for command, _options in coverage_commands] == ["xml", "report"] - assert coverage_commands[-1][0][-2:] == ("--fail-under=50", "--format=total") + assert [command[4] for command, _options in coverage_commands] == ["xml", "json", "report"] + assert coverage_commands[0][0][5:] == ("-o", str(noxfile.COVERAGE_XML_PATH)) + assert coverage_commands[1][0][5:] == ("-o", str(noxfile.COVERAGE_JSON_PATH)) + assert coverage_commands[-1][0][-1:] == ("--format=total",) assert all(options["env"] == {"COVERAGE_FILE": str(coverage_file)} for _, options in coverage_commands) + policy_script = str(noxfile.REPO_ROOT / "tools" / "check_changed_coverage.py") + policy_command = next(command for command, _options in session.commands if policy_script in command) + assert Path(policy_script).is_absolute() + config_index = policy_command.index("--coverage-config") + assert policy_command[config_index + 1] == str(noxfile.COVERAGE_CONFIG_PATH) + assert "--base-rev" not in policy_command def test_verification_lanes_run_concurrently_and_preserve_declared_order( @@ -299,19 +309,309 @@ def chdir(self, _path: Path): return nullcontext() session = FakeSession() - noxfile._finalize_parallel_coverage(session, tmp_path) + for artifact_name in noxfile.PARALLEL_COVERAGE_ARTIFACTS: + (tmp_path / artifact_name).write_bytes(b"coverage data") + noxfile._finalize_parallel_coverage(session, tmp_path, base_rev="base-sha") coverage_commands = [ (command, options) for command, options in session.commands if command[:4] == ("uv", "run", "--frozen", "coverage") ] - assert [command[4] for command, _options in coverage_commands] == ["combine", "xml", "report"] + assert [command[4] for command, _options in coverage_commands] == ["combine", "xml", "json", "report"] assert coverage_commands[0][0][5:] == ("--keep", str(tmp_path)) - assert coverage_commands[-1][0][-2:] == ("--fail-under=50", "--format=total") + assert coverage_commands[1][0][5:] == ("-o", str(noxfile.COVERAGE_XML_PATH)) + assert coverage_commands[2][0][5:] == ("-o", str(noxfile.COVERAGE_JSON_PATH)) + assert coverage_commands[-1][0][-1:] == ("--format=total",) assert all( options["env"] == {"COVERAGE_FILE": str(tmp_path / ".coverage")} for _command, options in coverage_commands ) + policy_script = str(noxfile.REPO_ROOT / "tools" / "check_changed_coverage.py") + policy_command = next(command for command, _options in session.commands if policy_script in command) + assert Path(policy_script).is_absolute() + config_index = policy_command.index("--coverage-config") + assert policy_command[config_index + 1] == str(noxfile.COVERAGE_CONFIG_PATH) + assert policy_command[-2:] == ("--base-rev", "base-sha") + + +@pytest.mark.parametrize("missing", [".coverage.unit", ".coverage.integration"]) +def test_parallel_coverage_requires_both_nonempty_lane_artifacts( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + missing: str, +) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + for artifact_name in noxfile.PARALLEL_COVERAGE_ARTIFACTS: + if artifact_name != missing: + (tmp_path / artifact_name).write_bytes(b"coverage data") + + session = types.SimpleNamespace() + expected_error = f"required coverage artifact is missing: {missing}" + with pytest.raises(RuntimeError, match=expected_error): + noxfile._finalize_parallel_coverage(session, tmp_path, base_rev="base-sha") + + +def test_parallel_coverage_rejects_empty_or_aliased_lane_artifacts( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + unit = tmp_path / ".coverage.unit" + integration = tmp_path / ".coverage.integration" + unit.write_bytes(b"coverage data") + integration.touch() + + session = types.SimpleNamespace() + with pytest.raises(RuntimeError, match="not a non-empty file: .coverage.integration"): + noxfile._finalize_parallel_coverage(session, tmp_path, base_rev="base-sha") + + integration.unlink() + integration.hardlink_to(unit) + with pytest.raises(RuntimeError, match="must be distinct files"): + noxfile._finalize_parallel_coverage(session, tmp_path, base_rev="base-sha") + + +def test_parallel_coverage_policy_script_resolves_from_real_project_cwd( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + for artifact_name in noxfile.PARALLEL_COVERAGE_ARTIFACTS: + (tmp_path / artifact_name).write_bytes(b"coverage data") + invocations: list[tuple[Path, Path]] = [] + + class FakeSession: + @contextmanager + def chdir(self, path: Path): + previous = Path.cwd() + os.chdir(path) + try: + yield + finally: + os.chdir(previous) + + def run(self, *args: str, **_kwargs: Any) -> None: + scripts = [Path(arg) for arg in args if arg.endswith("check_changed_coverage.py")] + if scripts: + invocations.append((Path.cwd(), scripts[0])) + + noxfile._finalize_parallel_coverage(FakeSession(), tmp_path, base_rev="base-sha") + + assert invocations == [(noxfile.PROJECT_ROOT, noxfile.REPO_ROOT / "tools" / "check_changed_coverage.py")] + assert invocations[0][1].is_file() + + +def test_coverage_report_paths_are_canonical_and_absolute(monkeypatch: pytest.MonkeyPatch) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + config = tomllib.loads(noxfile.COVERAGE_CONFIG_PATH.read_text(encoding="utf-8")) + + assert noxfile.COVERAGE_XML_PATH == noxfile.PROJECT_ROOT / "coverage.xml" + assert noxfile.COVERAGE_JSON_PATH == noxfile.PROJECT_ROOT / "coverage.json" + assert noxfile.COVERAGE_XML_PATH.is_absolute() + assert noxfile.COVERAGE_JSON_PATH.is_absolute() + assert config["tool"]["coverage"]["xml"]["output"] == noxfile.COVERAGE_XML_PATH.name + + +def test_nox_xml_command_overrides_configured_output( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + canonical_output = tmp_path / "canonical" / "coverage.xml" + coverage_file = tmp_path / ".coverage" + config_path = tmp_path / "pyproject.toml" + monkeypatch.setattr(noxfile, "COVERAGE_CONFIG_PATH", config_path) + monkeypatch.setattr(noxfile, "COVERAGE_XML_PATH", canonical_output) + + class FakeSession: + def __init__(self) -> None: + self.commands: list[tuple[tuple[str, ...], dict[str, Any]]] = [] + + def run(self, *args: str, **kwargs: Any) -> None: + self.commands.append((args, kwargs)) + + def chdir(self, _path: Path): + return nullcontext() + + session = FakeSession() + coverage_env = {"COVERAGE_FILE": str(coverage_file)} + noxfile._write_and_check_coverage(session, coverage_env=coverage_env) + xml_command = next( + command + for command, _options in session.commands + if "coverage" in command and command[command.index("coverage") + 1] == "xml" + ) + + config_path.write_text( + f"[tool.coverage.xml]\noutput = {json.dumps(os.devnull)}\n", + encoding="utf-8", + ) + (tmp_path / "covered.py").write_text("covered = True\n", encoding="utf-8") + process_env = os.environ | coverage_env + subprocess.run( + [sys.executable, "-m", "coverage", "run", "covered.py"], + cwd=tmp_path, + env=process_env, + check=True, + capture_output=True, + text=True, + ) + subprocess.run( + [sys.executable, "-m", "coverage", *xml_command[xml_command.index("coverage") + 1 :]], + cwd=tmp_path, + env=process_env, + check=True, + capture_output=True, + text=True, + ) + + assert canonical_output.is_file() + assert canonical_output.read_text(encoding="utf-8").startswith(" None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + project_root = tmp_path / "implementations" / "python" + tools_root = tmp_path / "tools" + packages_root = project_root / "packages" + packages_root.mkdir(parents=True) + tools_root.mkdir() + config_path = project_root / "pyproject.toml" + coverage_file = tmp_path / ".coverage" + xml_path = project_root / "coverage.xml" + config_path.write_text( + """[tool.coverage.run] +branch = true +relative_files = false +source = ["../.."] + +[tool.coverage.xml] +output = "coverage.xml" +""", + encoding="utf-8", + ) + (project_root / "hatch_build.py").write_text("hatch_value = 1\n", encoding="utf-8") + (packages_root / "demo.py").write_text("package_value = 1\n", encoding="utf-8") + (tools_root / "demo_tool.py").write_text("tool_value = 1\n", encoding="utf-8") + (tmp_path / "noxfile.py").write_text("nox_value = 1\n", encoding="utf-8") + (project_root / "exercise.py").write_text( + """import runpy + +runpy.run_path("hatch_build.py") +runpy.run_path("packages/demo.py") +runpy.run_path("../../tools/demo_tool.py") +runpy.run_path("../../noxfile.py") +""", + encoding="utf-8", + ) + coverage_env = {"COVERAGE_FILE": str(coverage_file)} + subprocess.run( + [sys.executable, "-m", "coverage", "run", "--rcfile", str(config_path), "exercise.py"], + cwd=project_root, + env=os.environ | coverage_env, + check=True, + capture_output=True, + text=True, + ) + monkeypatch.setattr(noxfile, "REPO_ROOT", tmp_path) + monkeypatch.setattr(noxfile, "PROJECT_ROOT", project_root) + monkeypatch.setattr(noxfile, "COVERAGE_CONFIG_PATH", config_path) + monkeypatch.setattr(noxfile, "COVERAGE_XML_PATH", xml_path) + + class FakeSession: + def __init__(self) -> None: + self.commands: list[tuple[tuple[str, ...], dict[str, Any]]] = [] + + def run(self, *args: str, **kwargs: Any) -> None: + self.commands.append((args, kwargs)) + + def chdir(self, _path: Path): + return nullcontext() + + session = FakeSession() + noxfile._write_and_check_coverage(session, coverage_env=coverage_env) + xml_command = next( + command + for command, _options in session.commands + if "coverage" in command and command[command.index("coverage") + 1] == "xml" + ) + subprocess.run( + [sys.executable, "-m", "coverage", *xml_command[xml_command.index("coverage") + 1 :]], + cwd=project_root, + env=os.environ | coverage_env, + check=True, + capture_output=True, + text=True, + ) + + xml_root = ET.parse(xml_path).getroot() + sources = [element.text for element in xml_root.iterfind("./sources/source")] + filenames = {element.attrib["filename"] for element in xml_root.iterfind(".//class")} + assert sources == [str(tmp_path)] + assert { + "implementations/python/hatch_build.py", + "implementations/python/packages/demo.py", + "noxfile.py", + "tools/demo_tool.py", + } <= filenames + + +def test_coverage_base_revision_defaults_and_validates_values(monkeypatch: pytest.MonkeyPatch) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + + assert noxfile._coverage_base_revision([]) == "HEAD^" + assert noxfile._coverage_base_revision(["--base-rev", "abc123"]) == "abc123" + with pytest.raises(ValueError, match="--base-rev requires a value"): + noxfile._coverage_base_revision(["--base-rev"]) + + +def test_parallel_verification_resolves_coverage_base_before_starting( + monkeypatch: pytest.MonkeyPatch, +) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + session = types.SimpleNamespace(posargs=["--base-rev", "base-sha"]) + reporter = types.SimpleNamespace() + + def stop_after_resolution(posargs: list[str]) -> str: + assert posargs == ["--base-rev", "base-sha"] + raise RuntimeError("base resolved") + + monkeypatch.setattr(noxfile, "_coverage_base_revision", stop_after_resolution) + + with pytest.raises(RuntimeError, match="base resolved"): + noxfile._run_parallel_verification(session, reporter, include_policy=True) + + +def test_parallel_verification_passes_exact_base_to_combined_coverage( + monkeypatch: pytest.MonkeyPatch, +) -> None: + noxfile = load_noxfile_with_fake_nox(monkeypatch) + finalized: list[str] = [] + + class FakeReporter: + def run(self, _name: str, action: object, *, detail: str | None = None) -> None: + assert callable(action) + action() + + session = types.SimpleNamespace(posargs=["--base-rev", "base-sha"], log=lambda _message: None) + monkeypatch.setattr(noxfile, "_sync_project", lambda _session: None) + monkeypatch.setattr(noxfile, "_run_project_python", lambda _session, *_args: None) + monkeypatch.setattr(noxfile, "_available_cpu_count", lambda: 2) + monkeypatch.setattr(noxfile, "_verification_lanes", lambda **_kwargs: ()) + monkeypatch.setattr(noxfile, "_verification_lane_workers", lambda **_kwargs: 1) + monkeypatch.setattr(noxfile, "run_verification_lanes", lambda *_args, **_kwargs: []) + monkeypatch.setattr( + noxfile, + "_finalize_parallel_coverage", + lambda _session, _coverage_dir, *, base_rev: finalized.append(base_rev), + ) + + noxfile._run_parallel_verification(session, FakeReporter(), include_policy=True) + + assert finalized == ["base-sha"] def test_docs_graph_uses_curated_root_and_reader_style_gate( diff --git a/noxfile.py b/noxfile.py index b529b538..33ce19a6 100644 --- a/noxfile.py +++ b/noxfile.py @@ -54,6 +54,11 @@ RUFF_CONFIG = PROJECT_ROOT / "pyproject.toml" OSV_LOCKFILE_PATH = PROJECT_ROOT / "uv.lock" OSV_REPORT_PATH = PROJECT_ROOT / "osv-scanner-report.json" +COVERAGE_XML_PATH = PROJECT_ROOT / "coverage.xml" +COVERAGE_JSON_PATH = PROJECT_ROOT / "coverage.json" +COVERAGE_CONFIG_PATH = PROJECT_ROOT / "pyproject.toml" +COVERAGE_RATCHET_PATH = REPO_ROOT / "tools" / "coverage_ratchet.json" +PARALLEL_COVERAGE_ARTIFACTS = (".coverage.unit", ".coverage.integration") TARGETED_POLICY_TESTS = [ "implementations/python/tests/test_repo_policy_tools.py", "implementations/python/tests/test_requirement_governance.py", @@ -268,18 +273,7 @@ def _run_pytest( with session.chdir(PROJECT_ROOT): _run(session, *command, env=coverage_env) if finalize_coverage: - _run(session, "uv", "run", "--frozen", "coverage", "xml", env=coverage_env) - _run( - session, - "uv", - "run", - "--frozen", - "coverage", - "report", - "--fail-under=50", - "--format=total", - env=coverage_env, - ) + _write_and_check_coverage(session, coverage_env=coverage_env) def _split_policy_session_args(posargs: list[str]) -> tuple[list[str], list[str], bool]: @@ -308,6 +302,17 @@ def _split_policy_session_args(posargs: list[str]) -> tuple[list[str], list[str] return repo_args, requirement_args, skip_requirement +def _coverage_base_revision(posargs: Sequence[str]) -> str: + values = list(posargs) + if "--base-rev" not in values: + return "HEAD^" + index = values.index("--base-rev") + value_index = index + 1 + if value_index >= len(values) or not values[value_index] or values[value_index].startswith("--"): + raise ValueError("--base-rev requires a value") + return values[value_index] + + def _parse_hygiene_posargs(posargs: Sequence[str], *, default_all_files: bool) -> HygieneSelection: staged = False base_rev: str | None = None @@ -869,7 +874,77 @@ def _run_integration_tests( ) -def _finalize_parallel_coverage(session: nox.Session, coverage_dir: Path) -> None: +def _write_and_check_coverage( + session: nox.Session, + *, + coverage_env: dict[str, str], + base_rev: str | None = None, +) -> None: + _run( + session, + "uv", + "run", + "--frozen", + "coverage", + "xml", + "-o", + str(COVERAGE_XML_PATH), + env=coverage_env, + ) + _run( + session, + "uv", + "run", + "--frozen", + "coverage", + "json", + "-o", + str(COVERAGE_JSON_PATH), + env=coverage_env, + ) + _run( + session, + "uv", + "run", + "--frozen", + "coverage", + "report", + "--format=total", + env=coverage_env, + ) + command = [ + str(REPO_ROOT / "tools" / "check_changed_coverage.py"), + "--coverage-json", + str(COVERAGE_JSON_PATH), + "--coverage-config", + str(COVERAGE_CONFIG_PATH), + "--ratchet", + str(COVERAGE_RATCHET_PATH), + "--repo-root", + str(REPO_ROOT), + "--project-root", + str(PROJECT_ROOT), + ] + if base_rev is not None: + command.extend(("--base-rev", base_rev)) + _run_project_python(session, *command) + + +def _finalize_parallel_coverage(session: nox.Session, coverage_dir: Path, *, base_rev: str) -> None: + identities: set[tuple[int, int]] = set() + for artifact_name in PARALLEL_COVERAGE_ARTIFACTS: + artifact = coverage_dir / artifact_name + try: + artifact_stat = artifact.stat() + except OSError as exc: + raise RuntimeError(f"required coverage artifact is missing: {artifact_name}") from exc + if not artifact.is_file() or artifact_stat.st_size == 0: + raise RuntimeError(f"required coverage artifact is not a non-empty file: {artifact_name}") + identity = (artifact_stat.st_dev, artifact_stat.st_ino) + if identity in identities: + raise RuntimeError("unit and integration coverage artifacts must be distinct files") + identities.add(identity) + coverage_file = coverage_dir / ".coverage" coverage_env = {"COVERAGE_FILE": str(coverage_file)} with session.chdir(PROJECT_ROOT): @@ -884,18 +959,7 @@ def _finalize_parallel_coverage(session: nox.Session, coverage_dir: Path) -> Non str(coverage_dir), env=coverage_env, ) - _run(session, "uv", "run", "--frozen", "coverage", "xml", env=coverage_env) - _run( - session, - "uv", - "run", - "--frozen", - "coverage", - "report", - "--fail-under=50", - "--format=total", - env=coverage_env, - ) + _write_and_check_coverage(session, coverage_env=coverage_env, base_rev=base_rev) def _run_docker_integration_tests(session: nox.Session, reporter: SessionReporter) -> None: @@ -1427,6 +1491,7 @@ def _run_parallel_verification( *, include_policy: bool, ) -> None: + coverage_base_rev = _coverage_base_revision(session.posargs) reporter.run( "verify / locked project environment", lambda: _sync_project(session), @@ -1492,8 +1557,8 @@ def _execute_lanes() -> None: ) reporter.run( "verify / combined coverage", - lambda: _finalize_parallel_coverage(session, coverage_dir), - detail="unit + integration data files", + lambda: _finalize_parallel_coverage(session, coverage_dir, base_rev=coverage_base_rev), + detail=f"unit + integration data files; changed-code base={coverage_base_rev}", ) diff --git a/tools/check_changed_coverage.py b/tools/check_changed_coverage.py new file mode 100644 index 00000000..e3e81da5 --- /dev/null +++ b/tools/check_changed_coverage.py @@ -0,0 +1,993 @@ +"""Enforce branch-aware aggregate and changed-code coverage policy.""" + +from __future__ import annotations + +import argparse +import ast +import io +import json +import math +import re +import subprocess +import tokenize +import tomllib +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any + +_HUNK = re.compile( + r"^@@ -(?P\d+)(?:,(?P\d+))? " + r"\+(?P\d+)(?:,(?P\d+))? @@", + re.MULTILINE, +) +_FORBIDDEN_COVERAGE_PRAGMA = re.compile( + r"#\x20?pragma\x20?:?\x20?no\x20(?:cover|branch)(?!\w)", + re.IGNORECASE, +) +_MEASURED_PREFIXES = ("implementations/python/packages/", "tools/") +_MEASURED_FILES = frozenset({"noxfile.py", "implementations/python/hatch_build.py"}) +_CANONICAL_PROJECT_ROOT = "implementations/python" +_CANONICAL_COVERAGE_JSON = f"{_CANONICAL_PROJECT_ROOT}/coverage.json" +_CANONICAL_COVERAGE_CONFIG = "implementations/python/pyproject.toml" +_CANONICAL_RATCHET = "tools/coverage_ratchet.json" +_COVERAGE_SOURCE = ("../..",) +_COVERAGE_OMIT = frozenset( + { + "*/.cache/*", + "*/docs/*", + "*/implementations/python/.venv/*", + "*/implementations/python/tests/*", + } +) +_COVERAGE_EXCLUDE_ALSO = ("def " + "_default_runner",) +_FORBIDDEN_RUN_OPTIONS = frozenset({"include", "plugins", "source_dirs", "source_pkgs"}) +_NAMESPACE_FACTORIES = frozenset({"globals", "locals", "vars"}) +_NAMESPACE_MAPPING_ATTRIBUTES = frozenset({"__dict__", "__globals__", "f_globals", "f_locals"}) +_READ_ONLY_NAMESPACE_METHODS = frozenset({"__contains__", "__getitem__", "copy", "get", "items", "keys", "values"}) +_UNSAFE_DYNAMIC_BUILTINS = frozenset({"delattr", "eval", "exec", "getattr", "setattr"}) +_UNSAFE_BUILTINS_EXPORTS = _NAMESPACE_FACTORIES | _UNSAFE_DYNAMIC_BUILTINS +_UNSAFE_ATTRIBUTE_NAMES = _NAMESPACE_MAPPING_ATTRIBUTES | _UNSAFE_BUILTINS_EXPORTS +_FORBIDDEN_REPORT_OPTIONS = frozenset( + { + "exclude_lines", + "ignore_errors", + "include", + "omit", + "partial_also", + "partial_branches", + } +) +_NON_CODE_TOKEN_TYPES = frozenset( + { + tokenize.COMMENT, + tokenize.DEDENT, + tokenize.ENCODING, + tokenize.ENDMARKER, + tokenize.INDENT, + tokenize.NEWLINE, + tokenize.NL, + } +) + + +class CoveragePolicyError(RuntimeError): + """Raised when coverage policy inputs cannot be trusted.""" + + +def _git(repo_root: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[bytes]: + result = subprocess.run( + ("git", *args), + cwd=repo_root, + check=False, + capture_output=True, + ) + if check and result.returncode != 0: + detail = result.stderr.decode("utf-8", errors="replace").strip() + raise CoveragePolicyError(detail or f"git {' '.join(args)} failed") + return result + + +def _validate_base(repo_root: Path, base_rev: str) -> None: + try: + _git(repo_root, "rev-parse", "--verify", f"{base_rev}^{{commit}}") + except CoveragePolicyError as exc: + raise CoveragePolicyError(f"coverage base {base_rev!r} does not resolve to a commit") from exc + ancestor = _git(repo_root, "merge-base", "--is-ancestor", base_rev, "HEAD", check=False) + if ancestor.returncode != 0: + raise CoveragePolicyError(f"coverage base {base_rev!r} is not an ancestor of HEAD") + + +def is_measured_python_path(path: str) -> bool: + """Return whether a repository-relative path belongs to the coverage policy.""" + + return path.endswith(".py") and (path in _MEASURED_FILES or path.startswith(_MEASURED_PREFIXES)) + + +def added_lines_from_patch(patch: str) -> set[int]: + """Return destination line numbers from zero-context unified diff hunks.""" + + lines: set[int] = set() + for match in _HUNK.finditer(patch): + start = int(match.group("start")) + count = int(match.group("count") or "1") + lines.update(range(start, start + count)) + return lines + + +def deletion_anchor_lines_from_patch( + patch: str, + *, + base_semantic_lines: set[int], + current_semantic_lines: set[int], +) -> set[int]: + """Map semantic deletion-only hunks to surviving destination neighbors.""" + + anchors: set[int] = set() + for match in _HUNK.finditer(patch): + old_start = int(match.group("old_start")) + old_count = int(match.group("old_count") or "1") + new_start = int(match.group("start")) + new_count = int(match.group("count") or "1") + if new_count != 0 or not set(range(old_start, old_start + old_count)) & base_semantic_lines: + continue + before = [line for line in current_semantic_lines if line <= max(1, new_start)] + after = [line for line in current_semantic_lines if line >= max(1, new_start + 1)] + if before: + anchors.add(max(before)) + if after: + anchors.add(min(after)) + return anchors + + +def _semantic_physical_lines(source: str, *, path: str) -> set[int]: + tree = _parse_source(source, path=path) + tokens = tuple(tokenize.generate_tokens(io.StringIO(source).readline)) + return _significant_source_lines(tokens) - _docstring_lines(tree) + + +def changed_python_lines(repo_root: Path, base_rev: str) -> dict[str, set[int]]: + """Return changed measured Python destination lines since an exact ancestor.""" + + _validate_base(repo_root, base_rev) + changed_names = _git( + repo_root, + "diff", + "--name-only", + "-z", + "--diff-filter=AMR", + "--find-renames", + base_rev, + "--", + ).stdout.split(b"\0") + untracked_names = _git(repo_root, "ls-files", "--others", "--exclude-standard", "-z").stdout.split(b"\0") + paths = sorted( + path + for raw in {*changed_names, *untracked_names} + if raw and is_measured_python_path(path := raw.decode("utf-8", errors="surrogateescape")) + ) + changed: dict[str, set[int]] = {} + for path in paths: + tracked = _git(repo_root, "ls-files", "--error-unmatch", "--", path, check=False).returncode == 0 + if tracked: + patch = _git( + repo_root, + "diff", + "--unified=0", + "--no-ext-diff", + "--find-renames", + base_rev, + "--", + path, + ).stdout.decode("utf-8", errors="replace") + destination_lines = added_lines_from_patch(patch) + base_source = _git(repo_root, "show", f"{base_rev}:{path}", check=False) + if base_source.returncode == 0: + try: + before = base_source.stdout.decode("utf-8") + after = (repo_root / path).read_text(encoding="utf-8") + except (OSError, UnicodeError) as exc: + raise CoveragePolicyError(f"could not inspect changed Python source {path}: {exc}") from exc + destination_lines.update( + deletion_anchor_lines_from_patch( + patch, + base_semantic_lines=_semantic_physical_lines(before, path=path), + current_semantic_lines=_semantic_physical_lines(after, path=path), + ) + ) + changed[path] = destination_lines + else: + changed[path] = set(range(1, len((repo_root / path).read_bytes().splitlines()) + 1)) + return changed + + +def _repository_path(raw_path: str, repo_root: Path, project_root: Path) -> str: + path = Path(raw_path) + if path.is_absolute(): + resolved = path.resolve() + else: + project_candidate = (project_root / path).resolve() + repo_candidate = (repo_root / path).resolve() + resolved = project_candidate if project_candidate.exists() else repo_candidate + try: + return resolved.relative_to(repo_root.resolve()).as_posix() + except ValueError as exc: + raise CoveragePolicyError(f"coverage path escapes the repository: {raw_path}") from exc + + +def normalized_file_records( + report: Mapping[str, Any], + *, + repo_root: Path, + project_root: Path, +) -> dict[str, Mapping[str, Any]]: + """Index Coverage.py file records by repository-relative destination path.""" + + raw_files = report.get("files") + if not isinstance(raw_files, Mapping): + raise CoveragePolicyError("coverage JSON has no files mapping") + records: dict[str, Mapping[str, Any]] = {} + for raw_path, record in raw_files.items(): + if not isinstance(raw_path, str) or not isinstance(record, Mapping): + raise CoveragePolicyError("coverage JSON contains a malformed file record") + path = _repository_path(raw_path, repo_root, project_root) + if is_measured_python_path(path): + records[path] = record + return records + + +def _canonical_policy_path(repo_root: Path, path: Path, expected: str, *, label: str) -> Path: + resolved_path = path.resolve() + try: + relative_path = resolved_path.relative_to(repo_root.resolve()).as_posix() + except ValueError as exc: + raise CoveragePolicyError(f"{label} path escapes the repository: {path}") from exc + if relative_path != expected: + raise CoveragePolicyError(f"{label} must remain at the canonical path {expected}") + return resolved_path + + +def load_toml(path: Path, *, trusted_root: Path) -> Mapping[str, Any]: + """Load one TOML object with a stable policy error on malformed input.""" + + resolved_path = path.resolve() + try: + resolved_path.relative_to(trusted_root.resolve()) + except ValueError as exc: + raise CoveragePolicyError(f"input path escapes the trusted root: {path}") from exc + try: + with resolved_path.open("rb") as stream: + return tomllib.load(stream) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise CoveragePolicyError(f"could not read {resolved_path}: {exc}") from exc + + +def _required_mapping(parent: Mapping[str, Any], key: str, *, label: str) -> Mapping[str, Any]: + value = parent.get(key) + if not isinstance(value, Mapping): + raise CoveragePolicyError(f"coverage config must contain {label}") + return value + + +def _canonical_omit_list(value: object) -> bool: + return ( + isinstance(value, Sequence) + and not isinstance(value, (str, bytes)) + and len(value) == len(_COVERAGE_OMIT) + and set(value) == _COVERAGE_OMIT + ) + + +def _validate_coverage_run(run: Mapping[str, Any]) -> None: + if run.get("branch") is not True: + raise CoveragePolicyError("coverage config must enable branch data") + if run.get("relative_files") is not False: + raise CoveragePolicyError("coverage config must retain same-checkout absolute source paths") + if tuple(run.get("source", ())) != _COVERAGE_SOURCE: + raise CoveragePolicyError("coverage config must measure the canonical repository source root") + if not _canonical_omit_list(run.get("omit")): + raise CoveragePolicyError("coverage config omit list must contain only canonical non-source paths") + forbidden_run = sorted(_FORBIDDEN_RUN_OPTIONS & run.keys()) + if forbidden_run: + raise CoveragePolicyError(f"coverage config run option can narrow measurement: {forbidden_run[0]}") + + +def _validate_coverage_report(report: Mapping[str, Any]) -> None: + if report.get("include_namespace_packages") is not True: + raise CoveragePolicyError("coverage config must discover namespace-package source files") + if tuple(report.get("exclude_also", ())) != _COVERAGE_EXCLUDE_ALSO: + raise CoveragePolicyError("coverage config may contain only the governed legacy exclusion") + forbidden_report = sorted(_FORBIDDEN_REPORT_OPTIONS & report.keys()) + if forbidden_report: + raise CoveragePolicyError(f"coverage config report option can suppress measurement: {forbidden_report[0]}") + + +def validate_coverage_config(config: Mapping[str, Any]) -> None: + """Reject coverage configuration that can narrow or suppress the gate.""" + + tool = _required_mapping(config, "tool", label="[tool]") + coverage = _required_mapping(tool, "coverage", label="[tool.coverage]") + if "paths" in coverage: + raise CoveragePolicyError("coverage config path aliases can merge unrelated source files") + run = _required_mapping(coverage, "run", label="[tool.coverage.run]") + report = _required_mapping(coverage, "report", label="[tool.coverage.report]") + _validate_coverage_run(run) + _validate_coverage_report(report) + + +def _is_ellipsis_declaration(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + body = node.body + if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant): + if isinstance(body[0].value.value, str): + body = body[1:] + return ( + len(body) == 1 + and isinstance(body[0], ast.Expr) + and isinstance(body[0].value, ast.Constant) + and body[0].value.value is Ellipsis + ) + + +def _literal_expression(node: ast.expr) -> bool: + try: + ast.literal_eval(node) + except (ValueError, TypeError): + return False + return True + + +def _annotation_expressions(node: ast.FunctionDef | ast.AsyncFunctionDef) -> tuple[ast.expr, ...]: + arguments = (*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs) + annotations = [argument.annotation for argument in arguments if argument.annotation is not None] + for variadic in (node.args.vararg, node.args.kwarg): + if variadic is not None and variadic.annotation is not None: + annotations.append(variadic.annotation) + if node.returns is not None: + annotations.append(node.returns) + return tuple(annotations) + + +def _postpones_annotations(tree: ast.Module) -> bool: + return any( + isinstance(statement, ast.ImportFrom) + and statement.module == "__future__" + and any(alias.name == "annotations" for alias in statement.names) + for statement in tree.body + ) + + +def _protocol_declaration_lines( + node: ast.FunctionDef | ast.AsyncFunctionDef, + *, + annotations_postponed: bool, +) -> set[int]: + lines = set(range(node.lineno, (node.end_lineno or node.lineno) + 1)) + unsafe_defaults = [default for default in node.args.defaults if not _literal_expression(default)] + unsafe_defaults.extend( + default for default in node.args.kw_defaults if default is not None and not _literal_expression(default) + ) + runtime_expressions = list(unsafe_defaults) + if not annotations_postponed: + runtime_expressions.extend(_annotation_expressions(node)) + for expression in runtime_expressions: + lines.difference_update(range(expression.lineno, (expression.end_lineno or expression.lineno) + 1)) + return lines + + +def _namespace_mapping_call(node: ast.AST) -> bool: + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in _NAMESPACE_FACTORIES + and not node.args + and not node.keywords + ) + + +def _parent_nodes(tree: ast.Module) -> dict[ast.AST, ast.AST]: + return {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)} + + +def _safe_namespace_read(node: ast.Call, parents: Mapping[ast.AST, ast.AST]) -> bool: + parent = parents.get(node) + if isinstance(parent, ast.Subscript) and parent.value is node and isinstance(parent.ctx, ast.Load): + return True + if ( + not isinstance(parent, ast.Attribute) + or parent.value is not node + or parent.attr not in _READ_ONLY_NAMESPACE_METHODS + ): + return False + invocation = parents.get(parent) + return isinstance(invocation, ast.Call) and invocation.func is parent + + +def _direct_namespace_factory_reference(node: ast.Name, parents: Mapping[ast.AST, ast.AST]) -> bool: + parent = parents.get(node) + return not isinstance(parent, ast.Call) or parent.func is not node + + +def _unsafe_import(node: ast.AST) -> bool: + if not isinstance(node, ast.ImportFrom): + return False + star_import = any(alias.name == "*" for alias in node.names) + unsafe_builtin = node.module == "builtins" and any(alias.name in _UNSAFE_BUILTINS_EXPORTS for alias in node.names) + return star_import or unsafe_builtin + + +def _unsafe_name(node: ast.AST, parents: Mapping[ast.AST, ast.AST]) -> bool: + if not isinstance(node, ast.Name): + return False + unsafe_builtin = isinstance(node.ctx, ast.Load) and node.id in _UNSAFE_DYNAMIC_BUILTINS + escaped_factory = node.id in _NAMESPACE_FACTORIES and _direct_namespace_factory_reference(node, parents) + return node.id == "__builtins__" or unsafe_builtin or escaped_factory + + +def _unsafe_call(node: ast.AST, parents: Mapping[ast.AST, ast.AST]) -> bool: + if not isinstance(node, ast.Call): + return False + callable_name = node.func.id if isinstance(node.func, ast.Name) else None + dynamic_builtin = callable_name in _UNSAFE_DYNAMIC_BUILTINS + parameterized_vars = callable_name == "vars" and bool(node.args or node.keywords) + namespace_escape = _namespace_mapping_call(node) and not _safe_namespace_read(node, parents) + return dynamic_builtin or parameterized_vars or namespace_escape + + +def _unsafe_namespace_access(tree: ast.Module) -> bool: + parents = _parent_nodes(tree) + return any( + _unsafe_import(node) + or isinstance(node, ast.Attribute) + and node.attr in _UNSAFE_ATTRIBUTE_NAMES + or _unsafe_name(node, parents) + or _unsafe_call(node, parents) + for node in ast.walk(tree) + ) + + +def _target_binding_name(node: ast.AST) -> str | None: + bound_name: str | None = None + if isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)): + bound_name = node.id + elif isinstance(node, ast.Attribute) and isinstance(node.ctx, (ast.Store, ast.Del)): + bound_name = node.attr + elif isinstance(node, ast.arg): + bound_name = node.arg + return bound_name + + +def _named_binding_name(node: ast.AST) -> str | None: + bound_name: str | None = None + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.ExceptHandler)): + bound_name = node.name + return bound_name + + +def _pattern_binding_name(node: ast.AST) -> str | None: + bound_name: str | None = None + if isinstance(node, (ast.MatchAs, ast.MatchStar)): + bound_name = node.name + elif isinstance(node, ast.MatchMapping): + bound_name = node.rest + return bound_name + + +def _rebinds_imported_symbol(node: ast.AST, symbol: str) -> bool: + return symbol in {_target_binding_name(node), _named_binding_name(node), _pattern_binding_name(node)} + + +def _canonical_typing_imports(tree: ast.Module, symbol: str) -> tuple[set[int], list[int]]: + canonical_aliases: set[int] = set() + import_lines: list[int] = [] + for statement in tree.body: + if not isinstance(statement, ast.ImportFrom) or statement.level != 0 or statement.module != "typing": + continue + for alias in statement.names: + if alias.name == symbol and alias.asname is None: + canonical_aliases.add(id(alias)) + import_lines.append(statement.lineno) + return canonical_aliases, import_lines + + +def _node_rebinds_typing_symbol(node: ast.AST, symbol: str, canonical_aliases: set[int]) -> bool: + if isinstance(node, ast.alias): + bound_name = node.asname or node.name.split(".", maxsplit=1)[0] + return bound_name == symbol and id(node) not in canonical_aliases + return _rebinds_imported_symbol(node, symbol) + + +def _typing_symbol_is_rebound(tree: ast.Module, symbol: str, canonical_aliases: set[int]) -> bool: + return any(_node_rebinds_typing_symbol(node, symbol, canonical_aliases) for node in ast.walk(tree)) + + +def _trusted_typing_import_lines(tree: ast.Module, symbol: str) -> tuple[int, ...]: + import_lines: tuple[int, ...] = () + if not _unsafe_namespace_access(tree): + canonical_aliases, candidate_lines = _canonical_typing_imports(tree, symbol) + if not _typing_symbol_is_rebound(tree, symbol, canonical_aliases): + import_lines = tuple(candidate_lines) + return import_lines + + +def _canonical_typing_reference(node: ast.expr, symbol: str) -> bool: + while isinstance(node, ast.Subscript): + node = node.value + return isinstance(node, ast.Name) and node.id == symbol + + +def _parse_source(source: str, *, path: str) -> ast.Module: + try: + return ast.parse(source, filename=path) + except SyntaxError as exc: + raise CoveragePolicyError(f"could not parse changed Python source {path}: {exc.msg}") from exc + + +def _type_checking_exclusion_lines(tree: ast.Module, import_lines: tuple[int, ...]) -> set[int]: + structural: set[int] = set() + for node in ast.walk(tree): + if ( + isinstance(node, ast.If) + and isinstance(node.test, ast.Name) + and node.test.id == "TYPE_CHECKING" + and any(line < node.lineno for line in import_lines) + ): + first_body_line = min(statement.lineno for statement in node.body) + structural.update(range(node.lineno, first_body_line)) + for statement in node.body: + structural.update(range(statement.lineno, (statement.end_lineno or statement.lineno) + 1)) + return structural + + +def _protocol_exclusion_lines( + tree: ast.Module, + import_lines: tuple[int, ...], + *, + annotations_postponed: bool, +) -> set[int]: + structural: set[int] = set() + for node in ast.walk(tree): + if ( + not isinstance(node, ast.ClassDef) + or not any(_canonical_typing_reference(base, "Protocol") for base in node.bases) + or not any(line < node.lineno for line in import_lines) + ): + continue + for declaration in node.body: + if not isinstance(declaration, (ast.FunctionDef, ast.AsyncFunctionDef)) or not _is_ellipsis_declaration( + declaration + ): + continue + structural.update( + _protocol_declaration_lines( + declaration, + annotations_postponed=annotations_postponed, + ) + ) + return structural + + +def _structural_exclusion_lines(tree: ast.Module) -> set[int]: + type_checking = _type_checking_exclusion_lines(tree, _trusted_typing_import_lines(tree, "TYPE_CHECKING")) + protocol = _protocol_exclusion_lines( + tree, + _trusted_typing_import_lines(tree, "Protocol"), + annotations_postponed=_postpones_annotations(tree), + ) + return type_checking | protocol + + +def _significant_source_lines(tokens: Sequence[tokenize.TokenInfo]) -> set[int]: + significant: set[int] = set() + for token in tokens: + if token.type in _NON_CODE_TOKEN_TYPES: + continue + significant.update(range(token.start[0], token.end[0] + 1)) + return significant + + +def _docstring_lines(tree: ast.Module) -> set[int]: + lines: set[int] = set() + containers = (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef) + for node in ast.walk(tree): + if not isinstance(node, containers) or not node.body: + continue + statement = node.body[0] + if isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Constant): + if isinstance(statement.value.value, str): + lines.update(range(statement.lineno, (statement.end_lineno or statement.lineno) + 1)) + return lines + + +def _owning_coverage_line(tree: ast.Module, physical_line: int, coverage_lines: set[int]) -> int | None: + candidates: list[tuple[int, int]] = [] + for node in ast.walk(tree): + start = getattr(node, "lineno", None) + end = getattr(node, "end_lineno", None) + if not isinstance(start, int) or not isinstance(end, int): + continue + if start in coverage_lines and start <= physical_line <= end: + candidates.append((end - start, start)) + return min(candidates)[1] if candidates else None + + +def _contains_line(node: ast.AST, physical_line: int) -> bool: + start = getattr(node, "lineno", None) + end = getattr(node, "end_lineno", None) + return isinstance(start, int) and isinstance(end, int) and start <= physical_line <= end + + +def _match_header_owners(node: ast.Match, physical_line: int, branch_sources: set[int]) -> set[int]: + owners: set[int] = set() + case_sources = {case.pattern.lineno for case in node.cases if case.pattern.lineno in branch_sources} + if _contains_line(node.subject, physical_line): + owners.update(case_sources) + for case in node.cases: + source = case.pattern.lineno + case_header = (case.pattern,) if case.guard is None else (case.pattern, case.guard) + if source in branch_sources and any(_contains_line(part, physical_line) for part in case_header): + owners.add(source) + return owners + + +def _comprehension_header_owners(node: ast.AST, physical_line: int, branch_sources: set[int]) -> set[int]: + if not _contains_line(node, physical_line): + return set() + start = node.lineno + end = node.end_lineno or start + return {source for source in branch_sources if start <= source <= end} + + +def _branch_header_nodes(node: ast.AST) -> tuple[ast.AST, ...]: + headers: tuple[ast.AST, ...] = () + if isinstance(node, (ast.If, ast.While, ast.IfExp)): + headers = (node.test,) + elif isinstance(node, (ast.For, ast.AsyncFor)): + headers = (node.target, node.iter) + return headers + + +def _branch_header_owners(tree: ast.Module, physical_line: int, branch_sources: set[int]) -> set[int]: + owners: set[int] = set() + comprehensions = (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp) + for node in ast.walk(tree): + if isinstance(node, ast.Match): + owners.update(_match_header_owners(node, physical_line, branch_sources)) + elif isinstance(node, comprehensions): + owners.update(_comprehension_header_owners(node, physical_line, branch_sources)) + else: + headers = _branch_header_nodes(node) + if ( + headers + and node.lineno in branch_sources + and any(_contains_line(header, physical_line) for header in headers) + ): + owners.add(node.lineno) + return owners + + +def _semantic_owner_spans(tree: ast.Module) -> tuple[tuple[int, int], ...]: + """Return statement-like spans to which an inline coverage pragma can apply.""" + + spans: set[tuple[int, int]] = set() + for node in ast.walk(tree): + if not isinstance(node, (ast.stmt, ast.ExceptHandler)): + continue + start = node.lineno + if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + start = min((decorator.lineno for decorator in node.decorator_list), default=start) + spans.add((start, node.end_lineno or node.lineno)) + if isinstance(node, ast.Match): + for case in node.cases: + case_end = max( + (statement.end_lineno or statement.lineno for statement in case.body), + default=case.pattern.end_lineno or case.pattern.lineno, + ) + spans.add((case.pattern.lineno, case_end)) + return tuple(sorted(spans, key=lambda span: (span[1] - span[0], span[0], span[1]))) + + +def _governing_coverage_pragmas( + tokens: Sequence[tokenize.TokenInfo], + *, + tree: ast.Module, + significant_changed: set[int], + significant_lines: set[int], +) -> set[int]: + """Return pragma lines whose semantic statement owns changed code.""" + + owner_spans = _semantic_owner_spans(tree) + governed: set[int] = set() + for token in tokens: + pragma_line = token.start[0] + if ( + token.type != tokenize.COMMENT + or pragma_line not in significant_lines + or not _FORBIDDEN_COVERAGE_PRAGMA.search(" ".join(token.string.split())) + ): + continue + owner = next((span for span in owner_spans if span[0] <= pragma_line <= span[1]), None) + if owner is not None and any(owner[0] <= line <= owner[1] for line in significant_changed): + governed.add(pragma_line) + return governed + + +def _branch_edges(record: Mapping[str, Any], key: str) -> set[tuple[int, int]]: + return { + (int(edge[0]), int(edge[1])) + for edge in record.get(key, []) + if isinstance(edge, Sequence) and not isinstance(edge, (str, bytes)) and len(edge) == 2 + } + + +def _expanded_execution_lines( + changed_lines: set[int], + *, + significant_lines: set[int], + tree: ast.Module, + record: Mapping[str, Any], +) -> tuple[set[int], set[int]]: + coverage_lines = { + int(line) for key in ("executed_lines", "missing_lines", "excluded_lines") for line in record.get(key, []) + } + branch_edges = _branch_edges(record, "executed_branches") | _branch_edges(record, "missing_branches") + branch_sources = {source for source, _destination in branch_edges} + for source, destination in branch_edges: + coverage_lines.add(source) + if destination > 0: + coverage_lines.add(destination) + + significant_changed = (changed_lines & significant_lines) - _docstring_lines(tree) + expanded = set(significant_changed) + unmapped: set[int] = set() + for physical_line in significant_changed: + expanded.update(_branch_header_owners(tree, physical_line, branch_sources)) + candidates = significant_changed - coverage_lines + for physical_line in sorted(candidates): + owner = _owning_coverage_line(tree, physical_line, coverage_lines) + if owner is None: + unmapped.add(physical_line) + else: + expanded.add(owner) + return expanded, unmapped + + +def _source_coverage_policy( + repo_root: Path, + path: str, + changed_lines: set[int], +) -> tuple[ast.Module, set[int], set[int], set[int]]: + source_path = repo_root / path + try: + source = source_path.read_text(encoding="utf-8") + except (OSError, UnicodeError) as exc: + raise CoveragePolicyError(f"could not inspect changed Python source {path}: {exc}") from exc + tree = _parse_source(source, path=path) + tokens = tuple(tokenize.generate_tokens(io.StringIO(source).readline)) + structural = _structural_exclusion_lines(tree) + significant_lines = _significant_source_lines(tokens) + significant_changed = (changed_lines & significant_lines) - _docstring_lines(tree) + forbidden = _governing_coverage_pragmas( + tokens, + tree=tree, + significant_changed=significant_changed, + significant_lines=significant_lines, + ) + return tree, structural, forbidden, significant_lines + + +def changed_coverage_failures( + changed: Mapping[str, set[int]], + records: Mapping[str, Mapping[str, Any]], + *, + repo_root: Path, +) -> list[str]: + """Return deterministic failures for uncovered changed lines and branches.""" + + failures: list[str] = [] + for path, changed_lines in sorted(changed.items()): + tree, structural_exclusions, forbidden_pragmas, significant_lines = _source_coverage_policy( + repo_root, path, changed_lines + ) + for line in sorted(forbidden_pragmas): + failures.append(f"{path}:{line}: coverage exclusion pragma governs changed executable code") + record = records.get(path) + if record is None: + failures.append(f"{path}: changed measured Python file is absent from coverage data") + continue + execution_lines, unmapped_lines = _expanded_execution_lines( + changed_lines, + significant_lines=significant_lines, + tree=tree, + record=record, + ) + for line in sorted(unmapped_lines): + failures.append(f"{path}:{line}: changed code is absent from the coverage line mapping") + executed = {int(line) for line in record.get("executed_lines", [])} + missing = {int(line) for line in record.get("missing_lines", [])} + excluded = {int(line) for line in record.get("excluded_lines", [])} + executable_changed = execution_lines & (executed | missing) + for line in sorted(executable_changed & missing): + failures.append(f"{path}:{line}: changed executable line is not covered") + for line in sorted((execution_lines & excluded) - structural_exclusions - forbidden_pragmas): + failures.append(f"{path}:{line}: changed line is excluded from coverage") + missing_branches = _branch_edges(record, "missing_branches") + for source, destination in sorted(edge for edge in missing_branches if edge[0] in execution_lines): + failures.append(f"{path}:{source}->{destination}: changed branch exit is not covered") + return failures + + +def aggregate_coverage_failures(report: Mapping[str, Any], ratchet: Mapping[str, Any]) -> list[str]: + """Return failures when line or branch totals fall below the recorded ratchet.""" + + meta = report.get("meta") + totals = report.get("totals") + if not isinstance(meta, Mapping) or meta.get("branch_coverage") is not True: + raise CoveragePolicyError("coverage JSON must contain branch coverage") + if not isinstance(totals, Mapping): + raise CoveragePolicyError("coverage JSON has no totals mapping") + covered_lines = int(totals["covered_lines"]) + statements = int(totals["num_statements"]) + covered_branches = int(totals["covered_branches"]) + branches = int(totals["num_branches"]) + line_percent = 100.0 if statements == 0 else 100.0 * covered_lines / statements + branch_percent = 100.0 if branches == 0 else 100.0 * covered_branches / branches + minimum_line = _minimum_percent(ratchet, "minimum_line_percent") + minimum_branch = _minimum_percent(ratchet, "minimum_branch_percent") + failures = [] + if line_percent + 1e-12 < minimum_line: + failures.append(f"aggregate line coverage {line_percent:.3f}% is below {minimum_line:.3f}%") + if branch_percent + 1e-12 < minimum_branch: + failures.append(f"aggregate branch coverage {branch_percent:.3f}% is below {minimum_branch:.3f}%") + return failures + + +def _minimum_percent(ratchet: Mapping[str, Any], key: str) -> float: + try: + value = float(ratchet[key]) + except (KeyError, TypeError, ValueError) as exc: + raise CoveragePolicyError(f"coverage ratchet {key!r} must be a number") from exc + if not math.isfinite(value) or not 0.0 <= value <= 100.0: + raise CoveragePolicyError(f"coverage ratchet {key!r} must be between 0 and 100") + return value + + +def base_ratchet( + repo_root: Path, + base_rev: str, + ratchet_path: Path, +) -> Mapping[str, Any] | None: + """Load the ratchet recorded at the exact base, if it already existed.""" + + ratchet_path = _canonical_policy_path( + repo_root, + ratchet_path, + _CANONICAL_RATCHET, + label="coverage ratchet", + ) + relative_path = ratchet_path.relative_to(repo_root.resolve()).as_posix() + result = _git(repo_root, "show", f"{base_rev}:{relative_path}", check=False) + if result.returncode != 0: + history = _git(repo_root, "log", "--format=%H", base_rev, "--", relative_path).stdout + if history.strip(): + raise CoveragePolicyError(f"coverage ratchet is missing at base {base_rev!r} after its canonical adoption") + return None + try: + value = json.loads(result.stdout.decode("utf-8")) + except (UnicodeError, json.JSONDecodeError) as exc: + raise CoveragePolicyError(f"coverage ratchet at base {base_rev!r} is not valid JSON") from exc + if not isinstance(value, Mapping): + raise CoveragePolicyError(f"coverage ratchet at base {base_rev!r} must contain a JSON object") + return value + + +def ratchet_regression_failures( + current: Mapping[str, Any], + previous: Mapping[str, Any] | None, +) -> list[str]: + """Return failures when a checked-in aggregate floor is lowered.""" + + if previous is None: + return [] + failures: list[str] = [] + for key, label in ( + ("minimum_line_percent", "line"), + ("minimum_branch_percent", "branch"), + ): + current_minimum = _minimum_percent(current, key) + previous_minimum = _minimum_percent(previous, key) + if current_minimum + 1e-12 < previous_minimum: + failures.append(f"aggregate {label} ratchet {current_minimum:.3f}% is below base {previous_minimum:.3f}%") + return failures + + +def load_json(path: Path, *, trusted_root: Path) -> Mapping[str, Any]: + """Load one JSON object with a stable policy error on malformed input.""" + + resolved_path = path.resolve() + try: + resolved_path.relative_to(trusted_root.resolve()) + except ValueError as exc: + raise CoveragePolicyError(f"input path escapes the trusted root: {path}") from exc + try: + value = json.loads(resolved_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise CoveragePolicyError(f"could not read {resolved_path}: {exc}") from exc + if not isinstance(value, Mapping): + raise CoveragePolicyError(f"{resolved_path} must contain a JSON object") + return value + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--coverage-json", type=Path, required=True) + parser.add_argument("--coverage-config", type=Path, required=True) + parser.add_argument("--ratchet", type=Path, required=True) + parser.add_argument("--repo-root", type=Path, required=True) + parser.add_argument("--project-root", type=Path, required=True) + parser.add_argument("--base-rev") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + coverage_config_path = _canonical_policy_path( + args.repo_root, + args.coverage_config, + _CANONICAL_COVERAGE_CONFIG, + label="coverage config", + ) + ratchet_path = _canonical_policy_path( + args.repo_root, + args.ratchet, + _CANONICAL_RATCHET, + label="coverage ratchet", + ) + project_root = _canonical_policy_path( + args.repo_root, + args.project_root, + _CANONICAL_PROJECT_ROOT, + label="project root", + ) + coverage_json_path = _canonical_policy_path( + args.repo_root, + args.coverage_json, + _CANONICAL_COVERAGE_JSON, + label="coverage JSON", + ) + validate_coverage_config(load_toml(coverage_config_path, trusted_root=args.repo_root)) + report = load_json(coverage_json_path, trusted_root=args.repo_root) + ratchet = load_json(ratchet_path, trusted_root=args.repo_root) + records = normalized_file_records( + report, + repo_root=args.repo_root, + project_root=project_root, + ) + failures = aggregate_coverage_failures(report, ratchet) + if args.base_rev is not None: + changed = changed_python_lines(args.repo_root, args.base_rev) + failures.extend( + ratchet_regression_failures( + ratchet, + base_ratchet(args.repo_root, args.base_rev, ratchet_path), + ) + ) + failures.extend( + changed_coverage_failures( + changed, + records, + repo_root=args.repo_root, + ) + ) + except CoveragePolicyError as exc: + print(f"COVERAGE_POLICY_ERROR: {exc}") + return 2 + if failures: + for failure in failures: + print(f"COVERAGE_POLICY_FAILURE: {failure}") + return 1 + print("COVERAGE_POLICY_PASS: aggregate ratchet and 100% changed-code coverage satisfied") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/coverage_ratchet.json b/tools/coverage_ratchet.json new file mode 100644 index 00000000..837cfe54 --- /dev/null +++ b/tools/coverage_ratchet.json @@ -0,0 +1,14 @@ +{ + "schema_version": 1, + "minimum_line_percent": 89.739, + "minimum_branch_percent": 75.638, + "baseline": { + "covered_lines": 74067, + "num_statements": 82536, + "covered_branches": 19542, + "num_branches": 25836, + "python": "CPython 3.12.3", + "platform": "Linux x86_64", + "suites": ["unit", "integration"] + } +}