diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 953e42689..c9ce0a5de 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -386,6 +386,39 @@ jobs: use_pypi: true version: '11.3.1' + # Mutates only the package lines a PR changes and runs the tests covering each (scripts/mutate_changed.py). + # Warn-only: survivors go to the step summary and as annotations; runtime and survivor counts are in the PR that added it. + mutation-diff: + name: Mutation of changed lines + if: github.event_name == 'pull_request' + needs: static + runs-on: ${{ github.repository_owner != 'intel' && 'ubuntu-latest' || vars.LINUX_RUNNER_OFF != 'true' && 'dokimi-ec2' || 'intel-ubuntu-latest' }} + timeout-minutes: 30 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + fetch-depth: 0 + - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 # v7.1.0 + with: + python-version: '3.13' + cache: pip + cache-dependency-path: dokimi/requirements.lock.txt + - name: Install deps + run: pip install --no-deps --require-hashes -r requirements.lock.txt + # No new mutant after 900 s, each at most 120 s: the step ends well inside its 22 minutes. + - name: Mutate changed lines + timeout-minutes: 22 + env: + BASE_REF: ${{ github.event.pull_request.base.ref }} + run: | + python scripts/mutate_changed.py --base "origin/$BASE_REF" --max-mutants 40 --budget 900 \ + --mutant-timeout 120 --json mutation-diff.json --markdown mutation-diff.md + - name: Mutation summary + if: ${{ !cancelled() }} + run: | + if [ -f mutation-diff.md ]; then cat mutation-diff.md; else echo "No report: see the Mutate changed lines log."; fi >> "$GITHUB_STEP_SUMMARY" + # Pins the request a docker-claude arm sends (#354); the golden and what it covers are in # tests/fixtures/fidelity/README.md. A diff here means the npm build or the arm changed. fidelity: diff --git a/.gitignore b/.gitignore index edfa580a0..e16514447 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,8 @@ coverage.json # Written by the CI steps (and `make ci-local`) /dokimi/diff-cover.json /dokimi/diff-cover.md +/dokimi/mutation-diff.json +/dokimi/mutation-diff.md /dokimi/compat.txt /dokimi/outcomes-result.json /dokimi/journeys-result.json diff --git a/AGENTS.md b/AGENTS.md index 93991e3a3..ac0c2dbd6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -144,6 +144,12 @@ finds little: - For each test of changed behaviour, break the code it covers (flip a condition, drop a branch), confirm the test fails, then revert the break. + CI's `Mutation of changed lines` job (`dokimi/scripts/mutate_changed.py`) + does this for the package lines a PR changes, up to 40 mutants, and lists + each surviving mutant in its step summary. It warns, it does not fail: kill + a survivor with an assertion, or say in the description why it is + equivalent. Run it locally with + `python scripts/mutate_changed.py --base origin/main` from `dokimi/`. - For a bug fix, find every other site with the same bug class. Fix the ones within this PR's scope; list the rest in the description or an issue. - Where it applies, run the changed command on a bad input: missing file, diff --git a/dokimi/pyproject.toml b/dokimi/pyproject.toml index 7dc5535ca..b7c626cbb 100644 --- a/dokimi/pyproject.toml +++ b/dokimi/pyproject.toml @@ -136,6 +136,8 @@ extend_exclude = [ "scripts/conda_forge_specs.py", # Run by fuzz.yml, which installs atheris from requirements-fuzz.lock.txt. "fuzz", + # Run by ci.yml's mutation-diff job; coverage comes with pytest-cov from requirements.lock.txt. + "scripts/mutate_changed.py", ] # Tools in the dev extra (pytest and plugins, ruff, mypy, scriv), never imported by # the package. A dev dep imported from the package is DEP004. diff --git a/dokimi/scripts/mutate_changed.py b/dokimi/scripts/mutate_changed.py new file mode 100755 index 000000000..7d86993c3 --- /dev/null +++ b/dokimi/scripts/mutate_changed.py @@ -0,0 +1,478 @@ +#!/usr/bin/env python +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Mutation-test only the lines a PR changes in the ``dokimi`` package. + +The by-hand step in ``AGENTS.md`` ("break the code it covers, confirm the test +fails"), run by CI: a surviving mutant is a changed line that no test notices +breaking. Steps: + +1. Changed lines: ``git diff -U0`` of the working tree against the merge base of BASE and HEAD, over ``dokimi/dokimi/**.py``. +2. Candidate tests: the files under ``tests/`` that import a changed module. +3. Covering tests: one coverage run of those files with per-test contexts + (``--cov-context=test``), one file at a time for at most half of ``--budget``, + maps each changed line to the tests that execute it. +4. Mutants: a small AST mutator on the changed lines only (flip a comparison, + swap ``and``/``or``, negate an ``if``/``while`` condition, return ``None``, + flip a boolean constant, bump an integer, swap ``+``/``-``, drop ``not``), + spliced into the source text so the rest of the file is untouched. + At most ``--max-mutants``, spread round-robin over the changed lines. +5. Each mutant runs the tests covering its line, ``-x``, in a copy of the + tracked tree (subprocess tests see the mutant too; the checkout is never + written). Exit 0 from pytest is a survivor, exit 1 (a test failed) a kill, and + so is a collection error the mutant caused (``_mutant_status``); a timeout and + any other exit code (no test ran, a usage error) get their own status, never a kill. + +A line no candidate test executes is reported as uncovered, not mutated: the +changed-lines coverage gate in ``ci.yml`` owns that. The run stops starting +mutants after ``--budget`` seconds and reports the rest as not run. + + python scripts/mutate_changed.py --base origin/main [--max-mutants 40] + [--json out.json] [--markdown out.md] [--fail-on-survivor] + +Tests that fail without a mutant are left out and listed. Exits 0, or 1 with +``--fail-on-survivor`` when a mutant survives; git failing to diff ``--base`` +exits with its message. +""" + +from __future__ import annotations + +import argparse +import ast +import json +import re +import shutil +import subprocess +import sys +import tempfile +import time +from collections.abc import Iterator +from dataclasses import asdict, dataclass, field +from pathlib import Path + +PROJECT = Path(__file__).resolve().parents[1] +REPO = PROJECT.parent +if str(PROJECT) not in sys.path: # run as a script from CI, not installed + sys.path.insert(0, str(PROJECT)) + +from dokimi import process # noqa: E402 +PKG = "dokimi/dokimi/" +HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@") + +_CMP_SWAP: dict[type, type] = { + ast.Eq: ast.NotEq, ast.NotEq: ast.Eq, ast.Lt: ast.GtE, ast.GtE: ast.Lt, + ast.Gt: ast.LtE, ast.LtE: ast.Gt, ast.In: ast.NotIn, ast.NotIn: ast.In, + ast.Is: ast.IsNot, ast.IsNot: ast.Is, +} + + +@dataclass +class Mutant: + path: str # repo-relative + line: int + op: str + start: tuple[int, int] + end: tuple[int, int] + replacement: str + original: str + status: str = "not run" + tests: list[str] = field(default_factory=list) + seconds: float = 0.0 + + +def changed_lines(base: str) -> dict[str, set[int]]: + """Added or edited line numbers per changed package file, the working tree against the merge base of ``base``. + + Committed and uncommitted edits of tracked files count, as the mutants are cut from the working tree + (untracked files do not: CI checks out a clean tree). + """ + mb = subprocess.run(["git", "merge-base", base, "HEAD"], cwd=REPO, capture_output=True, text=True, encoding="utf-8") + if mb.returncode: + raise SystemExit(f"git merge-base {base} HEAD failed: {mb.stderr.strip()}") + out = subprocess.run( + ["git", "-c", "core.quotePath=false", "diff", "-U0", "--no-color", "--diff-filter=AMR", mb.stdout.strip(), "--", f"{PKG}*.py"], + cwd=REPO, capture_output=True, text=True, encoding="utf-8", + ) + if out.returncode: + raise SystemExit(f"git diff {mb.stdout.strip()} failed: {out.stderr.strip()}") + result: dict[str, set[int]] = {} + path = None + for line in out.stdout.splitlines(): + if line.startswith("+++ "): + path = line[6:] if line.startswith("+++ b/") else None + elif path and (m := HUNK.match(line)): + start, count = int(m.group(1)), int(m.group(2) or "1") + result.setdefault(path, set()).update(range(start, start + count)) + return {p: s for p, s in result.items() if s} + + +def _docstring_lines(tree: ast.AST) -> set[int]: + lines: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) and node.body: + first = node.body[0] + if isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant) and isinstance(first.value.value, str): + lines.update(range(first.lineno, (first.end_lineno or first.lineno) + 1)) + return lines + + +def _span(node: ast.AST) -> tuple[tuple[int, int], tuple[int, int]]: + return (node.lineno, node.col_offset), (node.end_lineno or node.lineno, node.end_col_offset or 0) # type: ignore[attr-defined] + + +def _edits(node: ast.AST, src: str) -> Iterator[tuple[str, ast.AST, str]]: + """(operator name, node whose source span is replaced, replacement text).""" + yield from _logic_edits(node, src) + yield from _value_edits(node) + + +def _logic_edits(node: ast.AST, src: str) -> Iterator[tuple[str, ast.AST, str]]: + if isinstance(node, ast.Compare) and len(node.ops) == 1 and type(node.ops[0]) in _CMP_SWAP: + new = ast.Compare(left=node.left, ops=[_CMP_SWAP[type(node.ops[0])]()], comparators=node.comparators) + yield "flip-compare", node, ast.unparse(new) + elif isinstance(node, ast.BoolOp): + new_op = ast.Or() if isinstance(node.op, ast.And) else ast.And() + yield "swap-boolop", node, ast.unparse(ast.BoolOp(op=new_op, values=node.values)) + elif isinstance(node, (ast.If, ast.While, ast.IfExp)): + seg = ast.get_source_segment(src, node.test) + if seg is not None: + yield "negate-condition", node.test, f"not ({seg})" + elif isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not): + seg = ast.get_source_segment(src, node.operand) + if seg is not None: + yield "drop-not", node, f"({seg})" + + +def _value_edits(node: ast.AST) -> Iterator[tuple[str, ast.AST, str]]: + if isinstance(node, ast.Return) and node.value is not None: + if not (isinstance(node.value, ast.Constant) and node.value.value is None): + yield "return-none", node.value, "None" + elif isinstance(node, ast.Constant) and isinstance(node.value, bool): + yield "flip-bool", node, repr(not node.value) + elif isinstance(node, ast.Constant) and type(node.value) is int: + yield "bump-int", node, repr(node.value + 1) + elif isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Sub)): + if not any(isinstance(s, ast.Constant) and isinstance(s.value, str) for s in (node.left, node.right)): + new_bin = ast.BinOp(left=node.left, op=ast.Sub() if isinstance(node.op, ast.Add) else ast.Add(), right=node.right) + yield "swap-addsub", node, ast.unparse(new_bin) + + +def mutants_for(path: str, src: str, lines: set[int]) -> list[Mutant]: + """Every mutant whose node starts on a changed line, in source order.""" + tree = ast.parse(src) + skip = _docstring_lines(tree) + out: list[Mutant] = [] + for node in ast.walk(tree): + line = getattr(node, "lineno", None) + if line is None or line not in lines or line in skip: + continue + for op, target, replacement in _edits(node, src): + original = ast.get_source_segment(src, target) + if original is None or original == replacement: + continue + start, end = _span(target) + out.append(Mutant(path, line, op, start, end, replacement, original)) + out.sort(key=lambda m: (m.line, m.start, m.op)) + return out + + +def apply(src: str, m: Mutant) -> str: + """``src`` with the mutant's span replaced; columns are UTF-8 byte offsets.""" + lines = src.splitlines(keepends=True) + (sl, sc), (el, ec) = m.start, m.end + head = lines[sl - 1].encode("utf-8")[:sc].decode("utf-8") + tail = lines[el - 1].encode("utf-8")[ec:].decode("utf-8") + return "".join(lines[: sl - 1]) + head + m.replacement + tail + "".join(lines[el:]) + + +def pick(mutants: list[Mutant], cap: int) -> list[Mutant]: + """Round-robin over (file, line) so a long changed function does not take the whole cap.""" + by_line: dict[tuple[str, int], list[Mutant]] = {} + for m in mutants: + by_line.setdefault((m.path, m.line), []).append(m) + queues = list(by_line.values()) + chosen: list[Mutant] = [] + while queues and len(chosen) < cap: + for q in list(queues): + if len(chosen) >= cap: + break + chosen.append(q.pop(0)) + if not q: + queues.remove(q) + return chosen + + +def candidate_tests(paths: list[str], tests_dir: Path) -> list[str]: + """Test files that import a changed module (``dokimi.a.b`` or ``from dokimi.a import b``). + + Files named after a changed module (``test_b*.py``) come first: the coverage pass + runs them in this order and stops at its share of the budget. + """ + specs, leaves = [], [] + for p in paths: + dotted = p[len("dokimi/"):-len(".py")].replace("/", ".").removesuffix(".__init__") + parent, _, leaf = dotted.rpartition(".") + leaves.append(f"test_{leaf}") + specs.append((re.compile(rf"\b{re.escape(dotted)}\b"), parent, leaf)) + found = [] + for f in sorted(tests_dir.glob("test_*.py")): + text = f.read_text(encoding="utf-8", errors="replace") + froms = _from_imports(text) + if any(p.search(text) or (parent, leaf) in froms for p, parent, leaf in specs): + found.append(f"tests/{f.name}") + return sorted(found, key=lambda t: not any(Path(t).stem.startswith(leaf) for leaf in leaves)) + + +def _from_imports(text: str) -> set[tuple[str, str]]: + """(module, name) of each ``from module import name``, however the statement is wrapped; none if it does not parse.""" + try: + tree = ast.parse(text) + except SyntaxError: + return set() + return {(n.module, a.name) for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module for a in n.names} + + +def _write_source(target: Path, text: str) -> None: + # A same-size rewrite in the same second would otherwise run the cached bytecode of the previous text. + target.write_text(text, encoding="utf-8") + for pyc in (target.parent / "__pycache__").glob(f"{target.stem}.*.pyc"): + pyc.unlink() + + +def copy_tree(dest: Path) -> None: + files = subprocess.run(["git", "ls-files", "-z"], cwd=REPO, capture_output=True, check=True).stdout.decode("utf-8") + for rel in filter(None, files.split("\0")): + src = REPO / rel + if src.is_file(): + target = dest / rel + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src, target) + + +def _pytest( + cwd: Path, args: list[str], timeout: float, env: dict[str, str] | None = None +) -> tuple[int, str]: + cmd = [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", "-o", "console_output_style=classic", *args] + # process.run kills the whole process tree on exit or timeout: no test's child outlives its mutant. + # capture_limit=None: the coverage pass parses every FAILED/PASSED line, a truncated summary would lose some. + done = process.run(cmd, timeout=timeout, env=process.Env.inherit(env), cwd=cwd, capture_limit=None) + if done.timed_out: + return -1, "timeout" + return done.returncode, done.stdout + done.stderr + + +def _failed(log: str, ids: set[str]) -> tuple[set[str], list[str]]: + """The ids in ``ids`` that pytest's summary names as failed, and the summary's other names. + + A summary line is ``FAILED - ``, and a node id can hold `` - `` (a parameter), so a line is + attributed to the longest known id it starts with (``f[a]`` is a prefix of the line of ``f[a] - b]``), among + the ids with no ``PASSED`` line of their own when there are any (a message can complete a longer id: ``f[a] - Failed: b]``); + only a name that matches none is cut at the first `` - ``. + """ + named: set[str] = set() + unmatched: list[str] = [] + passed = set(re.findall(r"^PASSED (.+)$", log, re.MULTILINE)) + for line in re.findall(r"^(?:FAILED|ERROR) (.+)$", log, re.MULTILINE): + found = [i for i in ids if line == i or line.startswith(i + " - ")] + match = max([i for i in found if i not in passed] or found, key=len, default=None) + if match is None: + unmatched.append(line.split(" - ", 1)[0]) + else: + named.add(match) + return named, unmatched + + +_PASSED = re.compile(r"\b[1-9]\d* passed\b") # the final pytest line of a run in which a test ran +_COLLECTION_ERROR = "ERROR collecting " # a module that does not import: the mutant broke it (pytest's exit code for this varies) + + +def _mutant_status(rc: int, log: str) -> str: + """1 is a failed test. A collection error (exit 2 or 4 by pytest version) is one too: the file collected unmutated, + so the mutant broke the import. Other exits (internal, usage, nothing collected) say nothing about the mutant.""" + if rc == 0: + return "survived" + if rc == -1: + return "timeout" + if rc == 1 or (rc in (2, 4) and _COLLECTION_ERROR in log): + return "killed" + return "error" + + +def covering_tests( + work: Path, tests: list[str], paths: list[str], budget: float +) -> tuple[dict[str, dict[int, list[str]]], list[str], list[str]]: + """Run each test file with per-test coverage contexts, in order, until ``budget`` seconds pass. + + Returns line -> passing test node ids per changed file, what was left out because + it fails unmutated (a node id, or a file that does not collect: it would kill every + mutant), and the files not run for lack of time. One process per file, not xdist: + pytest-cov's per-test contexts break under xdist workers. + """ + import coverage + + t0 = time.monotonic() + by_file: dict[str, dict[int, list[str]]] = {} + failing: list[str] = [] + data_file = work / "dokimi" / ".coverage" + for i, test in enumerate(tests): + left = budget - (time.monotonic() - t0) + if left <= 0: + return by_file, failing, tests[i:] + data_file.unlink(missing_ok=True) + rc, log = _pytest( + work / "dokimi", ["-rfEp", "--cov=dokimi", "--cov-context=test", "--cov-report=", test], left, + # -rp lists PASSED ids too (without the captured output of passing tests): a test that reads an import-time value has no context on the file's lines. + # sys.monitoring (the default core on 3.14+) records no per-test contexts. + env={"COVERAGE_CORE": "ctrace"}, + ) + if rc == -1: + return by_file, failing, tests[i:] + if rc not in (0, 1, 5) or not data_file.exists(): + failing.append(test) + continue + ran = _PASSED.search(log) is not None + data = coverage.CoverageData(basename=str(data_file)) + data.read() + measured = {Path(f).resolve(): f for f in data.measured_files()} + per_file = {rel: data.contexts_by_lineno(measured[(work / rel).resolve()]) + for rel in paths if (work / rel).resolve() in measured} + executed = {c.rsplit("|", 1)[0] for ctxs in per_file.values() for cs in ctxs.values() for c in cs if c} + passed = set(re.findall(r"^PASSED (.+)$", log, re.MULTILINE)) + # a passing test that errors in teardown is both PASSED and ERROR: its ERROR makes it bad + bad, unmatched = _failed(log, executed | passed) if rc == 1 else (set(), []) + failing += sorted(bad) + unmatched + for rel, ctxs in per_file.items(): + lines = by_file.setdefault(rel, {}) + for ln, cs in ctxs.items(): + ids = {c.rsplit("|", 1)[0] for c in cs if c} + if "" in cs and ran: # run at collection (an import-time constant or default): every test of the file runs it + # with no failure, the file's tests; with one, the tests of the file that passed + ids |= {test} if not (bad or unmatched) else passed - bad + if ids := ids - bad: + lines[ln] = sorted(set(lines.get(ln, [])) | ids) + return by_file, failing, [] + + +def run(args: argparse.Namespace) -> int: + t0 = time.monotonic() + report: dict = {"base": args.base, "max_mutants": args.max_mutants, "files": {}, "mutants": [], "uncovered_lines": {}} + changed = changed_lines(args.base) + all_mutants: list[Mutant] = [] + for path, lines in sorted(changed.items()): + src = (REPO / path).read_text(encoding="utf-8") + try: + ms = mutants_for(path, src, lines) + except SyntaxError: + continue + report["files"][path] = {"changed_lines": len(lines), "mutants_possible": len(ms)} + all_mutants += ms + if not all_mutants: + report["note"] = "no mutable changed lines in the dokimi package" + return _finish(report, args, t0, 0) + + tests = candidate_tests(sorted({m.path for m in all_mutants}), PROJECT / "tests") + report["candidate_tests"] = tests + with tempfile.TemporaryDirectory(prefix="dokimi-mut-") as tmp: + work = Path(tmp) + copy_tree(work) + cover: dict[str, dict[int, list[str]]] = {} + if tests: + cover, failing, skipped = covering_tests( + work, tests, sorted({m.path for m in all_mutants}), args.budget / 2 + ) + report["failing_unmutated"] = failing + report["candidate_tests_not_run"] = skipped + covered = [m for m in all_mutants if cover.get(m.path, {}).get(m.line)] + for m in all_mutants: + if not cover.get(m.path, {}).get(m.line): + report["uncovered_lines"].setdefault(m.path, set()).add(m.line) + chosen = pick(covered, args.max_mutants) + report["mutants_covered"] = len(covered) + for m in chosen: + if time.monotonic() - t0 > args.budget: + break + m.tests = cover[m.path][m.line] + # Every covering test, read by pytest from a file: a widely covered line can outgrow a command line. + tests_file = work / "mutant-tests.txt" + tests_file.write_text("\n".join(m.tests), encoding="utf-8") + target = work / m.path + original = target.read_text(encoding="utf-8") + _write_source(target, apply(original, m)) + start = time.monotonic() + try: + rc, log = _pytest(work / "dokimi", ["-x", "-p", "no:xdist", f"@{tests_file}"], args.mutant_timeout) + finally: + _write_source(target, original) + m.seconds = round(time.monotonic() - start, 1) + m.status = _mutant_status(rc, log) + report["mutants"] = [asdict(m) for m in chosen] + report["uncovered_lines"] = {p: sorted(s) for p, s in report["uncovered_lines"].items()} + survived = sum(m.status == "survived" for m in chosen) + return _finish(report, args, t0, 1 if survived and args.fail_on_survivor else 0) + + +def markdown(report: dict) -> str: + ms = report["mutants"] + count = {s: sum(m["status"] == s for m in ms) for s in ("killed", "timeout", "error", "survived", "not run")} + out = ["## Mutation of changed lines\n"] + if note := report.get("note"): + out.append(f"{note}.\n") + out.append( + f"{len(ms)} mutants of {report.get('mutants_covered', 0)} on covered changed lines (cap {report['max_mutants']}): " + f"{count['killed']} killed, {count['timeout']} timed out, {count['error']} errors, " + f"**{count['survived']} survived**, {count['not run']} not run (time budget). {report['seconds']} s.\n" + ) + survivors = [m for m in ms if m["status"] == "survived"] + if survivors: + out.append("A survivor is a change to that line that the tests covering it do not notice; " + "add an assertion that fails on it, or say in the PR why it is equivalent.\n") + out.append("| Line | Operator | Was | Became | Tests run |\n|---|---|---|---|---:|") + for m in survivors: + was, became = (s.replace("|", "\\|").replace("\n", " ")[:80] for s in (m["original"], m["replacement"])) + out.append(f"| `{m['path']}:{m['line']}` | {m['op']} | `{was}` | `{became}` | {len(m['tests'])} |") + out.append("") + if report["uncovered_lines"]: + lines = "; ".join(f"`{p}`: {', '.join(map(str, ls))}" for p, ls in report["uncovered_lines"].items()) + out.append(f"Changed lines no candidate test executes (not mutated): {lines}\n") + if skipped := report.get("candidate_tests_not_run"): + out.append(f"{len(skipped)} candidate test files not run (coverage pass out of time): " + + ", ".join(f"`{t}`" for t in skipped[:20]) + ("…" if len(skipped) > 20 else "") + "\n") + if failing := report.get("failing_unmutated"): + out.append(f"{len(failing)} candidate tests fail without a mutant and were left out: " + + ", ".join(f"`{t}`" for t in failing[:20]) + ("…" if len(failing) > 20 else "") + "\n") + return "\n".join(out) + + +def _finish(report: dict, args: argparse.Namespace, t0: float, code: int) -> int: + report.setdefault("mutants", []) + report["uncovered_lines"] = {p: sorted(s) for p, s in report.get("uncovered_lines", {}).items()} + report["seconds"] = round(time.monotonic() - t0, 1) + if args.json: + Path(args.json).write_text(json.dumps(report, indent=2), encoding="utf-8") + text = markdown(report) + if args.markdown: + Path(args.markdown).write_text(text, encoding="utf-8") + print(text) + for m in report["mutants"]: + if m["status"] == "survived": + level = "error" if args.fail_on_survivor else "warning" + print(f"::{level} file={m['path']},line={m['line']}::mutant survived: {m['op']} `{m['original']}` -> `{m['replacement']}`") + return code + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("--base", required=True, help="git ref the PR is diffed against (merge base of BASE...HEAD)") + ap.add_argument("--max-mutants", type=int, default=40) + ap.add_argument("--mutant-timeout", type=float, default=180.0, help="seconds per mutant; a mutant that times out is reported as timed out, neither killed nor survived") + ap.add_argument("--budget", type=float, default=1200.0, help="seconds after which no new mutant starts") + ap.add_argument("--json") + ap.add_argument("--markdown") + ap.add_argument("--fail-on-survivor", action="store_true") + return run(ap.parse_args(argv)) + + +if __name__ == "__main__": + process.install_signal_handlers() + sys.exit(main()) diff --git a/dokimi/tests/test_ci_label_triggers.py b/dokimi/tests/test_ci_label_triggers.py index 0ca1005a2..9f49ea451 100644 --- a/dokimi/tests/test_ci_label_triggers.py +++ b/dokimi/tests/test_ci_label_triggers.py @@ -119,4 +119,5 @@ def test_a_label_no_step_reads_neither_cancels_the_run_nor_runs_the_jobs(action) def test_a_push_to_a_pr_supersedes_its_run_and_a_main_push_keeps_its_own(): assert (_group(_pr("synchronize")), _ran(_pr("synchronize"))) == ("ci-512", _PR_JOBS) push = {"github.event_name": "push"} - assert (_group(push), _ran(push)) == ("ci-9001", _PR_JOBS) + # mutation-diff mutates the lines a PR changes against its base: there is no base on a push. + assert (_group(push), _ran(push)) == ("ci-9001", _PR_JOBS - {"mutation-diff"}) diff --git a/dokimi/tests/test_mutate_changed.py b/dokimi/tests/test_mutate_changed.py new file mode 100644 index 000000000..ae98d5aa0 --- /dev/null +++ b/dokimi/tests/test_mutate_changed.py @@ -0,0 +1,568 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for scripts/mutate_changed.py — mutation of a PR's changed lines.""" + +from __future__ import annotations + +import ast +import importlib.util +import json +import os +import signal +import subprocess +import sys +import time +import types +from pathlib import Path + +import pytest + +_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "mutate_changed.py" +_spec = importlib.util.spec_from_file_location("mutate_changed", _SCRIPT) +assert _spec and _spec.loader +mc = importlib.util.module_from_spec(_spec) +sys.modules["mutate_changed"] = mc +_spec.loader.exec_module(mc) + +SRC = '''def f(a, b): + """Doc: a == b.""" + if a == b and a > 0: + return a + 1 + return not b +''' + + +def _ops(lines: set[int]) -> list[tuple[int, str, str]]: + return [(m.line, m.op, mc.apply(SRC, m)) for m in mc.mutants_for("x.py", SRC, lines)] + + +def test_mutants_only_on_changed_lines_and_each_parses() -> None: + got = _ops({3}) + assert {op for _, op, _ in got} == {"flip-compare", "swap-boolop", "negate-condition", "bump-int"} + assert all(line == 3 for line, _, _ in got) + for _, _, text in got: + ast.parse(text) + assert text.splitlines()[3:] == SRC.splitlines()[3:] # lines after the mutant untouched + texts = {(op, t.splitlines()[2]) for _, op, t in got} + assert texts == { + ("flip-compare", " if a != b and a > 0:"), + ("flip-compare", " if a == b and a <= 0:"), + ("swap-boolop", " if a == b or a > 0:"), + ("negate-condition", " if not (a == b and a > 0):"), + ("bump-int", " if a == b and a > 1:"), + } + + +def test_return_addsub_and_not_mutants() -> None: + got = {(line, op, t.splitlines()[line - 1].strip()) for line, op, t in _ops({4, 5})} + assert got == { + (4, "return-none", "return None"), + (4, "swap-addsub", "return a - 1"), + (4, "bump-int", "return a + 2"), + (5, "return-none", "return None"), + (5, "drop-not", "return (b)"), + } + + +def test_docstring_lines_are_not_mutated() -> None: + assert _ops({2}) == [] + + +def test_string_concatenation_is_not_swapped() -> None: + assert mc.mutants_for("x.py", "y = 'a' + b\n", {1}) == [] + + +def test_apply_uses_utf8_byte_columns() -> None: + src = "x = 'é' if a < b else 0\n" + (m,) = [m for m in mc.mutants_for("x.py", src, {1}) if m.op == "flip-compare"] + assert mc.apply(src, m) == "x = 'é' if a >= b else 0\n" + + +def test_pick_spreads_the_cap_over_lines() -> None: + ms = mc.mutants_for("x.py", SRC, {3, 4, 5}) + chosen = mc.pick(ms, 3) + assert [m.line for m in chosen] == [3, 4, 5] + assert len(mc.pick(ms, 100)) == len(ms) + + +def test_candidate_tests_match_dotted_and_from_imports(tmp_path: Path) -> None: + (tmp_path / "test_a.py").write_text("from dokimi.subjects import scorecard\n", encoding="utf-8") + (tmp_path / "test_b.py").write_text("import dokimi.subjects.scorecard as s\n", encoding="utf-8") + (tmp_path / "test_c.py").write_text("from dokimi.subjects import scorecards\n", encoding="utf-8") + assert mc.candidate_tests(["dokimi/dokimi/subjects/scorecard.py"], tmp_path) == ["tests/test_a.py", "tests/test_b.py"] + (tmp_path / "test_scorecard_x.py").write_text("from dokimi.subjects import scorecard\n", encoding="utf-8") + assert mc.candidate_tests(["dokimi/dokimi/subjects/scorecard.py"], tmp_path)[0] == "tests/test_scorecard_x.py" + + +def _git(cwd: Path, *args: str) -> str: + return subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True, text=True).stdout + + +def test_changed_lines_reads_added_and_edited_lines(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + _git(tmp_path, "init", "-q") + _git(tmp_path, "config", "user.email", "t@example.com") + _git(tmp_path, "config", "user.name", "t") + (pkg / "m.py").write_text("a = 1\nb = 2\nc = 3\n", encoding="utf-8") + (tmp_path / "other.py").write_text("x = 1\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "base") + (pkg / "m.py").write_text("a = 1\nb = 5\nc = 3\nd = 4\n", encoding="utf-8") + (tmp_path / "other.py").write_text("x = 2\n", encoding="utf-8") + _git(tmp_path, "commit", "-qam", "head") + monkeypatch.setattr(mc, "REPO", tmp_path) + assert mc.changed_lines("HEAD~1") == {"dokimi/dokimi/m.py": {2, 4}} + with pytest.raises(SystemExit, match="git merge-base"): + mc.changed_lines("no-such-ref") + + +def test_markdown_lists_survivors_and_uncovered() -> None: + m = mc.Mutant("dokimi/dokimi/x.py", 3, "flip-compare", (3, 4), (3, 10), "a != b", "a == b", status="survived", tests=["t"]) + k = mc.Mutant("dokimi/dokimi/x.py", 4, "return-none", (4, 8), (4, 9), "None", "a", status="killed") + k2 = mc.Mutant("dokimi/dokimi/x.py", 5, "return-none", (5, 8), (5, 9), "None", "a", status="killed") + e = mc.Mutant("dokimi/dokimi/x.py", 6, "return-none", (6, 8), (6, 9), "None", "a", status="error") + text = mc.markdown({"mutants": [vars(m), vars(k), vars(k2), vars(e)], "mutants_covered": 4, "max_mutants": 40, + "uncovered_lines": {"dokimi/dokimi/x.py": [7]}, "seconds": 1.0}) + assert "2 killed" in text and "**1 survived**" in text and "1 errors" in text + assert "| `dokimi/dokimi/x.py:3` | flip-compare | `a == b` | `a != b` | 1 |" in text + assert "`dokimi/dokimi/x.py`: 7" in text + + +def _mutation_run( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, src: str, tests: str, cover: dict[int, list[str]], *flags: str +) -> tuple[int, dict]: + """Run the script on a repo whose head adds ``dokimi/zz.py`` and its tests; returns (exit code, JSON report). + + ``cover`` (zz.py line -> covering test ids) stands in for the coverage pass: the suite's + test environments have no coverage package, and covering_tests has its own test below. + """ + if not (tmp_path / ".git").exists(): + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + (tmp_path / "dokimi" / "tests").mkdir() + (pkg / "__init__.py").write_text("", encoding="utf-8") + _git(tmp_path, "init", "-q") + _git(tmp_path, "config", "user.email", "t@example.com") + _git(tmp_path, "config", "user.name", "t") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "base") + (pkg / "zz.py").write_text(src, encoding="utf-8") + (tmp_path / "dokimi" / "tests" / "test_zz.py").write_text(tests, encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "head") + monkeypatch.setattr(mc, "REPO", tmp_path) + monkeypatch.setattr(mc, "PROJECT", tmp_path / "dokimi") + monkeypatch.delenv("PYTHONDONTWRITEBYTECODE", raising=False) + # ci/test.py sets PYTHONSAFEPATH=1: pytest in the copy would then import the installed dokimi, not its zz.py. + monkeypatch.delenv("PYTHONSAFEPATH", raising=False) + + def covering(work: Path, files: list[str], paths: list[str], budget: float) -> tuple[dict, list[str], list[str]]: + mc._pytest(work / "dokimi", files, budget) # imports the unmutated zz.py first, as the coverage pass does + return {"dokimi/dokimi/zz.py": cover}, [], [] + + monkeypatch.setattr(mc, "covering_tests", covering) + out = tmp_path / "report.json" + rc = mc.main(["--base", "HEAD~1", "--budget", "100", "--mutant-timeout", "30", "--json", str(out), *flags]) + return rc, json.loads(out.read_text(encoding="utf-8")) + + +def _statuses(report: dict) -> dict[tuple[int, str], str]: + return {(m["line"], m["replacement"]): m["status"] for m in report["mutants"]} + + +def test_a_same_size_mutant_in_the_same_second_runs_its_own_source(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + real = mc._pytest + + def same_second(cwd: Path, args: list[str], timeout: float, env: dict[str, str] | None = None) -> tuple[int, str]: + os.utime(cwd / "dokimi" / "zz.py", (1_700_000_000, 1_700_000_000)) + return real(cwd, args, timeout, env) + + monkeypatch.setattr(mc, "_pytest", same_second) + src = "def a(x):\n return x == 1\n" + tests = "from dokimi.zz import a\n\n\ndef test_a():\n assert a(1) is True\n assert a(2) is False\n" + _, report = _mutation_run(tmp_path, monkeypatch, src, tests, {2: ["tests/test_zz.py::test_a"]}) + assert _statuses(report) == {(2, "x != 1"): "killed", (2, "2"): "killed", (2, "None"): "killed"} + + +def test_a_mutant_only_the_last_of_many_covering_tests_catches_is_killed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + src = "def a(x):\n return x > 3\n" + tests = ( + "import pytest\n\nfrom dokimi.zz import a\n\n\n" + "@pytest.mark.parametrize('n', range(30))\ndef test_a_weak(n):\n assert a(10 + n)\n\n\n" + "def test_z_strict():\n assert a(4) and not a(3)\n" + ) + weak = [f"tests/test_zz.py::test_a_weak[{n}]" for n in range(30)] + _, report = _mutation_run(tmp_path, monkeypatch, src, tests, {2: sorted([*weak, "tests/test_zz.py::test_z_strict"])}) + assert _statuses(report) == {(2, "x <= 3"): "killed", (2, "4"): "killed", (2, "None"): "killed"} + assert {len(m["tests"]) for m in report["mutants"]} == {31} + + +def test_survivors_kills_and_the_fail_on_survivor_exit_code(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + src = "def a(x):\n return x == 1\n\n\ndef b(x):\n return x == 2\n" + tests = ( + "from dokimi.zz import a, b\n\n\n" + "def test_a():\n assert a(1) is True\n assert a(2) is False\n\n\n" + "def test_b_is_only_called():\n b(2)\n" + ) + cover = {2: ["tests/test_zz.py::test_a"], 6: ["tests/test_zz.py::test_b_is_only_called"]} + rc, report = _mutation_run(tmp_path, monkeypatch, src, tests, cover, "--fail-on-survivor") + assert _statuses(report) == { + (2, "x != 1"): "killed", (2, "2"): "killed", (2, "None"): "killed", + (6, "x != 2"): "survived", (6, "3"): "survived", (6, "None"): "survived", + } + assert rc == 1 + assert _mutation_run(tmp_path, monkeypatch, src, tests, cover)[0] == 0 + + +def test_covering_tests_leaves_out_a_test_that_fails_unmutated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + zz = tmp_path / "dokimi" / "dokimi" / "zz.py" + zz.parent.mkdir(parents=True) + contexts = {2: {"tests/test_zz.py::test_a|run"}, + 6: {"tests/test_zz.py::test_b|run", "tests/test_zz.py::test_broken|run"}} + + class Data: + def __init__(self, basename: str) -> None: + assert basename == str(tmp_path / "dokimi" / ".coverage") + + def read(self) -> None: + pass + + def measured_files(self) -> list[str]: + return [str(zz)] + + def contexts_by_lineno(self, f: str) -> dict[int, set[str]]: + return contexts + + def pytest_run(cwd: Path, args: list[str], timeout: float, env: dict[str, str] | None = None) -> tuple[int, str]: + (cwd / ".coverage").write_text("", encoding="utf-8") + log = "FAILED tests/test_zz.py::test_broken - assert False\n" + return (1, log) if args[-1] == "tests/test_zz.py" else (2, "ERROR collecting") + + monkeypatch.setitem(sys.modules, "coverage", types.SimpleNamespace(CoverageData=Data)) + monkeypatch.setattr(mc, "_pytest", pytest_run) + by_file, failing, not_run = mc.covering_tests( + tmp_path, ["tests/test_zz.py", "tests/test_bad.py"], ["dokimi/dokimi/zz.py"], 60 + ) + assert by_file == {"dokimi/dokimi/zz.py": {2: ["tests/test_zz.py::test_a"], 6: ["tests/test_zz.py::test_b"]}} + assert failing == ["tests/test_zz.py::test_broken", "tests/test_bad.py"] + assert not_run == [] + + +def _alive(pid: int) -> bool: + try: + return Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").rsplit(")", 1)[1].split()[0] != "Z" + except FileNotFoundError: + return False + + +@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="reads /proc") +def test_a_mutant_that_times_out_leaves_no_process_of_its_tests_behind(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pid_file = tmp_path / "grandchild.pid" + monkeypatch.setenv("ZZ_PID_FILE", str(pid_file)) + src = "def a(x):\n return x == 1\n" + tests = ( + "import os\nimport subprocess\nimport sys\nimport time\n\nfrom dokimi.zz import a\n\n\n" + "def test_a():\n if a(1) is not True:\n" + " child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(120)'])\n" + " with open(os.environ['ZZ_PID_FILE'], 'w') as f:\n f.write(str(child.pid))\n" + " time.sleep(120)\n" + ) + _, report = _mutation_run(tmp_path, monkeypatch, src, tests, {2: ["tests/test_zz.py::test_a"]}, "--max-mutants", "1", "--mutant-timeout", "5") + pid = int(pid_file.read_text(encoding="utf-8")) + try: + assert [m["status"] for m in report["mutants"]] == ["timeout"] + deadline = time.monotonic() + 5 + while _alive(pid) and time.monotonic() < deadline: + time.sleep(0.1) + assert not _alive(pid) + finally: + if _alive(pid): + os.kill(pid, signal.SIGKILL) + + +def test_candidate_tests_match_a_wrapped_from_import(tmp_path: Path) -> None: + (tmp_path / "test_a.py").write_text("from dokimi.subjects import (\n other,\n scorecard,\n)\n", encoding="utf-8") + (tmp_path / "test_b.py").write_text("from dokimi.subjects import \\\n scorecard\n", encoding="utf-8") + assert mc.candidate_tests(["dokimi/dokimi/subjects/scorecard.py"], tmp_path) == ["tests/test_a.py", "tests/test_b.py"] + + +def test_changed_lines_counts_an_uncommitted_edit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + _git(tmp_path, "init", "-q") + _git(tmp_path, "config", "user.email", "t@example.com") + _git(tmp_path, "config", "user.name", "t") + (pkg / "m.py").write_text("a = 1\nb = 2\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "base") + (pkg / "m.py").write_text("a = 1\nb = 5\n", encoding="utf-8") + _git(tmp_path, "commit", "-qam", "head") + (pkg / "m.py").write_text("a = 1\nb = 5\nc = 6\n", encoding="utf-8") + monkeypatch.setattr(mc, "REPO", tmp_path) + assert mc.changed_lines("HEAD~1") == {"dokimi/dokimi/m.py": {2, 3}} + + +def test_covering_tests_maps_an_import_time_line_to_the_file_tests(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pytest.importorskip("pytest_cov") + monkeypatch.delenv("PYTHONSAFEPATH", raising=False) + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "zz.py").write_text("FLAG = True\n\n\ndef f(x=3):\n return FLAG\n", encoding="utf-8") + (tmp_path / "dokimi" / "tests").mkdir() + (tmp_path / "dokimi" / "tests" / "test_zz.py").write_text( + "from dokimi.zz import f\n\n\ndef test_f():\n assert f()\n", encoding="utf-8") + by_file, failing, not_run = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 120) + assert (failing, not_run) == ([], []) + assert by_file == {"dokimi/dokimi/zz.py": {1: ["tests/test_zz.py"], 4: ["tests/test_zz.py"], 5: ["tests/test_zz.py::test_f"]}} + + +@pytest.mark.parametrize("tests, cover", [ + ("from dokimi.zz import a\n\n\ndef test_a():\n assert a(1) is True\n", {2: ["tests/test_zz.py::test_missing"]}), + ("from dokimi.zz import a\n", {2: ["tests/test_zz.py"]}), +], ids=["node-id-not-found", "nothing-collected"]) +def test_a_mutant_no_test_runs_for_is_an_error_not_a_kill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, tests: str, cover: dict[int, list[str]] +) -> None: + rc, report = _mutation_run(tmp_path, monkeypatch, "def a(x):\n return x == 1\n", tests, cover, "--fail-on-survivor") + assert rc == 0 + assert report["mutants"] and {m["status"] for m in report["mutants"]} == {"error"} + + +def test_a_mutant_is_restored_to_the_source_after_its_run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + writes: list[tuple[str, str]] = [] + real = mc._write_source + + def spy(target: Path, text: str) -> None: + writes.append((target.name, text)) + real(target, text) + + monkeypatch.setattr(mc, "_write_source", spy) + src = "def a(x):\n return x == 1\n" + tests = "from dokimi.zz import a\n\n\ndef test_a():\n assert a(1) is True\n" + _mutation_run(tmp_path, monkeypatch, src, tests, {2: ["tests/test_zz.py::test_a"]}, "--max-mutants", "1") + assert len(writes) == 2 and writes[-1] == ("zz.py", src) + + +def test_fail_on_survivor_passes_when_every_mutant_is_killed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + tests = "from dokimi.zz import a\n\n\ndef test_a():\n assert a(1) is True\n assert a(2) is False\n" + rc, report = _mutation_run(tmp_path, monkeypatch, "def a(x):\n return x == 1\n", tests, + {2: ["tests/test_zz.py::test_a"]}, "--fail-on-survivor") + assert rc == 0 + assert report["mutants"] and {m["status"] for m in report["mutants"]} == {"killed"} + + +def test_a_survivor_is_a_warning_unless_fail_on_survivor(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + src = "def b(x):\n return x == 2\n" + tests = "from dokimi.zz import b\n\n\ndef test_b_is_only_called():\n b(2)\n" + cover = {2: ["tests/test_zz.py::test_b_is_only_called"]} + _mutation_run(tmp_path, monkeypatch, src, tests, cover) + out = capsys.readouterr().out + assert "::warning file=dokimi/dokimi/zz.py,line=2::mutant survived" in out and "::error" not in out + _mutation_run(tmp_path, monkeypatch, src, tests, cover, "--fail-on-survivor") + out = capsys.readouterr().out + assert "::error file=dokimi/dokimi/zz.py,line=2::mutant survived" in out + + +def test_no_mutable_changed_line_exits_clean(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(mc, "changed_lines", lambda base: {}) + assert mc.main(["--base", "HEAD", "--fail-on-survivor"]) == 0 + + +def test_a_mutant_that_breaks_the_import_is_killed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + src = "ITEMS = [0, 1]\nFIRST = ITEMS[1]\n\n\ndef a(x):\n return x == 1\n" + tests = "from dokimi.zz import a\n\n\ndef test_a():\n assert a(1) is True\n" + _, report = _mutation_run(tmp_path, monkeypatch, src, tests, {2: ["tests/test_zz.py::test_a"]}) + assert report["mutants"] and {m["status"] for m in report["mutants"]} == {"killed"} + + +@pytest.mark.parametrize("summary, attributed", [("1 skipped in 0.01s", False), ("1 passed in 0.01s", True)]) +def test_an_import_time_line_is_attributed_only_when_a_test_ran( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, summary: str, attributed: bool +) -> None: + zz = tmp_path / "dokimi" / "dokimi" / "zz.py" + zz.parent.mkdir(parents=True) + contexts = {1: {""}, 4: {"tests/test_zz.py::test_a|run"}} + + class Data: + def __init__(self, basename: str) -> None: + pass + + def read(self) -> None: + pass + + def measured_files(self) -> list[str]: + return [str(zz)] + + def contexts_by_lineno(self, f: str) -> dict[int, set[str]]: + return contexts + + def pytest_run(cwd: Path, args: list[str], timeout: float, env: dict[str, str] | None = None) -> tuple[int, str]: + (cwd / ".coverage").write_text("", encoding="utf-8") + return 0, f"=== {summary} ===" + + monkeypatch.setitem(sys.modules, "coverage", types.SimpleNamespace(CoverageData=Data)) + monkeypatch.setattr(mc, "_pytest", pytest_run) + by_file, failing, not_run = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 60) + assert (failing, not_run) == ([], []) + expected = {1: ["tests/test_zz.py"], 4: ["tests/test_zz.py::test_a"]} if attributed else {4: ["tests/test_zz.py::test_a"]} + assert by_file == {"dokimi/dokimi/zz.py": expected} + + +def _stub_covering(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, contexts: dict[int, set[str]], rc: int, log: str) -> None: + zz = tmp_path / "dokimi" / "dokimi" / "zz.py" + zz.parent.mkdir(parents=True, exist_ok=True) + + class Data: + def __init__(self, basename: str) -> None: + pass + + def read(self) -> None: + pass + + def measured_files(self) -> list[str]: + return [str(zz)] + + def contexts_by_lineno(self, f: str) -> dict[int, set[str]]: + return contexts + + def pytest_run(cwd: Path, args: list[str], timeout: float, env: dict[str, str] | None = None) -> tuple[int, str]: + (cwd / ".coverage").write_text("", encoding="utf-8") + return rc, log + + monkeypatch.setitem(sys.modules, "coverage", types.SimpleNamespace(CoverageData=Data)) + monkeypatch.setattr(mc, "_pytest", pytest_run) + + +def test_a_failing_id_with_a_dash_in_its_parameter_is_matched_whole(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + node = "tests/test_zz.py::test_f[a - b]" + _stub_covering(monkeypatch, tmp_path, {2: {f"{node}|run"}}, 1, f"FAILED {node} - AssertionError\n") + by_file, failing, _ = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 60) + assert by_file == {"dokimi/dokimi/zz.py": {}} + assert failing == [node] + + +def test_an_import_time_line_in_a_mixed_file_is_credited_to_its_passing_tests(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + contexts = {1: {""}, 4: {"tests/test_zz.py::test_ok|run", "tests/test_zz.py::test_bad|run"}} + log = "FAILED tests/test_zz.py::test_bad - assert False\nPASSED tests/test_zz.py::test_ok\n=== 1 failed, 1 passed in 0.01s ===\n" + _stub_covering(monkeypatch, tmp_path, contexts, 1, log) + by_file, failing, _ = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 60) + assert by_file == {"dokimi/dokimi/zz.py": {1: ["tests/test_zz.py::test_ok"], 4: ["tests/test_zz.py::test_ok"]}} + assert failing == ["tests/test_zz.py::test_bad"] + + +def test_an_import_time_line_in_a_mixed_file_is_credited_to_every_passing_test(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # test_const passes and reads the constant without executing a line of the file: it has no context there + contexts = {1: {""}, 4: {"tests/test_zz.py::test_fn|run"}} + log = ("FAILED tests/test_zz.py::test_bad - assert False\nPASSED tests/test_zz.py::test_const\n" + "PASSED tests/test_zz.py::test_fn\n=== 1 failed, 2 passed in 0.01s ===\n") + _stub_covering(monkeypatch, tmp_path, contexts, 1, log) + by_file, _, _ = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 60) + assert by_file["dokimi/dokimi/zz.py"][1] == ["tests/test_zz.py::test_const", "tests/test_zz.py::test_fn"] + + +def test_a_passing_test_that_errors_in_teardown_is_not_credited_to_an_import_time_line(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + contexts = {1: {""}, 4: {"tests/test_zz.py::test_fn|run"}} + log = ("FAILED tests/test_zz.py::test_bad - assert False\nPASSED tests/test_zz.py::test_const\n" + "PASSED tests/test_zz.py::test_fn\nERROR tests/test_zz.py::test_const - teardown boom\n" + "=== 1 failed, 2 passed, 1 error in 0.01s ===\n") + _stub_covering(monkeypatch, tmp_path, contexts, 1, log) + by_file, _, _ = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 60) + assert by_file["dokimi/dokimi/zz.py"][1] == ["tests/test_zz.py::test_fn"] + + +def test_a_passing_id_that_prefixes_a_failing_id_is_not_failed() -> None: + passing, failing = "tests/test_zz.py::test_f[a]", "tests/test_zz.py::test_f[a] - b]" + bad, unmatched = mc._failed(f"FAILED {failing} - AssertionError\n", {passing, failing}) + assert (bad, unmatched) == ({failing}, []) + bad, _ = mc._failed(f"FAILED {passing} - AssertionError\n", {passing, failing}) + assert bad == {passing} + + +def test_changed_lines_diffs_against_the_merge_base_not_the_base_tip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + _git(tmp_path, "init", "-q", "-b", "main") + _git(tmp_path, "config", "user.email", "t@example.com") + _git(tmp_path, "config", "user.name", "t") + (pkg / "m.py").write_text("a = 1\nb = 2\nc = 3\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "base") + _git(tmp_path, "checkout", "-q", "-b", "feature") + (pkg / "m.py").write_text("a = 1\nb = 2\nc = 30\n", encoding="utf-8") + _git(tmp_path, "commit", "-qam", "feature") + _git(tmp_path, "checkout", "-q", "main") + (pkg / "m.py").write_text("a = 10\nb = 2\nc = 3\n", encoding="utf-8") + _git(tmp_path, "commit", "-qam", "main moves on") + _git(tmp_path, "checkout", "-q", "feature") + monkeypatch.setattr(mc, "REPO", tmp_path) + assert mc.changed_lines("main") == {"dokimi/dokimi/m.py": {3}} + + +def test_changed_lines_reads_a_non_ascii_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + _git(tmp_path, "init", "-q") + _git(tmp_path, "config", "user.email", "t@example.com") + _git(tmp_path, "config", "user.name", "t") + _git(tmp_path, "config", "core.quotePath", "true") + (tmp_path / "other.py").write_text("x = 1\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "base") + (pkg / "café.py").write_text("FLAG = True\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-qm", "head") + monkeypatch.setattr(mc, "REPO", tmp_path) + assert mc.changed_lines("HEAD~1") == {"dokimi/dokimi/café.py": {1}} + + +@pytest.mark.parametrize("rc, log, status", [ + (0, "", "survived"), + (-1, "timeout", "timeout"), + (1, "", "killed"), + (2, "ERROR collecting x", "killed"), + (4, "ERROR collecting x", "killed"), + (3, "ERROR collecting x", "error"), + (2, "", "error"), + (4, "", "error"), + (5, "", "error"), +]) +def test_mutant_status_by_exit_code(rc: int, log: str, status: str) -> None: + assert mc._mutant_status(rc, log) == status + + +def test_the_coverage_pass_asks_for_passed_ids(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + pytest.importorskip("pytest_cov") + monkeypatch.delenv("PYTHONSAFEPATH", raising=False) + pkg = tmp_path / "dokimi" / "dokimi" + pkg.mkdir(parents=True) + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "zz.py").write_text("FLAG = True\n\n\ndef f():\n return FLAG\n", encoding="utf-8") + (tmp_path / "dokimi" / "tests").mkdir() + (tmp_path / "dokimi" / "tests" / "test_zz.py").write_text( + "from dokimi.zz import FLAG\n\n\ndef test_flag():\n assert FLAG\n\n\ndef test_bad():\n assert False\n", + encoding="utf-8") + by_file, failing, _ = mc.covering_tests(tmp_path, ["tests/test_zz.py"], ["dokimi/dokimi/zz.py"], 120) + assert failing == ["tests/test_zz.py::test_bad"] + assert by_file["dokimi/dokimi/zz.py"][1] == ["tests/test_zz.py::test_flag"] + + +def test_pytest_output_is_not_truncated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, object] = {} + + def run(argv: object, **kw: object) -> object: + seen.update(kw) + return types.SimpleNamespace(timed_out=False, returncode=0, stdout="", stderr="") + + monkeypatch.setattr(mc.process, "run", run) + mc._pytest(tmp_path, [], 5) + assert seen["capture_limit"] is None + + +def test_a_failure_message_that_completes_a_longer_passing_id_is_not_read_as_its_failure() -> None: + failing, passing = "tests/test_x.py::test_f[a]", "tests/test_x.py::test_f[a] - Failed: b]" + log = f"PASSED {passing}\nFAILED {failing} - Failed: b]\n" + bad, unmatched = mc._failed(log, {failing, passing}) + assert (bad, unmatched) == ({failing}, [])