diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md index efaaeeb295..ef38f44eeb 100644 --- a/.github/PULL_REQUEST_TEMPLATE.md +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -8,5 +8,10 @@ Fixes # - [ ] `make format` - [ ] Changelog fragment in `changelog.d/` +## Reproducer + + +Reproducer: + axiom: encoded-correct | encoded | queued | n/a: diff --git a/.github/check_fix_pr.py b/.github/check_fix_pr.py new file mode 100644 index 0000000000..9c9452a38c --- /dev/null +++ b/.github/check_fix_pr.py @@ -0,0 +1,570 @@ +"""Require a changed test and a named reproducer for fix pull requests. + +This runs before the package is installed. Read test definitions rather than +importing PR code. YAML validation uses PyYAML when available and fails closed +without it; checking evidence never needs model dependencies or datasets. +""" + +import argparse +import ast +import html +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + + +TESTS_PREFIX = "policyengine_uk/tests/" +POLICY_PREFIX = TESTS_PREFIX + "policy/" +# pytest marks and unittest decorators under which a named reproducer would +# not run, or would not need to pass. skipif (and unittest's skipIf) is allowed: +# its condition is usually a platform check that holds false in CI. +NON_RUNNING_MARKS = frozenset({"skip", "xfail", "expectedFailure"}) +# pytest does not collect a test class that defines either. +CONSTRUCTORS = ("__init__", "__new__") +# Base classes a collectable test class may name; any other base could bring a +# constructor or __test__ = False with it. A bare TestCase counts only when its +# last binding before the class is an absolute import from unittest. +TEST_CLASS_BASES = frozenset({"object", "unittest.TestCase"}) +REPRODUCER_HINT = ( + 'Use "Reproducer: policyengine_uk/tests/test_example.py::test_regression", ' + '"Reproducer: policyengine_uk/tests/policy/example.yaml::Case name", ' + 'or "Reproducer: n/a: " when model source and dependencies are unchanged.' +) + + +@dataclass +class Report: + title: str + is_fix: bool + passed: bool + messages: list[str] + + +def _relative_path(path: str) -> bool: + return bool(path) and all(part not in ("", ".", "..") for part in path.split("/")) + + +def parse_changes(text: str) -> list[tuple[str, str]]: + """Read status/path pairs from git diff --name-status --no-renames -z.""" + if not text: + return [] + fields = text.split("\0") + if fields.pop() != "" or len(fields) % 2: + raise ValueError("Changes must be NUL-terminated status/path pairs") + changes = [] + for status, path in zip(fields[::2], fields[1::2]): + if not re.fullmatch(r"[A-Z]", status): + raise ValueError(f"Unparseable change status: {status!r}") + if not _relative_path(path): + raise ValueError(f"Invalid repository path: {path!r}") + changes.append((status, path)) + return changes + + +def classify_fix_pr( + title: str, body: str, labels: list[str], changes: list[tuple[str, str]] +) -> bool: + """Classify from PR metadata only; title and body are reserved for later rules.""" + return ( + bool({"bug", "bug-fix"}.intersection(labels)) + or any( + status in ("A", "M", "T") + and path.startswith("changelog.d/") + and len(path.split("/")) == 2 + # Towncrier searches every basename part after the issue for its type. + and "fixed" in PurePosixPath(path).name.split(".")[1:] + for status, path in changes + ) + ) + + +def _pytest_module(path: str) -> bool: + name = PurePosixPath(path).name + return ( + _relative_path(path) + and path.startswith(TESTS_PREFIX) + and name not in ("conftest.py", "__init__.py") + and (name.startswith("test_") or name.endswith("_test.py")) + and name.endswith(".py") + ) + + +def _policy_test(path: str) -> bool: + return ( + _relative_path(path) + and path.startswith(POLICY_PREFIX) + and PurePosixPath(path).suffix in (".yaml", ".yml") + ) + + +def changed_executable_tests(changes: list[tuple[str, str]]) -> list[str]: + return [ + path + for status, path in changes + if status in ("A", "M", "T") and (_pytest_module(path) or _policy_test(path)) + ] + + +def _exemption(reproducer: str) -> re.Match | None: + return re.fullmatch(r"(?:n/a\s*[:\-]|none\s*:)\s*(.*)", reproducer, re.IGNORECASE) + + +def _unquote_id(value: str) -> str: + value = value.strip() + return value[1:-1] if value.startswith("`") and value.endswith("`") else value + + +def parse_reproducers(body: str) -> list[str]: + # Template guidance is not evidence supplied by the contributor. + body = re.sub(r"", "", body, flags=re.DOTALL) + ids = [] + for line in body.splitlines(): + match = re.match(r"^\s*reproducer:\s*(.*)$", line, re.IGNORECASE) + if not match or not match[1].strip(): + continue + value = match[1].strip() + start = 0 + quoted = False + brackets = 0 + for index, character in enumerate(value): + if character == "`": + quoted = not quoted + elif not quoted: + if character == "[": + brackets += 1 + elif character == "]" and brackets: + brackets -= 1 + elif character in ",;" and brackets == 0: + ids.append(_unquote_id(value[start:index])) + start = index + 1 + ids.append(_unquote_id(value[start:])) + return ids + + +def _read_test(repo_root: Path, path: str, head_sha: str | None) -> bytes: + if head_sha is not None: + + def git(*args: str) -> bytes: + return subprocess.run( + ["git", "-C", str(repo_root), "--literal-pathspecs", *args], + check=True, + capture_output=True, + timeout=15, + ).stdout + + entries = git("ls-tree", "-z", head_sha, "--", path).split(b"\0") + if len(entries) != 2 or entries[-1] != b"": + raise ValueError("the test file is missing from the PR head") + metadata, entry_path = entries[0].split(b"\t", 1) + mode, kind, _ = metadata.split() + if entry_path.decode("utf-8", errors="surrogateescape") != path: + raise ValueError("the test file is missing from the PR head") + # A symlink's blob is its target text, not an executable test file. + if mode == b"120000": + raise ValueError("a symlink cannot be a reproducer") + if mode not in (b"100644", b"100755") or kind != b"blob": + raise ValueError("the reproducer must be a regular test file") + return git("cat-file", "blob", f"{head_sha}:{path}") + + # Filesystem mode supports unit fixtures; CI always reads committed blobs. + file = repo_root + for part in path.split("/"): + file = file / part + if file.is_symlink(): + raise ValueError("a symlink cannot be a reproducer") + if not file.resolve().is_relative_to(repo_root.resolve()): + raise ValueError("the test file must be inside the PR checkout") + return file.read_bytes() + + +def _assigns_name(node: ast.AST, name: str) -> bool: + targets = ( + node.targets + if isinstance(node, ast.Assign) + else [node.target] + if isinstance(node, ast.AnnAssign) + else [] + ) + return any(isinstance(target, ast.Name) and target.id == name for target in targets) + + +def _suppresses_tests(body: list[ast.stmt]) -> bool: + # pytest skips a class whose __test__ is falsy; anything but a literal True + # counts as switching collection off. + return any( + _assigns_name(node, "__test__") + and not (isinstance(node.value, ast.Constant) and node.value.value is True) + for node in body + ) + + +def _dotted_name(node: ast.AST) -> str | None: + parts = [] + while isinstance(node, ast.Attribute): + parts.append(node.attr) + node = node.value + if not isinstance(node, ast.Name): + return None + parts.append(node.id) + return ".".join(reversed(parts)) + + +def _sets_test_attribute(body: list[ast.stmt], name: str) -> bool: + """Whether the body assigns name.__test__, which pytest reads on collection.""" + for node in body: + if isinstance(node, ast.Assign): + targets = node.targets + elif isinstance(node, (ast.AnnAssign, ast.AugAssign)): + targets = [node.target] + else: + continue + if any( + isinstance(target, ast.Attribute) + and target.attr == "__test__" + and _dotted_name(target.value) == name + for target in targets + ): + return True + return False + + +def _binds_name(node: ast.AST, name: str) -> bool: + """Whether a statement could bind name, including inside compound blocks.""" + for part in ast.walk(node): + if isinstance(part, ast.Name) and isinstance(part.ctx, ast.Store): + if part.id == name: + return True + elif isinstance(part, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + if part.name == name: + return True + elif isinstance(part, (ast.Import, ast.ImportFrom)): + if any( + alias.name == "*" or (alias.asname or alias.name.split(".")[0]) == name + for alias in part.names + ): + return True + return False + + +def _imports_unittest_testcase(body: list[ast.stmt]) -> bool: + """Whether the last top-level binding of TestCase in body is unittest's.""" + from_unittest = False + for node in body: + if ( + isinstance(node, ast.ImportFrom) + and node.module == "unittest" + and node.level == 0 + and all(alias.name != "*" for alias in node.names) + ): + for alias in node.names: + if (alias.asname or alias.name) == "TestCase": + from_unittest = alias.name == "TestCase" + elif _binds_name(node, "TestCase"): + from_unittest = False + return from_unittest + + +def _decorator_names(definition: ast.AST) -> list[str | None]: + return [ + _dotted_name(decorator.func if isinstance(decorator, ast.Call) else decorator) + for decorator in definition.decorator_list + ] + + +def _mark_attributes(node: ast.AST): + """Attribute nodes in a pytestmark value, leaving out call arguments.""" + if isinstance(node, ast.Call): + yield from _mark_attributes(node.func) + return + if isinstance(node, ast.Attribute): + yield node + for child in ast.iter_child_nodes(node): + yield from _mark_attributes(child) + + +def _pytestmark_names(body: list[ast.stmt]) -> list[str | None]: + """The marks in a module's or class's pytestmark assignment. + + Call arguments are left out, so config.skip in + pytest.mark.skipif(config.skip, ...) is not taken for a mark, while an + aliased mark such as m.skip still counts. + """ + names = [] + for node in body: + if _assigns_name(node, "pytestmark") and node.value is not None: + names.extend(_dotted_name(part) for part in _mark_attributes(node.value)) + return names + + +def _not_run_problem(names: list[str | None]) -> str | None: + """Why marks or decorators would keep the named test from running, if so.""" + for name in names: + if name is None: + continue + parts = name.split(".") + if parts[-1] == "fixture": + return "the named test is a fixture, which pytest does not run as a test" + if parts[-1] in NON_RUNNING_MARKS: + return ( + f"the named test is marked {parts[-1]}, so it would not run as a " + "reproducer" + ) + return None + + +def _reference_problem( + reproducer: str, tests: set[str], repo_root: Path, head_sha: str | None +) -> str | None: + path, separator, name = reproducer.partition("::") + if path not in tests: + return "the path must be an added, modified or type-changed executable test in this PR" + try: + source = _read_test(repo_root, path, head_sha) + if _policy_test(path): + try: + import yaml + except ImportError: + return "cannot validate YAML tests without PyYAML" + try: + entries = yaml.safe_load(source.decode("utf-8")) + except yaml.YAMLError as error: + return f"the YAML file cannot be parsed: {error}" + names = ( + { + entry["name"] + for entry in entries + if isinstance(entry, dict) + and isinstance(entry.get("name"), str) + and isinstance(entry.get("output"), dict) + and entry["output"] + } + if isinstance(entries, list) + else set() + ) + if not names: + return "the YAML file has no test mapping with a string name and a non-empty output mapping" + if separator and name not in names: + return f"the YAML file has no test entry named {name!r}" + return None + if not separator or not name: + return "a pytest reproducer must name a test with ::test_name" + # Parameters can contain ::, so remove them before splitting the node id. + if "[" in name: + if not name.endswith("]"): + return "pytest parameters must be a terminal [param] suffix" + name = name.split("[", 1)[0] + parts = name.split("::") + tree = ast.parse(source, filename=path) + if _suppresses_tests(tree.body): + return "pytest collection is disabled by module __test__" + module_marks = _pytestmark_names(tree.body) + definitions = (ast.FunctionDef, ast.AsyncFunctionDef) + # A later definition of the same name replaces an earlier one. + functions = [ + node + for node in tree.body + if isinstance(node, definitions) and node.name == parts[0] + ] + if len(parts) == 1 and parts[0].startswith("test") and functions: + if _sets_test_attribute(tree.body, parts[0]): + return "the named test's __test__ attribute is set after its definition" + return _not_run_problem(module_marks + _decorator_names(functions[-1])) + classes = [ + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == parts[0] + ] + if ( + len(parts) == 2 + and parts[0].startswith("Test") + and parts[1].startswith("test") + and classes + ): + cls = classes[-1] + methods = [ + node + for node in cls.body + if isinstance(node, definitions) and node.name == parts[1] + ] + # pytest does not collect a class with its own constructor or + # __test__ switched off, and a base class could bring either. + collectable = ( + not _suppresses_tests(cls.body) + and not any( + (isinstance(node, definitions) and node.name in CONSTRUCTORS) + or any(_assigns_name(node, name) for name in CONSTRUCTORS) + for node in cls.body + ) + and all( + _dotted_name(base) in TEST_CLASS_BASES + or ( + _dotted_name(base) == "TestCase" + and _imports_unittest_testcase( + tree.body[: tree.body.index(cls)] + ) + ) + for base in cls.bases + ) + and not cls.keywords + and not _sets_test_attribute(tree.body, parts[0]) + and not _sets_test_attribute(cls.body, parts[1]) + ) + if collectable and methods: + return _not_run_problem( + module_marks + + _decorator_names(cls) + + _pytestmark_names(cls.body) + + _decorator_names(methods[-1]) + ) + return "the pytest file does not define a collectable named test" + except ( + OSError, + RuntimeError, + SyntaxError, + UnicodeError, + ValueError, + subprocess.SubprocessError, + ) as error: + return f"the test file cannot be read or parsed: {error}" + + +def check_fix_pr( + title: str, + body: str, + labels: list[str], + changes: list[tuple[str, str]], + repo_root: Path, + head_sha: str | None = None, +) -> Report: + if head_sha is not None and not re.fullmatch( + r"(?:[0-9a-fA-F]{40}|[0-9a-fA-F]{64})", head_sha + ): + raise ValueError("The PR head must be a full hexadecimal commit SHA") + if not classify_fix_pr(title, body, labels, changes): + return Report( + title, + False, + True, + [ + "Not a fix PR: no added/modified/type-changed fixed fragment or bug/bug-fix label." + ], + ) + tests = set(changed_executable_tests(changes)) + model_changed = any( + status in ("A", "M", "T", "D") + and ( + (path.startswith("policyengine_uk/") and not path.startswith(TESTS_PREFIX)) + or path in ("pyproject.toml", "uv.lock") + ) + for status, path in changes + ) + reproducer_messages = [] + valid_exemption = False + reproducers = parse_reproducers(body) + for reproducer in reproducers: + exemption = _exemption(reproducer) + if exemption: + problem = ( + "n/a requires a non-empty reason" + if not exemption[1].strip() + else "n/a is allowed only when model source outside tests, pyproject.toml and uv.lock are unchanged" + if model_changed + else None + ) + valid_exemption |= problem is None + else: + problem = _reference_problem(reproducer, tests, repo_root, head_sha) + if problem: + reproducer_messages.append( + f"Invalid reproducer {reproducer!r}: {problem}. {REPRODUCER_HINT}" + ) + if not reproducers: + reproducer_messages.append(f"Missing named reproducer. {REPRODUCER_HINT}") + messages = [] + if not tests and not valid_exemption: + messages.append( + "No added, modified or type-changed executable test. Add a pytest module under " + "policyengine_uk/tests/ or YAML test under policyengine_uk/tests/policy/. " + + REPRODUCER_HINT + ) + messages.extend(reproducer_messages) + return Report( + title, + True, + not messages, + messages + or [ + "Fix PR has valid reproducer evidence. " + + ( + "Exemption accepted: model source and dependencies are unchanged." + if valid_exemption + else f"Changed executable tests: {len(tests)}." + ) + ], + ) + + +def _emit(report: Report, summary_file: Path | None, unusable: bool = False) -> None: + outcome = "ERROR" if unusable else "PASS" if report.passed else "FAIL" + title = " ".join(report.title.splitlines()) + if not report.is_fix and not unusable: + print(f"Fix evidence: {outcome} ({title!r}). {report.messages[0]}") + else: + print(f"Fix evidence: {outcome} ({title!r})") + for message in report.messages: + print(f"- {message}") + if summary_file: + + def markdown(text: str) -> str: + return re.sub(r"([\\`*_[\]])", r"\\\1", html.escape(text)) + + with summary_file.open("a", encoding="utf-8") as summary: + summary.write(f"\n## Fix evidence\n\n{outcome}: {markdown(title)}\n\n") + for message in report.messages: + summary.write(f"- {markdown(message)}\n") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + for option in ("title", "body", "labels", "changes"): + parser.add_argument(f"--{option}-file", type=Path, required=True) + parser.add_argument("--repo-root", type=Path, required=True) + parser.add_argument( + "--head-sha", + help="Read tests from this commit's git objects instead of the filesystem", + ) + parser.add_argument("--summary-file", type=Path) + args = parser.parse_args(argv) + title = "" + unusable = False + try: + title = args.title_file.read_text(encoding="utf-8").strip() + body = args.body_file.read_text(encoding="utf-8") + labels = [ + label.strip() + for label in args.labels_file.read_text(encoding="utf-8").splitlines() + if label.strip() + ] + # Preserve path bytes, including literal newlines, rather than decoding Git quotes. + changes = parse_changes( + args.changes_file.read_bytes().decode("utf-8", errors="surrogateescape") + ) + if not args.repo_root.is_dir(): + raise ValueError(f"Repository root is not a directory: {args.repo_root}") + report = check_fix_pr( + title, body, labels, changes, args.repo_root, args.head_sha + ) + except (OSError, UnicodeError, ValueError) as error: + report = Report(title, False, False, [f"Unusable input: {error}"]) + unusable = True + try: + _emit(report, args.summary_file, unusable) + except (OSError, UnicodeError) as error: + print(f"Fix evidence: ERROR. Cannot append summary: {error}") + return 2 + return 2 if unusable else 0 if report.passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/pr_fix_evidence.yaml b/.github/workflows/pr_fix_evidence.yaml new file mode 100644 index 0000000000..af26b99cf2 --- /dev/null +++ b/.github/workflows/pr_fix_evidence.yaml @@ -0,0 +1,79 @@ +# Fix pull requests must change an executable test and name the test that +# reproduces the bug (.github/check_fix_pr.py; docs/engineering/skills/github-prs.md). +# +# pull_request_target is safe here because the job never executes, installs or +# imports code from the PR. The base checkout supplies the trusted checker, +# which reads the PR's files as data from git objects. +# +# It also reports on merge queue groups (merge_group), so a branch rule can +# require it with a merge queue on: a group passes, because each pull +# request's evidence was checked on the pull request before it could join +# the queue, and a group has no pull request text to check. + +name: Fix evidence +on: + pull_request_target: + branches: [main] + types: [opened, edited, synchronize, reopened, labeled, unlabeled, ready_for_review] + merge_group: + +permissions: + contents: read + +concurrency: + group: fix-evidence-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request_target' }} + +jobs: + Fix-evidence: + name: Fix evidence + runs-on: ubuntu-24.04-arm + timeout-minutes: 5 + steps: + - name: Merge queue group + if: github.event_name == 'merge_group' + run: echo "Each pull request in this group passed Fix evidence before it joined the queue." + - name: Check out the trusted base branch + if: github.event_name == 'pull_request_target' + uses: actions/checkout@v6 + with: + fetch-depth: 0 + # The repository is public, so the fetches below need no token. + persist-credentials: false + - name: Collect pull request evidence + id: evidence + if: github.event_name == 'pull_request_target' + env: + PR_NUMBER: ${{ github.event.pull_request.number }} + PR_TITLE: ${{ github.event.pull_request.title }} + PR_BODY: ${{ github.event.pull_request.body }} + PR_LABELS_JSON: ${{ toJSON(github.event.pull_request.labels.*.name) }} + PR_BASE_SHA: ${{ github.event.pull_request.base.sha }} + PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + # A fresh directory keeps metadata writes away from any PR paths. + evidence_dir=$(mktemp -d "$RUNNER_TEMP/fix-evidence.XXXXXX") + printf 'directory=%s\n' "$evidence_dir" >> "$GITHUB_OUTPUT" + # PR text stays in environment variables so shell syntax in it cannot run. + printf '%s' "$PR_TITLE" > "$evidence_dir/title.txt" + printf '%s' "$PR_BODY" > "$evidence_dir/body.md" + python3 -c 'import json, os; print("\n".join(json.loads(os.environ["PR_LABELS_JSON"])), end="")' > "$evidence_dir/labels.txt" + git fetch --no-tags origin "+refs/pull/${PR_NUMBER}/head:refs/remotes/pr/head" + # Fetch the event's immutable head too in case the PR changed again. + git fetch --no-tags origin "$PR_BASE_SHA" "$PR_HEAD_SHA" + merge_base=$(git merge-base "$PR_BASE_SHA" "$PR_HEAD_SHA") + git diff --name-status --no-renames -z "$merge_base" "$PR_HEAD_SHA" > "$evidence_dir/changes.bin" + - name: Set up uv for the trusted checker + if: github.event_name == 'pull_request_target' + uses: astral-sh/setup-uv@v8.1.0 + - name: Check fix evidence + if: github.event_name == 'pull_request_target' + env: + EVIDENCE_DIR: ${{ steps.evidence.outputs.directory }} + PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + uv run --no-project --with pyyaml==6.0.2 python .github/check_fix_pr.py \ + --title-file "$EVIDENCE_DIR/title.txt" --body-file "$EVIDENCE_DIR/body.md" \ + --labels-file "$EVIDENCE_DIR/labels.txt" --changes-file "$EVIDENCE_DIR/changes.bin" \ + --repo-root . --head-sha "$PR_HEAD_SHA" \ + --summary-file "$GITHUB_STEP_SUMMARY" diff --git a/changelog.d/fix-pr-evidence-check.added.md b/changelog.d/fix-pr-evidence-check.added.md new file mode 100644 index 0000000000..7b08b4ac57 --- /dev/null +++ b/changelog.d/fix-pr-evidence-check.added.md @@ -0,0 +1 @@ +- Added a Fix evidence pull request check using a trusted base-branch checker to require bug fixes to name a collectable Python or executable YAML test from the PR's git objects, with an exemption only when model files and dependencies are unchanged. diff --git a/docs/engineering/skills/github-prs.md b/docs/engineering/skills/github-prs.md index a1ac512107..ecc892b694 100644 --- a/docs/engineering/skills/github-prs.md +++ b/docs/engineering/skills/github-prs.md @@ -54,6 +54,56 @@ unless the user or maintainer explicitly asks. The type controls the inferred version bump: `breaking` is major, `added` and `removed` are minor, and other types are patch. +## Fix evidence + +The **Fix evidence** pull request check applies when a PR adds, modifies or +type-changes a fixed fragment directly in `changelog.d/`, or has the `bug` or +`bug-fix` label. A fragment counts as fixed when its basename has `fixed` in +any dot-separated part after the first, including `x.fixed.md`, `x.fixed.1.md`, +`x.fixed.md.md` and `x.fixed`. Use the canonical fragment naming convention +above when writing new fragments. A fix PR must add, modify or type-change an +executable test under `policyengine_uk/tests/` and name its reproducer in the +PR description: + +```text +Reproducer: policyengine_uk/tests/path/to/test_file.py::test_name +``` + +Accepted ids are pytest node ids (`file.py::test_name` or +`file.py::TestClass::test_name`) and YAML policy tests under +`policyengine_uk/tests/policy/` (`file.yaml` or `file.yaml::test name`, also +accepting `.yml`). Each named file must be added, modified or type-changed in +the PR. Python files must match `test_*.py` or `*_test.py`, and names must +resolve to module-level functions starting with `test`, or methods starting +with `test` in module-level classes starting with `Test`. Such a class may +derive only from `object` or `unittest.TestCase` (also as `TestCase` imported +from `unittest`) and may not define `__init__` or `__new__`. A module or class +whose `__test__` is anything but `True`, a test or class whose `__test__` +attribute is set after it is defined, a fixture, and definitions nested in +functions or conditional blocks are excluded. The reproducer must run and +pass: a `skip` or `xfail` mark, or unittest's `skip` or `expectedFailure`, on +it, its class or its module (`pytestmark`) is rejected. A `skipif` (or +unittest `skipIf`/`skipUnless`) condition must let the reproducer run in CI, +and the test must not skip itself at run time (`pytest.skip`, +`pytest.importorskip`), which the check cannot see. A test marked +`microsimulation` runs only where the dataset is available, which includes +merge-queue and main runs. The check reads test files statically, so it +catches mistakes, not deliberate evasion; reviewers still read the test. YAML files must parse to a non-empty list with at +least one mapping containing a string `name` and a non-empty `output` mapping; +a named YAML reproducer must match such a mapping exactly. +Separate multiple ids with commas or semicolons. Backticks around each id +are optional; use them when a name or reason contains these separators. +Reproducer lines inside HTML comments are ignored. + +For CI/docs-only fixes, use `Reproducer: n/a: ` with a non-empty reason. +This exempts the PR from changing a test only when no file under +`policyengine_uk/` outside `policyengine_uk/tests/` is added, modified, +type-changed or deleted, and neither `pyproject.toml` nor `uv.lock` changes. +Dependency changes can change model results, so they also require a test. +The workflow runs the trusted base-branch checker and reads test files from +the PR head's git objects, rejecting symlinked test paths. It validates the +named reproducer without running PR code or testing against the old code. + ## Required Axiom Line Every PR that adds, updates or fixes policy (programmes, parameters, eligibility diff --git a/policyengine_uk/tests/code_health/test_check_fix_pr.py b/policyengine_uk/tests/code_health/test_check_fix_pr.py new file mode 100644 index 0000000000..986767f1d1 --- /dev/null +++ b/policyengine_uk/tests/code_health/test_check_fix_pr.py @@ -0,0 +1,1300 @@ +"""Fix PRs carry executable, named evidence of the bug they fix. + +The check runs without installing the model. These tests hold its rules to +the invariants reviewers rely on: + +1. Exactly an added, modified or type-changed Towncrier fixed fragment, + or a bug/bug-fix label, classifies a PR as a fix. +2. Deleted tests never count as changed executable tests or reproducers. +3. Formatting reproducer ids with supported separators and quoting round + trips through the body parser. +4. A pytest reproducer requires a changed file and a directly defined test + function or class method that pytest collects, checked against real pytest + collection, ignoring parametrisation suffixes. A collected test marked to + skip or expected to fail is rejected too, since it reproduces nothing. +5. An n/a reason is accepted exactly when it is non-empty and no model + source outside the test tree or dependency file was added, modified, + type-changed or deleted. +6. Non-fix PRs pass regardless of their bodies. +7. The CLI returns 0/1/2 for passing/failing/unusable input and appends its + Markdown report without replacing an existing job summary. +8. NUL-delimited git diffs preserve literal paths, and git-object reads + validate committed content without following symlinks or executing PR code. + The standalone checker entry point obeys the same passing and failing rules. +9. The workflow runs the trusted base checker on pull_request_target events, + isolates artifacts under RUNNER_TEMP and reports on merge queue groups. +10. YAML reproducers require parsed list entries with a string name and a + non-empty output mapping; missing PyYAML fails closed. +11. Reproducer lines inside HTML comments never count as evidence. +""" + +import importlib.util +import os +import re +import runpy +import subprocess +import sys +from pathlib import Path + +import pytest +from hypothesis import HealthCheck, example, given, settings +from hypothesis import strategies as st + +TESTS_DIR = Path(__file__).resolve().parents[1] +REPO_ROOT = TESTS_DIR.parents[1] +CHECKER = REPO_ROOT / ".github" / "check_fix_pr.py" + +if not CHECKER.exists(): + pytest.skip( + "the .github folder is not part of an installed package", + allow_module_level=True, + ) + +_spec = importlib.util.spec_from_file_location("check_fix_pr", CHECKER) +checker = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = checker +_spec.loader.exec_module(checker) + +PROPERTY_SETTINGS = settings( + max_examples=20, + deadline=None, + derandomize=True, + # Cold imports can be charged to input generation on busy runners. + suppress_health_check=[HealthCheck.function_scoped_fixture, HealthCheck.too_slow], +) + +_identifier = st.from_regex(r"[A-Za-z_][A-Za-z0-9_]{0,10}", fullmatch=True) +_status = st.sampled_from(list("ABCDEFGHIJKLMNOPQRSTUVWXYZ")) +_label = st.one_of( + st.sampled_from( + ["bug", "bug-fix", "Bug", "bugfix", "enhancement", "documentation"] + ), + st.text(alphabet="abcdefghijklmnopqrstuvwxyz-_", max_size=20), +) +_change_path = st.one_of( + st.builds( + lambda name, suffix: f"changelog.d/{name}.{suffix}", + _identifier, + st.sampled_from(["fixed.md", "fixed.1.md", "fixed.md.md", "fixed"]), + ), + st.sampled_from( + [ + "changelog.d/regression.added.md", + "changelog.d/regression.changed.md", + "changelog.d/fixed.md", + "changelog.d/nested/regression.fixed.md", + "docs/regression.fixed.md", + "policyengine_uk/variables/regression.py", + "policyengine_uk/parameters/regression.yaml", + "policyengine_uk/__init__.py", + "policyengine_uk/tests/test_regression.py", + "policyengine_uk/tests/policy/regression.yaml", + "policyengine_uk/tests2/test_regression.py", + ".github/workflows/check.yaml", + "docs/regression.md", + "pyproject.toml", + "uv.lock", + ] + ), +) +_changes = st.lists(st.tuples(_status, _change_path), max_size=12) +_python_path = _identifier.map( + lambda name: f"policyengine_uk/tests/code_health/test_{name}.py" +) +_yaml_path = st.builds( + lambda name, extension: f"policyengine_uk/tests/policy/{name}.{extension}", + _identifier, + st.sampled_from(["yaml", "yml"]), +) +_executable_path = st.one_of(_python_path, _yaml_path) +_params = st.one_of( + st.just(""), + st.text(alphabet="abc012 ::/-,;", max_size=15).map(lambda value: f"[{value}]"), +) +_valid_id = st.one_of( + st.builds( + lambda path, name, params: f"{path}::test_{name}{params}", + _python_path, + _identifier, + _params, + ), + st.builds( + lambda path, name, params: f"{path}::TestRegression::test_{name}{params}", + _python_path, + _identifier, + _params, + ), + _yaml_path, + st.builds(lambda path, name: f"{path}::{name}", _yaml_path, _identifier), + st.builds( + lambda prefix, reason: prefix + reason, + st.sampled_from(["n/a: ", "n/a - ", "none: "]), + _identifier, + ), +) + + +def _write(repo_root, path, text): + destination = repo_root / path + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(text, encoding="utf-8") + return destination + + +def _report(repo_root, body, changes, labels=("bug",), *, head_sha=None): + return checker.check_fix_pr( + "Fix a regression", body, list(labels), changes, repo_root, head_sha=head_sha + ) + + +def _fixed_fragment(path): + return ( + Path(path).parent == Path("changelog.d") + and "fixed" in Path(path).name.split(".")[1:] + ) + + +@PROPERTY_SETTINGS +@given(labels=st.sets(_label, max_size=10), changes=_changes) +def test_classification_matches_only_fragments_and_labels(labels, changes): + expected = bool({"bug", "bug-fix"} & labels) or any( + status in {"A", "M", "T"} and _fixed_fragment(path) for status, path in changes + ) + for title, body in [("Fix a regression", "Reproducer: missing"), ("", "")]: + assert checker.classify_fix_pr(title, body, list(labels), changes) is expected + + +@pytest.mark.parametrize("status", ["A", "M", "T"]) +@pytest.mark.parametrize( + "basename", ["x.fixed.md", "x.fixed.1.md", "x.fixed.md.md", "x.fixed"] +) +def test_towncrier_fixed_fragments_require_evidence(tmp_path, status, basename): + changes = [ + (status, f"changelog.d/{basename}"), + ("M", "policyengine_uk/variables/regression.py"), + ] + report = _report(tmp_path, "No evidence.", changes, labels=()) + assert report.is_fix + assert not report.passed + + +@PROPERTY_SETTINGS +@given(paths=st.lists(_executable_path, min_size=1, max_size=6, unique=True)) +def test_deleted_tests_never_count_as_executable_or_reproducers(tmp_path, paths): + changes = [("D", path) for path in paths] + assert checker.changed_executable_tests(changes) == [] + + path = paths[0] + content = ( + "def test_regression():\n pass\n" + if path.endswith(".py") + else "- name: regression\n output: {x: 1}\n" + ) + _write(tmp_path, path, content) + # A separate changed test isolates reproducer eligibility from the + # requirement that the PR change some executable test. + companion = "policyengine_uk/tests/test_companion.py" + _write(tmp_path, companion, "def test_companion():\n pass\n") + target = path + ("::test_regression" if path.endswith(".py") else "") + assert not _report( + tmp_path, f"Reproducer: {target}", changes + [("A", companion)] + ).passed + + +@PROPERTY_SETTINGS +@given(ids=st.lists(_valid_id, max_size=8), data=st.data()) +def test_reproducer_line_round_trips_supported_formatting(ids, data): + pieces = [] + for index, nodeid in enumerate(ids): + if index: + pieces.append(data.draw(st.sampled_from([",", ";", ", ", " ; ", "\t;\t"]))) + spaces = data.draw(st.sampled_from(["", " ", "\t", " "])) + quoted = data.draw(st.booleans()) + pieces.append(spaces + (f"`{nodeid}`" if quoted else nodeid) + spaces) + key = data.draw( + st.sampled_from(["Reproducer", "reproducer", "REPRODUCER", "RePrOdUcEr"]) + ) + body = "A description.\n" + key + ": " + "".join(pieces) + "\nA test plan." + assert checker.parse_reproducers(body) == ids + + +@PROPERTY_SETTINGS +@given( + status=_status, + name=_identifier, + defines_name=st.booleans(), + in_class=st.booleans(), + asynchronous=st.booleans(), + params=_params, +) +@example( + status="A", + name="case", + defines_name=True, + in_class=True, + asynchronous=False, + params="[a::b]", +) +@example( + status="M", + name="case", + defines_name=True, + in_class=False, + asynchronous=True, + params="[2026]", +) +def test_pytest_reproducer_requires_changed_file_and_definition( + tmp_path, status, name, defines_name, in_class, asynchronous, params +): + path = "policyengine_uk/tests/test_regression.py" + wanted = f"test_{name}" + defined = wanted if defines_name else f"test_other_{name}" + prefix = "async " if asynchronous else "" + if in_class: + source = ( + f"class TestRegression:\n {prefix}def {defined}(self):\n pass\n" + ) + nodeid = f"{path}::TestRegression::{wanted}{params}" + else: + source = f"{prefix}def {defined}():\n pass\n" + nodeid = f"{path}::{wanted}{params}" + _write(tmp_path, path, source) + + report = _report(tmp_path, f"Reproducer: {nodeid}", [(status, path)]) + assert report.passed is (status in {"A", "M", "T"} and defines_name) + + +@PROPERTY_SETTINGS +@given( + changes=_changes, + reason=st.one_of(st.sampled_from(["", " ", "\t"]), _identifier), + prefix=st.sampled_from(["n/a: ", "n/a - ", "none: "]), +) +@example( + changes=[("D", "policyengine_uk/variables/regression.py")], + reason="CI only", + prefix="n/a: ", +) +@example( + changes=[("M", "policyengine_uk/tests2/test_regression.py")], + reason="CI only", + prefix="none: ", +) +@example( + changes=[("T", "policyengine_uk/variables/regression.py")], + reason="CI only", + prefix="n/a: ", +) +@example(changes=[("M", "pyproject.toml")], reason="CI only", prefix="n/a: ") +@example(changes=[("M", "uv.lock")], reason="CI only", prefix="n/a: ") +def test_na_requires_reason_and_no_model_source_changes( + tmp_path, changes, reason, prefix +): + has_model_source = any( + status in {"A", "M", "T", "D"} + and ( + path in {"pyproject.toml", "uv.lock"} + or ( + path.startswith("policyengine_uk/") + and not path.startswith("policyengine_uk/tests/") + ) + ) + for status, path in changes + ) + report = _report(tmp_path, f"Reproducer: {prefix}{reason}", changes) + assert report.passed is (bool(reason.strip()) and not has_model_source) + + +@pytest.mark.parametrize("status", ["A", "M", "T", "D"]) +@pytest.mark.parametrize( + "path", ["policyengine_uk/variables/regression.py", "pyproject.toml", "uv.lock"] +) +def test_na_rejects_every_model_or_dependency_change(tmp_path, status, path): + assert not _report(tmp_path, "Reproducer: n/a: CI only", [(status, path)]).passed + + +@PROPERTY_SETTINGS +@given(title=st.text(max_size=80), body=st.text(max_size=300), changes=_changes) +@example( + title="Fix the model", + body="Reproducer: n/a:\nReproducer: missing::test_x", + changes=[], +) +def test_non_fix_pr_always_passes(tmp_path, title, body, changes): + non_fix_changes = [ + (status, path) + for status, path in changes + if not (status in {"A", "M", "T"} and _fixed_fragment(path)) + ] + report = checker.check_fix_pr( + title, body, ["enhancement", "Bug"], non_fix_changes, tmp_path + ) + assert not report.is_fix + assert report.passed + assert report.title == title + assert len(report.messages) == 1 + + +@pytest.mark.parametrize( + "path, executable", + [ + ("policyengine_uk/tests/test_regression.py", True), + ("policyengine_uk/tests/code_health/regression_test.py", True), + ("policyengine_uk/tests/policy/regression.yaml", True), + ("policyengine_uk/tests/policy/regression.yml", True), + ("policyengine_uk/tests/conftest.py", False), + ("policyengine_uk/tests/__init__.py", False), + ("policyengine_uk/tests/helpers.py", False), + ("policyengine_uk/tests/gov/regression.yaml", False), + ("policyengine_uk/tests/policy2/regression.yaml", False), + ("tests/test_regression.py", False), + ("policyengine_uk/variables/test_regression.py", False), + ], +) +def test_only_collected_test_paths_are_executable(path, executable): + for status in ["A", "M", "D", "T", "U"]: + assert checker.changed_executable_tests([(status, path)]) == ( + [path] if executable and status in {"A", "M", "T"} else [] + ) + + +@pytest.mark.parametrize( + "source", + [ + "- name: Regression case\n period: 2026\n input: {}\n output: {x: 1}\n", + "- name: 'Regression case'\n output: {x: 1}\n", + '- name: "Regression case"\n output: {x: 1}\n', + "- period: 2026\n name: Regression case\n output: {x: 1}\n", + "- name: Regression case # An inline comment\n output: {x: 1}\n", + ], +) +@pytest.mark.parametrize("extension", ["yaml", "yml"]) +def test_yaml_reproducer_accepts_bare_files_and_named_entries( + tmp_path, source, extension +): + path = f"policyengine_uk/tests/policy/regression.{extension}" + _write(tmp_path, path, source) + changes = [("M", path)] + assert _report(tmp_path, f"Reproducer: {path}", changes).passed + assert _report(tmp_path, f"Reproducer: {path}::Regression case", changes).passed + assert not _report(tmp_path, f"Reproducer: {path}::Different case", changes).passed + + +@pytest.mark.parametrize( + "source, name", + [ + ("- name: 'It''s a case' # Comment\n", "It's a case"), + ('- name: "A \\"quoted\\" # case" # Comment\n', 'A "quoted" # case'), + ('- name: "Case, with; punctuation"\n', "Case, with; punctuation"), + ('- name: "Regression\\x20case"\n', "Regression case"), + ('- name: "\\U00000041"\n', "A"), + ('- name: "\\\\x41"\n', "\\x41"), + ], +) +def test_quoted_yaml_names_resolve_without_treating_punctuation_as_ids( + tmp_path, source, name +): + path = "policyengine_uk/tests/policy/regression.yaml" + _write(tmp_path, path, source + " output: {x: 1}\n") + assert _report(tmp_path, f"Reproducer: `{path}::{name}`", [("A", path)]).passed + + +def test_nested_yaml_name_is_not_a_test_entry(tmp_path): + path = "policyengine_uk/tests/policy/regression.yaml" + _write( + tmp_path, + path, + "- name: Regression case\n input:\n name: Impostor\n output: {x: 1}\n", + ) + changes = [("M", path)] + assert _report(tmp_path, f"Reproducer: {path}::Regression case", changes).passed + assert not _report(tmp_path, f"Reproducer: {path}::Impostor", changes).passed + + +@pytest.mark.parametrize( + "source", + [ + "[]\n", + "# No policy tests here.\n", + "- |\n name: Ghost\n", + "- name: Ghost\n period: 2026\n", + "name: Ghost\noutput: {x: 1}\n", + "- name: 123\n output: {x: 1}\n", + "- name: [invalid\n output: {x: 1}\n", + ], +) +@pytest.mark.parametrize("suffix", ["", "::Ghost"]) +def test_yaml_reproducer_rejects_non_executable_documents(tmp_path, source, suffix): + path = "policyengine_uk/tests/policy/fake.yaml" + _write(tmp_path, path, source) + assert not _report(tmp_path, f"Reproducer: {path}{suffix}", [("A", path)]).passed + + +@pytest.mark.parametrize("output", ["{}", "[]", "5", "null", "[x]"]) +def test_yaml_reproducer_needs_a_non_empty_output_mapping(tmp_path, output): + # An entry whose output is empty or not a mapping asserts nothing. + path = "policyengine_uk/tests/policy/regression.yaml" + _write(tmp_path, path, f"- name: Regression\n output: {output}\n") + changes = [("A", path)] + assert not _report(tmp_path, f"Reproducer: {path}", changes).passed + assert not _report(tmp_path, f"Reproducer: {path}::Regression", changes).passed + + +def test_yaml_named_reproducer_requires_output_on_the_named_entry(tmp_path): + path = "policyengine_uk/tests/policy/regression.yaml" + _write( + tmp_path, + path, + "- name: Ghost\n period: 2026\n- name: Real\n output: {x: 1}\n", + ) + changes = [("A", path)] + assert _report(tmp_path, f"Reproducer: {path}", changes).passed + assert _report(tmp_path, f"Reproducer: {path}::Real", changes).passed + assert not _report(tmp_path, f"Reproducer: {path}::Ghost", changes).passed + + +def test_yaml_reproducer_fails_closed_without_pyyaml(tmp_path, monkeypatch): + path = "policyengine_uk/tests/policy/regression.yaml" + _write(tmp_path, path, "- name: Regression\n output: {x: 1}\n") + monkeypatch.setitem(sys.modules, "yaml", None) + report = _report(tmp_path, f"Reproducer: {path}::Regression", [("A", path)]) + assert not report.passed + assert any( + "cannot validate YAML tests without PyYAML" in message + for message in report.messages + ) + # Python evidence still works when the optional YAML parser is unavailable. + python_path = "policyengine_uk/tests/test_regression.py" + _write(tmp_path, python_path, "def test_regression():\n pass\n") + assert _report( + tmp_path, f"Reproducer: {python_path}::test_regression", [("A", python_path)] + ).passed + + +def test_backticked_na_reason_can_include_commas_and_semicolons(tmp_path): + assert _report(tmp_path, "Reproducer: `n/a: CI, docs; no model source`", []).passed + + +@pytest.mark.parametrize( + "source, suffix", + [ + ("def test_regression():\n pass\n", ""), + ("def test_regression():\n pass\n", "::test_regression[unterminated"), + ("def test_regression():\n pass\n", "::test_regression[p]::junk"), + ("def test_regression(:\n pass\n", "::test_regression"), + ("test_regression = True\n", "::test_regression"), + ("def helper():\n pass\n", "::helper"), + ( + "class Regression:\n def test_regression(self):\n pass\n", + "::Regression::test_regression", + ), + ( + "class TestRegression:\n def __init__(self):\n pass\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "__test__ = False\ndef test_regression():\n pass\n", + "::test_regression", + ), + ( + "def test_regression():\n pass\n__test__: bool = False\n", + "::test_regression", + ), + ( + "class TestRegression:\n __test__ = False\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "class TestRegression:\n __test__: bool = False\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "def wrapper():\n def test_regression():\n pass\n", + "::test_regression", + ), + ( + "if False:\n def test_regression():\n pass\n", + "::test_regression", + ), + ( + "if True:\n class TestRegression:\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "class TestRegression:\n if True:\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "class TestOther:\n def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "class TestRegression:\n def test_regression(self):\n pass\n", + "::TestRegression::Nested::test_regression", + ), + ( + "__test__ = 0\ndef test_regression():\n pass\n", + "::test_regression", + ), + ( + "class TestRegression:\n __test__ = None\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "def test_regression():\n pass\ntest_regression.__test__ = False\n", + "::test_regression", + ), + ( + "class TestRegression:\n def test_regression(self):\n pass\n" + "TestRegression.__test__ = False\n", + "::TestRegression::test_regression", + ), + ( + "class TestRegression:\n def __new__(cls):\n pass\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "class Base:\n def __init__(self):\n pass\n" + "class TestRegression(Base):\n def test_regression(self):\n" + " pass\n", + "::TestRegression::test_regression", + ), + ( + "import pytest\n@pytest.fixture\ndef test_regression():\n pass\n", + "::test_regression", + ), + ( + "def test_regression():\n pass\ndef test_regression(x):\n pass\n" + "test_regression.__test__ = False\n", + "::test_regression", + ), + ( + "from helpers import TestCase\nclass TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from .unittest import TestCase\nclass TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from unittest import TestCase\nfrom helpers import TestCase\n" + "class TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from helpers import TestCase\nclass TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n" + "from unittest import TestCase\n", + "::TestRegression::test_regression", + ), + ( + "from unittest import TestCase\nfrom helpers import *\n" + "class TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from unittest import TestCase\nif True:\n" + " from helpers import TestCase\n" + "class TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "import helpers\nfrom unittest import TestCase\n" + "TestCase, _ = helpers.TestCase, 0\n" + "class TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ], +) +def test_pytest_reproducer_rejects_bare_paths_and_nonexistent_definitions( + tmp_path, source, suffix +): + path = "policyengine_uk/tests/test_regression.py" + _write(tmp_path, path, source) + assert not _report(tmp_path, f"Reproducer: {path}{suffix}", [("M", path)]).passed + + +def test_pytest_default_collection_confirms_function_and_class_names( + tmp_path, monkeypatch +): + path = "policyengine_uk/tests/test_regression.py" + _write( + tmp_path, + path, + "def testament():\n pass\n" + "def helper():\n pass\n" + "class Testimony:\n def testimony(self):\n pass\n" + "class Regression:\n def test_hidden(self):\n pass\n" + "class TestConstructor:\n def __init__(self):\n pass\n" + " def test_hidden(self):\n pass\n" + "class TestDisabled:\n __test__ = False\n" + " def test_hidden(self):\n pass\n" + "class TestFalsy:\n __test__ = 0\n" + " def test_hidden(self):\n pass\n" + "class TestAllocator:\n def __new__(cls):\n pass\n" + " def test_hidden(self):\n pass\n" + "class TestInherited(TestConstructor):\n" + " def test_hidden(self):\n pass\n" + "def test_switched_off():\n pass\n" + "test_switched_off.__test__ = False\n" + "from unittest import TestCase\n" + "class TestUnittest(TestCase):\n" + " def test_case(self):\n pass\n", + ) + + class Collection: + def pytest_collection_finish(self, session): + self.nodeids = [item.nodeid for item in session.items] + + collection = Collection() + # Reuse pytest's interpreter so the oracle only pays the cost of collection. + with monkeypatch.context() as context: + context.chdir(tmp_path) + context.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + result = pytest.main( + [ + "-c", + os.devnull, + "--collect-only", + "--import-mode=importlib", + "--confcutdir", + str(tmp_path), + "-q", + "-p", + "no:cacheprovider", + path, + ], + plugins=[collection], + ) + assert result == pytest.ExitCode.OK + collected = {nodeid.split("::", 1)[1] for nodeid in collection.nodeids} + assert collected == {"testament", "Testimony::testimony", "TestUnittest::test_case"} + for suffix in [ + "testament", + "helper", + "Testimony::testimony", + "Regression::test_hidden", + "TestConstructor::test_hidden", + "TestDisabled::test_hidden", + "TestFalsy::test_hidden", + "TestAllocator::test_hidden", + "TestInherited::test_hidden", + "test_switched_off", + "TestUnittest::test_case", + ]: + nodeid = f"{path}::{suffix}" + report = _report(tmp_path, f"Reproducer: {nodeid}", [("A", path)]) + assert report.passed is (suffix in collected) + + +@pytest.mark.parametrize( + "source, suffix", + [ + ( + "import pytest\n@pytest.mark.skip(reason='later')\n" + "def test_regression():\n pass\n", + "::test_regression", + ), + ( + "from pytest import mark\n@mark.xfail\ndef test_regression():\n pass\n", + "::test_regression", + ), + ( + "import pytest\npytestmark = [pytest.mark.skip]\n" + "def test_regression():\n pass\n", + "::test_regression", + ), + ( + "import pytest\n@pytest.mark.xfail(strict=True)\nclass TestRegression:\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "import pytest\nclass TestRegression:\n" + " pytestmark = pytest.mark.skip(reason='later')\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "import unittest\nclass TestRegression(unittest.TestCase):\n" + " @unittest.skip('later')\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from unittest import TestCase, expectedFailure\n" + "class TestRegression(TestCase):\n @expectedFailure\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from pytest import mark as m\npytestmark = m.skip\n" + "def test_regression():\n pass\n", + "::test_regression", + ), + ( + "import pytest\nm = pytest.mark\nclass TestRegression:\n" + " pytestmark = [m.xfail]\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ], +) +def test_pytest_reproducer_must_not_be_skipped_or_expected_to_fail( + tmp_path, source, suffix +): + # pytest collects these, but a skipped or xfailed test reproduces nothing. + path = "policyengine_uk/tests/test_regression.py" + _write(tmp_path, path, source) + report = _report(tmp_path, f"Reproducer: {path}{suffix}", [("M", path)]) + assert not report.passed + assert any("so it would not run" in message for message in report.messages) + + +@pytest.mark.parametrize( + "source, suffix", + [ + ( + "from hypothesis import given, strategies as st\n" + "@given(st.integers())\ndef test_regression(x):\n pass\n", + "::test_regression", + ), + ( + "import pytest\n@pytest.mark.parametrize('x', [1, 2])\n" + "@pytest.mark.microsimulation\ndef test_regression(x):\n pass\n", + "::test_regression[1]", + ), + ( + "import unittest\nclass TestRegression(unittest.TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "import sys, pytest\n" + "@pytest.mark.skipif(sys.platform == 'win32', reason='no FIFOs')\n" + "def test_regression():\n pass\n", + "::test_regression", + ), + ( + "from unittest import TestCase\nclass TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "from unittest import TestCase\nclass TestRegression(TestCase):\n" + " def test_regression(self):\n pass\n" + "from helpers import TestCase\n", + "::TestRegression::test_regression", + ), + ( + "import pytest, config\n" + "pytestmark = pytest.mark.skipif(config.skip, reason='optional')\n" + "def test_regression():\n pass\n", + "::test_regression", + ), + ( + "import unittest, sys\nclass TestRegression(unittest.TestCase):\n" + " @unittest.skipIf(sys.platform == 'win32', 'no FIFOs')\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ( + "class TestRegression(object):\n __test__ = True\n" + " def test_regression(self):\n pass\n", + "::TestRegression::test_regression", + ), + ], +) +def test_pytest_reproducer_accepts_marks_and_decorators_that_still_run( + tmp_path, source, suffix +): + path = "policyengine_uk/tests/test_regression.py" + _write(tmp_path, path, source) + assert _report(tmp_path, f"Reproducer: {path}{suffix}", [("M", path)]).passed + + +def test_reproducer_file_must_exist_at_pr_head(tmp_path): + path = "policyengine_uk/tests/test_regression.py" + assert not _report( + tmp_path, f"Reproducer: {path}::test_regression", [("A", path)] + ).passed + + +@pytest.mark.parametrize("outside_checkout", [True, False]) +def test_reproducer_symlinks_cannot_escape_checkout_or_crash_on_loops( + tmp_path, outside_checkout +): + repo_root = tmp_path / "checkout" + path = "policyengine_uk/tests/test_regression.py" + destination = repo_root / path + destination.parent.mkdir(parents=True) + if outside_checkout: + target = _write(tmp_path, "outside.py", "def test_regression():\n pass\n") + else: + target = destination + destination.symlink_to(target) + assert not _report( + repo_root, f"Reproducer: {path}::test_regression", [("A", path)] + ).passed + + +def test_every_reproducer_on_every_line_must_resolve(tmp_path): + path = "policyengine_uk/tests/test_regression.py" + _write(tmp_path, path, "def test_one():\n pass\n\ndef test_two():\n pass\n") + changes = [("M", path)] + body = f"Reproducer: `{path}::test_one`\nreproducer: {path}::test_two" + assert _report(tmp_path, body, changes).passed + assert not _report(tmp_path, body + f"; {path}::test_missing", changes).passed + assert not _report(tmp_path, "Reproducer: n/a: CI only; missing", []).passed + + +def test_html_commented_reproducer_is_ignored(tmp_path): + path = "policyengine_uk/tests/test_regression.py" + _write(tmp_path, path, "def test_regression():\n pass\n") + commented = "\n" + body = commented + f"Reproducer: {path}::test_regression" + assert checker.parse_reproducers(body) == [f"{path}::test_regression"] + assert _report(tmp_path, body, [("A", path)]).passed + assert not _report( + tmp_path, f"", [("A", path)] + ).passed + + +def test_report_explains_each_failed_requirement(tmp_path): + report = _report(tmp_path, "A description without evidence.", []) + assert not report.passed + assert len(report.messages) >= 2 + assert sum("Reproducer:" in message for message in report.messages) >= 2 + + +def _arguments( + input_dir, + repo_root, + *, + body="Reproducer: n/a: CI only", + labels="bug\n", + changes="", + summary=None, + head_sha=None, +): + values = { + "title": "Fix a regression", + "body": body, + "labels": labels, + "changes": changes, + } + args = [] + for name, value in values.items(): + path = _write(input_dir, f"{name}.txt", value) + args.extend([f"--{name}-file", str(path)]) + args.extend(["--repo-root", str(repo_root)]) + if head_sha is not None: + args.extend(["--head-sha", head_sha]) + if summary is not None: + args.extend(["--summary-file", str(summary)]) + return args + + +def _standalone_checker(args, repo_root, monkeypatch): + # Execute the real trusted entry point without starting another interpreter. + with monkeypatch.context() as context: + context.chdir(repo_root) + context.setattr(sys, "argv", [str(CHECKER), *args]) + with pytest.raises(SystemExit) as exit: + runpy.run_path(str(CHECKER), run_name="__main__") + return exit.value.code + + +@pytest.mark.parametrize( + "body, labels, changes, expected", + [ + ("Reproducer: n/a: CI only", "bug\n", "M\0.github/workflows/check.yaml\0", 0), + ( + "Reproducer: policyengine_uk/tests/test_regression.py::test_regression", + "bug-fix\n", + "M\0policyengine_uk/tests/test_regression.py\0", + 0, + ), + ("No evidence.", "bug\n", "M\0policyengine_uk/variables/regression.py\0", 1), + ( + "Reproducer: invalid", + "enhancement\n", + "D\0changelog.d/regression.fixed.md\0", + 0, + ), + ( + "Reproducer: n/a: CI only", + "bug\n", + "T\0.github/workflows/check.yaml\0Q\0docs/regression.md\0", + 0, + ), + ], +) +def test_main_returns_zero_or_one_for_usable_inputs( + tmp_path, capsys, body, labels, changes, expected +): + _write( + tmp_path, + "policyengine_uk/tests/test_regression.py", + "def test_regression():\n pass\n", + ) + args = _arguments(tmp_path, tmp_path, body=body, labels=labels, changes=changes) + assert checker.main(args) == expected + assert "Fix a regression" in capsys.readouterr().out + + +@pytest.mark.parametrize("missing", ["title", "body", "labels", "changes"]) +def test_main_returns_two_for_missing_input_files(tmp_path, capsys, missing): + args = _arguments(tmp_path, tmp_path) + (tmp_path / f"{missing}.txt").unlink() + assert checker.main(args) == 2 + assert capsys.readouterr().out.strip() + + +@pytest.mark.parametrize( + "changes", + [ + "M path/to/test.py\n", + "M\tpath.py\n", + "M\0\0", + "MM\0path.py\0", + "R100\0old.py\0new.py\0", + "M\0first.py\0second.py\0", + "M\0path.py", + "M\0../outside.py\0", + "M\0/absolute.py\0", + ], +) +def test_main_returns_two_for_unparseable_changes(tmp_path, capsys, changes): + args = _arguments(tmp_path, tmp_path, changes=changes) + assert checker.main(args) == 2 + assert capsys.readouterr().out.strip() + + +def test_main_returns_two_for_missing_repository_root(tmp_path): + args = _arguments(tmp_path, tmp_path / "missing") + assert checker.main(args) == 2 + + +def test_main_returns_two_for_unwritable_summary(tmp_path): + # A directory is unwritable as a summary file even when tests run as root. + args = _arguments(tmp_path, tmp_path, summary=tmp_path) + assert checker.main(args) == 2 + + +def test_main_appends_markdown_to_existing_summary(tmp_path): + summary = tmp_path / "summary.md" + summary.write_text("Existing summary.\n", encoding="utf-8") + args = _arguments(tmp_path, tmp_path, summary=summary) + assert checker.main(args) == 0 + first = summary.read_text(encoding="utf-8") + assert first.startswith("Existing summary.\n") + assert "## Fix evidence" in first + + (tmp_path / "body.txt").write_text("No reproducer.", encoding="utf-8") + assert checker.main(args) == 1 + second = summary.read_text(encoding="utf-8") + assert second.startswith(first) + assert second.count("## Fix evidence") == 2 + assert "Reproducer:" in second + + (tmp_path / "changes.txt").write_text("unparseable", encoding="utf-8") + assert checker.main(args) == 2 + third = summary.read_text(encoding="utf-8") + assert third.startswith(second) + assert third.count("## Fix evidence") == 3 + assert "Unusable input" in third + + +def _git_env(): + return { + **os.environ, + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_NOSYSTEM": "1", + } + + +def _git(repo_root, *args): + result = subprocess.run( + [ + "git", + "-c", + "user.name=Fix evidence tests", + "-c", + "user.email=tests@example.invalid", + "-c", + "commit.gpgsign=false", + "-c", + "core.hooksPath=/dev/null", + "-c", + "gc.auto=0", + "-c", + "maintenance.auto=false", + *args, + ], + cwd=repo_root, + capture_output=True, + text=True, + timeout=15, + env=_git_env(), + ) + assert result.returncode == 0, result.stdout + result.stderr + # Stripping output would corrupt paths ending in whitespace in NUL diffs. + return result.stdout + + +def _commit(repo_root, message): + _git(repo_root, "add", ".") + _git(repo_root, "commit", "-m", message) + return _git(repo_root, "rev-parse", "HEAD").strip() + + +@pytest.fixture +def git_repo(tmp_path, monkeypatch): + monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull) + monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1") + repo_root = tmp_path / "repo" + repo_root.mkdir() + _git(repo_root, "init", "--initial-branch=fixture", "--template=") + return repo_root + + +def test_nul_delimited_git_diff_preserves_literal_paths(git_repo): + _write(git_repo, "baseline.txt", "Before the change.\n") + base = _commit(git_repo, "Fixture baseline") + paths = ["docs/é\\303.md", "docs/tab\tnewline\ntrailing .md ", 'docs/"quoted".md'] + for path in paths: + _write(git_repo, path, "A documentation change.\n") + head = _commit(git_repo, "Fixture literal paths") + changes = _git( + git_repo, + "-c", + "core.quotePath=false", + "diff", + "--name-status", + "--no-renames", + "-z", + base, + head, + ) + assert set(checker.parse_changes(changes)) == {("A", path) for path in paths} + + +def test_real_git_diff_and_standalone_checker(tmp_path, git_repo, monkeypatch, capsys): + repo_root = git_repo + python_path = "policyengine_uk/tests/test_regression.py" + yaml_path = "policyengine_uk/tests/gov/regression.yaml" + _write(repo_root, python_path, "def test_regression():\n pass\n") + _write(repo_root, yaml_path, "- name: Regression\n period: 2025\n") + base = _commit(repo_root, "Fixture before the fix") + + _write(repo_root, python_path, "def test_regression():\n assert 1 + 1 == 2\n") + _write(repo_root, yaml_path, "- name: Regression\n period: 2026\n") + _write(repo_root, "changelog.d/regression.fixed.md", "- Fixed a regression.\n") + head = _commit(repo_root, "Fixture fix and its evidence") + merge_base = _git(repo_root, "merge-base", base, head).strip() + changes = _git( + repo_root, "diff", "--name-status", "--no-renames", "-z", merge_base, head + ) + assert f"M\0{python_path}\0" in changes + assert f"M\0{yaml_path}\0" in changes + assert "A\0changelog.d/regression.fixed.md\0" in changes + + # The base checkout need not contain any of the PR's changed test files. + _git(repo_root, "checkout", "--detach", base) + + for nodeid, expected in [(f"{python_path}::test_regression", 0), (yaml_path, 1)]: + args = _arguments( + tmp_path / "inputs", + repo_root, + body=f"Reproducer: {nodeid}", + labels="", + changes=changes, + head_sha=head, + ) + code = _standalone_checker(args, repo_root, monkeypatch) + output = capsys.readouterr().out + assert code == expected, output + assert "Fix a regression" in output + if expected: + assert "Reproducer:" in output + + +@pytest.mark.parametrize("target_in_repository", [True, False]) +def test_git_object_mode_rejects_symlinked_test_paths( + tmp_path, git_repo, target_in_repository +): + path = "policyengine_uk/tests/test_fake.py" + target_root = git_repo if target_in_repository else tmp_path + target = _write(target_root, "test_real.py", "def test_regression():\n pass\n") + link = git_repo / path + link.parent.mkdir(parents=True) + link.symlink_to(target) + head = _commit(git_repo, "Fixture symlink reproducer") + report = _report( + git_repo, + f"Reproducer: {path}::test_regression", + [("A", path)], + head_sha=head, + ) + assert not report.passed + assert any("symlink" in message.lower() for message in report.messages) + + +def test_git_object_mode_ignores_fabricated_worktree_evidence( + tmp_path, git_repo, monkeypatch, capsys +): + path = "policyengine_uk/tests/test_fake.py" + _write(git_repo, path, "# No test exists in the committed file.\n") + _write(git_repo, ".github/check_fix_pr.py", "raise SystemExit(0)\n") + (git_repo / "body.md").symlink_to(path) + head = _commit(git_repo, "Fixture metadata symlink") + body = ( + f'"""\nReproducer: {path}::test_regression\n"""\n' + "def test_regression():\n pass\n" + ) + # This recreates the old metadata write so the blob must remain authoritative. + (git_repo / "body.md").write_text(body, encoding="utf-8") + args = _arguments( + tmp_path / "inputs", + git_repo, + body=body, + changes=f"A\0{path}\0A\0.github/check_fix_pr.py\0A\0body.md\0", + head_sha=head, + ) + code = _standalone_checker(args, git_repo, monkeypatch) + output = capsys.readouterr().out + assert code == 1, output + assert "does not define a collectable named test" in output + + +def test_git_object_mode_reads_head_without_importing_pr_code(git_repo): + path = "policyengine_uk/tests/test_regression.py" + marker = git_repo / "pr-code-executed" + _write( + git_repo, + path, + f"from pathlib import Path\nPath({str(marker)!r}).touch()\n" + "def test_regression():\n pass\n", + ) + _write(git_repo, ".github/check_fix_pr.py", "raise SystemExit(0)\n") + head = _commit(git_repo, "Fixture PR code") + (git_repo / path).unlink() + report = _report( + git_repo, + f"Reproducer: {path}::test_regression", + [("A", path)], + head_sha=head, + ) + assert report.passed + assert not marker.exists() + + +def test_git_object_mode_rejects_paths_missing_from_head(git_repo): + _write(git_repo, "baseline.txt", "A committed baseline.\n") + head = _commit(git_repo, "Fixture without test") + path = "policyengine_uk/tests/test_regression.py" + _write(git_repo, path, "def test_regression():\n pass\n") + assert not _report( + git_repo, + f"Reproducer: {path}::test_regression", + [("A", path)], + head_sha=head, + ).passed + + +def test_the_workflow_reports_on_pull_requests_and_merge_groups(): + import yaml + + workflow = yaml.safe_load( + (REPO_ROOT / ".github" / "workflows" / "pr_fix_evidence.yaml").read_text( + encoding="utf-8" + ) + ) + # PyYAML reads the key "on" as True. + triggers = workflow.get("on", workflow.get(True)) + assert set(triggers) == {"pull_request_target", "merge_group"} + assert "paths" not in triggers["pull_request_target"] + assert workflow["permissions"] == {"contents": "read"} + [job] = workflow["jobs"].values() + assert job["name"] == "Fix evidence" + assert "if" not in job + assert job.get("permissions", workflow["permissions"]) == {"contents": "read"} + + steps = job["steps"] + [checkout] = [ + step for step in steps if step.get("uses", "").startswith("actions/checkout@") + ] + assert checkout["if"] == "github.event_name == 'pull_request_target'" + # The event's default checkout supplies the trusted base workflow and checker. + assert "ref" not in checkout.get("with", {}) + # No token stays in .git/config while PR objects are fetched and read. + assert checkout["with"]["persist-credentials"] is False + for step in steps: + if step.get("uses", "").startswith("actions/checkout@"): + reference = str(step.get("with", {}).get("ref", "")) + assert "github.event.pull_request.head.sha" not in reference + assert "github.event.pull_request.head.ref" not in reference + run = step.get("run", "") + assert not re.search(r"\b(?:pip(?:3)?\s+install|uv\s+sync|make)\b", run) + assert not re.search(r"\bgit\s+(?:checkout|switch|reset)\b", run) + + [setup_uv] = [ + step for step in steps if step.get("uses", "").startswith("astral-sh/setup-uv@") + ] + assert setup_uv["uses"] == "astral-sh/setup-uv@v8.1.0" + assert setup_uv["if"] == "github.event_name == 'pull_request_target'" + + [collect] = [step for step in steps if step.get("id") == "evidence"] + assert collect["if"] == "github.event_name == 'pull_request_target'" + assert collect["env"]["PR_NUMBER"] == "${{ github.event.pull_request.number }}" + collect_run = collect["run"] + assert 'mktemp -d "$RUNNER_TEMP/fix-evidence.XXXXXX"' in collect_run + assert "directory=%s\\n" in collect_run + assert '"$GITHUB_OUTPUT"' in collect_run + for filename in ["title.txt", "body.md", "labels.txt", "changes.bin"]: + assert f'> "$evidence_dir/{filename}"' in collect_run + assert ( + 'git fetch --no-tags origin "+refs/pull/${PR_NUMBER}/head:refs/remotes/pr/head"' + in collect_run + ) + assert 'git merge-base "$PR_BASE_SHA" "$PR_HEAD_SHA"' in collect_run + assert "git diff --name-status --no-renames -z" in collect_run + + [check] = [ + step for step in steps if ".github/check_fix_pr.py" in step.get("run", "") + ] + assert check["if"] == "github.event_name == 'pull_request_target'" + assert check.get("working-directory", ".") == "." + assert check["env"]["EVIDENCE_DIR"] == "${{ steps.evidence.outputs.directory }}" + check_run = check["run"] + assert ( + "uv run --no-project --with pyyaml==6.0.2 python .github/check_fix_pr.py" + in check_run + ) + for option, filename in [ + ("title", "title.txt"), + ("body", "body.md"), + ("labels", "labels.txt"), + ("changes", "changes.bin"), + ]: + assert f'--{option}-file "$EVIDENCE_DIR/{filename}"' in check_run + assert '--repo-root . --head-sha "$PR_HEAD_SHA"' in check_run + + [merge_group] = [ + step for step in steps if step.get("if") == "github.event_name == 'merge_group'" + ] + assert merge_group["run"].startswith("echo ")