From ba4065f3f1df276331b7815e72acea99aa64c584 Mon Sep 17 00:00:00 2001 From: "Robert C. Martin" Date: Sat, 3 Oct 2026 17:48:32 -0500 Subject: [PATCH] Stop shelling generated test commands and fail closed on git and coverage. A package path is one argument. Git status uses NUL records and returns git's status. A failed coverage command leaves reports unread, coverage keeps the longest path match, and site walks no longer recurse. --- src/mutator/cli.py | 147 ++++++++++++++++++++++++++++++++------- src/mutator/coverage.py | 58 +++++++++++---- src/mutator/engine.py | 20 ++++-- src/mutator/runner.py | 82 ++++++++++++++-------- src/mutator/sites.py | 22 ++++-- src/mutator/workers.py | 9 +-- tests/test_boundaries.py | 122 ++++++++++++++++++++++++-------- tests/test_cli.py | 27 +++++++ tests/test_coverage.py | 41 +++++++++++ tests/test_runner.py | 62 ++++++++++++----- tests/test_sites.py | 9 +++ 11 files changed, 473 insertions(+), 126 deletions(-) diff --git a/src/mutator/cli.py b/src/mutator/cli.py index e23d346..9a556b5 100644 --- a/src/mutator/cli.py +++ b/src/mutator/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os import subprocess import sys from dataclasses import dataclass, field @@ -13,7 +14,44 @@ from mutator.engine import mutate_file, restore_backups, scan_file from mutator.report import format_results, format_site_log from mutator.runner import CommandRunner -HELP = """\ + +def _skipped_directory_text() -> str: + """The directories the walker skips, named the same way in --help.""" + + names = { + ".clj-kondo", + ".git", + ".hg", + ".idea", + ".metrics", + ".svn", + ".venv", + "__pycache__", + "__tests__", + "build", + "coverage", + "dist", + "node_modules", + "out", + "spec", + "specs", + "target", + "test", + "testdata", + "tests", + "venv", + "vendor", + } + try: + crapper = ensure_crapper() + names = set(crapper.discover.SKIP_DIRS) | set(crapper.discover.TEST_DIRS) + except ImportError: + pass + ordered = sorted(names) + return ", ".join(ordered[:-1]) + ", and " + ordered[-1] + + +HELP = f"""\ Usage: mutator [options] [path-or-filter ...] Discover mutation sites in Clojure, Java, Go, TypeScript, Rust, and Python, @@ -67,7 +105,7 @@ Arguments: path contains this text are mutated. With no paths, source files under the project root are mutated. Directories -named test, tests, spec, specs, vendor, node_modules, and target are skipped. +named {_skipped_directory_text()} are skipped. The default, once a snapshot exists, reruns survivors and sites in functions whose text changed. Killed mutants in unchanged functions are kept. @@ -254,26 +292,55 @@ def parse_args(argv: list[str] | None = None) -> Options: return options -def _changed_files(root: Path) -> list[Path]: +class GitStatusError(Exception): + """git status could not be read. `code` is git's own status.""" + + def __init__(self, code: int, message: str): + super().__init__(message) + self.code = code + self.message = message + + +def _git(root: Path, *args: str) -> bytes: result = subprocess.run( - ["git", "status", "--porcelain"], - cwd=root, + ["git", "-C", os.fspath(root), *args], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, check=False, - capture_output=True, - text=True, ) if result.returncode != 0: - print(result.stderr.strip() or "git status failed", file=sys.stderr) - return [] + message = os.fsdecode(result.stderr).strip() or "git status failed" + raise GitStatusError(result.returncode, message) + return result.stdout + + +def _changed_files(root: Path) -> list[Path]: + """Added and modified files under `root`, as paths from the repo root. + + `-z` keeps spaces and non-ASCII names intact. A missing path, including a + deletion, is left out. A rename is the new file only. + """ + + toplevel = Path(os.fsdecode(_git(root, "rev-parse", "--show-toplevel")).strip()) + status = _git( + root, + "status", + "--porcelain", + "-z", + "--no-renames", + "--untracked-files=all", + "--", + ".", + ) + root_resolved = root.resolve() found: list[Path] = [] - for line in result.stdout.splitlines(): - if len(line) < 4: + for entry in status.split(b"\0"): + if len(entry) <= 3: continue - path_text = line[3:].strip() - if " -> " in path_text: - path_text = path_text.split(" -> ", 1)[1] - path_text = path_text.strip('"') - found.append((root / path_text).resolve()) + path = (toplevel / os.fsdecode(entry[3:])).resolve() + if not path.is_file() or not path.is_relative_to(root_resolved): + continue + found.append(path) return found @@ -292,8 +359,13 @@ def _positionals(root: Path, args: list[str]) -> tuple[list[Path], list[str]]: def _changed_sources(root: Path, crapper) -> list[Path]: + skip = set(crapper.discover.SKIP_DIRS) | set(crapper.discover.TEST_DIRS) + root_resolved = root.resolve() files = [] for path in _changed_files(root): + relative = path.relative_to(root_resolved) + if not skip.isdisjoint(relative.parts): + continue if crapper.discover.language_of(path) is None: continue if crapper.discover.is_test_file(path): @@ -358,13 +430,32 @@ def select_files(options: Options) -> list[Path]: return sorted(set(_inside_root(_matching(files, filters), root)), key=lambda item: item.as_posix()) -def _prepare_coverage(options: Options, root: Path, files: list[Path]) -> None: - if options.no_coverage or options.use_existing_coverage: - return - if options.scan: - return +def _prepare_coverage(options: Options, root: Path, files: list[Path]) -> int: + """Run coverage generation. A failed command's code is returned as-is. + + Reports already on disk are left unread when the command fails, so a stale + 100% report cannot score the run. + """ + + if options.no_coverage or options.use_existing_coverage or options.scan: + return 0 + if options.coverage_command: + result = CommandRunner(verbose=options.verbose).run(options.coverage_command, root, None) + if result.code != 0: + print( + f"Coverage command exited {result.code}. Reports already on disk will not be read.", + file=sys.stderr, + ) + tail = "\n".join(result.output.splitlines()[-20:]) + if tail: + print(tail, file=sys.stderr) + return result.code + return 0 crapper = ensure_crapper() - crapper.runners.run_coverage(root, files, options.coverage_command) + code = crapper.runners.run_coverage(root, files, None) + if isinstance(code, int) and code != 0: + return code + return 0 def _coverage_for(options: Options, root: Path, path: Path) -> set[int] | None: @@ -437,7 +528,7 @@ def _finish(baseline_failed: bool, survived: bool) -> int: def _mutate_files(options: Options, root: Path, files: list[Path]) -> int: runner = CommandRunner(verbose=options.verbose) - baselines: dict[tuple[str, str], tuple[bool, float, str]] = {} + baselines: dict[tuple[tuple[str, ...], str], tuple[bool, float, str]] = {} forms = [] written: list[str] = [] baseline_failed = False @@ -478,13 +569,19 @@ def run(argv: list[str] | None = None) -> int: print(missing, file=sys.stderr) return 2 root = options.project_root.resolve() - files = select_files(options) + try: + files = select_files(options) + except GitStatusError as exc: + print(exc.message, file=sys.stderr) + return exc.code if not files: print("No source files to mutate.") return 0 if not options.scan: _restore(root) - _prepare_coverage(options, root, files) + coverage = _prepare_coverage(options, root, files) + if coverage != 0: + return coverage if options.scan: return _scan(options, root, files) return _mutate_files(options, root, files) diff --git a/src/mutator/coverage.py b/src/mutator/coverage.py index c158e76..586f865 100644 --- a/src/mutator/coverage.py +++ b/src/mutator/coverage.py @@ -29,18 +29,32 @@ def _exact(normalized: dict, candidates: list[str]): return None -def _suffix_hit(key: str, candidates: list[str], suffix_match) -> bool: - for candidate in candidates: - if suffix_match(key, candidate) or suffix_match(candidate, key): - return True - return False +def _parts(path: str) -> tuple[str, ...]: + return tuple(part for part in path.split("/") if part) -def _by_suffix(normalized: dict, candidates: list[str], suffix_match): - for key, value in normalized.items(): - if _suffix_hit(key, candidates, suffix_match): - return value - return None +def _rank(key_parts: tuple[str, ...], sources: list[tuple[str, ...]]) -> tuple[int, int] | None: + """How well one report key names one of the source paths. + + Rank `(0, extra)` is a key that ends with the source and may add a prefix. + Rank `(1, missing)` is a shorter key that the source path ends with. The + smaller rank is the longer shared suffix. Two different keys at the same + rank are ambiguous. + """ + + best: tuple[int, int] | None = None + for source in sources: + if not source or not key_parts: + continue + if len(key_parts) >= len(source) and key_parts[-len(source) :] == source: + rank = (0, len(key_parts) - len(source)) + elif len(source) > len(key_parts) and source[-len(key_parts) :] == key_parts: + rank = (1, len(source) - len(key_parts)) + else: + continue + if best is None or rank < best: + best = rank + return best def _lookup(index: dict, source_path: str): @@ -53,7 +67,26 @@ def _lookup(index: dict, source_path: str): found = _exact(normalized, candidates) if found is not None: return found - return _by_suffix(normalized, candidates, crapper.coverage.suffix_match) + sources: list[tuple[str, ...]] = [] + for candidate in candidates: + parts = _parts(candidate) + if parts and parts not in sources: + sources.append(parts) + best_rank: tuple[int, int] | None = None + best: list[tuple[tuple[str, ...], object]] = [] + for key, value in normalized.items(): + key_parts = _parts(key) + rank = _rank(key_parts, sources) + if rank is None: + continue + if best_rank is None or rank < best_rank: + best_rank = rank + best = [(key_parts, value)] + elif rank == best_rank: + best.append((key_parts, value)) + if len({parts for parts, _value in best}) != 1: + return None + return best[0][1] def _covered_from_pairs(lines: dict[int, tuple[int, int]] | None) -> set[int] | None: @@ -131,8 +164,7 @@ def _jacoco_index(root: Path) -> dict[str, set[int]]: def _go_lines(profile, source_path: str) -> set[int] | None: if not profile: return None - crapper = ensure_crapper() - segments = crapper.coverage._profile_segments(profile, source_path) + segments = _lookup(profile, source_path) if segments is None: return None covered: set[int] = set() diff --git a/src/mutator/engine.py b/src/mutator/engine.py index 9312d2b..e529424 100644 --- a/src/mutator/engine.py +++ b/src/mutator/engine.py @@ -10,7 +10,7 @@ from mutator.functions import form_digests, form_spans, sites_in_file from mutator.metrics import load_history, source_key, write_results from mutator.model import FormResult, RunResult, Site from mutator.report import format_scan -from mutator.runner import CommandRunner, test_plan +from mutator.runner import Command, CommandRunner, display_command, test_plan from mutator.workers import run_mutants @@ -180,7 +180,15 @@ def _warn_many(file_key: str, selected, mutation_warning: int) -> None: ) -def _remember_baseline(baselines, cache_key, runner, command: str, cwd: Path) -> None: +def _command_key(command: Command) -> tuple[str, ...]: + """A hashable identity. A list cannot be a dict key.""" + + if isinstance(command, str): + return ("sh", command) + return ("argv", *command) + + +def _remember_baseline(baselines, cache_key, runner, command: Command, cwd: Path) -> None: if cache_key in baselines: return baseline = runner.run(command, cwd, None) @@ -191,8 +199,8 @@ def _remember_baseline(baselines, cache_key, runner, command: str, cwd: Path) -> baselines[cache_key] = (True, baseline.seconds, "") -def _baseline_failure(file_key: str, command: str, tail: str) -> RunResult: - message = f"Baseline failed for {file_key}: {command}" +def _baseline_failure(file_key: str, command: Command, tail: str) -> RunResult: + message = f"Baseline failed for {file_key}: {display_command(command)}" if tail: message = f"{message}\n{tail}" return RunResult(path=file_key, forms=[], written=[], baseline_failed=True, baseline_message=message) @@ -215,7 +223,7 @@ def _apply_selected( if not selected: return None command, cwd = test_plan(root, path, language, test_command) - cache_key = (command, str(cwd)) + cache_key = (_command_key(command), str(cwd)) _remember_baseline(baselines, cache_key, runner, command, cwd) ok, seconds, tail = baselines[cache_key] if not ok: @@ -264,7 +272,7 @@ def mutate_file( test_command: str | None, timeout_factor: float, mutation_warning: int, - baselines: dict[tuple[str, str], tuple[bool, float, str]], + baselines: dict[tuple[tuple[str, ...], str], tuple[bool, float, str]], max_workers: int | None = None, ) -> RunResult: """Mutate one file and write its namespaces into `.metrics/mutate`.""" diff --git a/src/mutator/runner.py b/src/mutator/runner.py index 86f829a..6cde33a 100644 --- a/src/mutator/runner.py +++ b/src/mutator/runner.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +import shlex import signal import subprocess import sys @@ -10,6 +11,8 @@ import time from dataclasses import dataclass from pathlib import Path +Command = str | list[str] + @dataclass(frozen=True) class CommandResult: @@ -59,27 +62,46 @@ def _prefer_worker_sources(cwd: Path, environment: dict) -> None: environment["PYTHONPATH"] = os.pathsep.join(entries) +def display_command(command: Command) -> str: + """A shell-looking rendering. A list is quoted so it can be pasted.""" + + if isinstance(command, str): + return command + return " ".join(shlex.quote(part) for part in command) + + class CommandRunner: def __init__(self, verbose: bool = False): self.verbose = verbose - def run(self, command: str, cwd: Path, timeout: float | None) -> CommandResult: + def run(self, command: Command, cwd: Path, timeout: float | None) -> CommandResult: + """Run `command` in `cwd`. + + A list is an argument vector and does not go through a shell, so a + path with spaces or metacharacters stays one argument. A string is a + command the user typed in `--test-command` or `--coverage-command`. + """ + if self.verbose: - print(f"+ ({cwd}) {command}", file=sys.stderr) + print(f"+ ({cwd}) {display_command(command)}", file=sys.stderr) started = time.monotonic() environment = os.environ.copy() environment["PYTHONDONTWRITEBYTECODE"] = "1" _prefer_worker_sources(cwd, environment) - process = subprocess.Popen( - command, - cwd=cwd, - shell=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - start_new_session=True, - text=True, - env=environment, - ) + try: + process = subprocess.Popen( + command, + cwd=cwd, + shell=isinstance(command, str), + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + start_new_session=True, + text=True, + env=environment, + ) + except OSError as exc: + seconds = time.monotonic() - started + return CommandResult(code=127, timed_out=False, seconds=seconds, output=str(exc)) try: output, _err = process.communicate(timeout=timeout) code = process.returncode if process.returncode is not None else 1 @@ -112,20 +134,20 @@ def _read(path: Path) -> str: return "" -def _clojure_command(directory: Path) -> str: +def _clojure_command(directory: Path) -> list[str]: deps = directory / "deps.edn" bb = directory / "bb.edn" if bb.is_file() and not deps.is_file(): text = _read(bb) if "spec" in text: - return "bb spec --tag ~no-mutate" - return "bb test" + return ["bb", "spec", "--tag", "~no-mutate"] + return ["bb", "test"] text = _read(deps) if ":spec" in text and "speclj" in text: - return "clj -M:spec --tag ~no-mutate" + return ["clj", "-M:spec", "--tag", "~no-mutate"] if ":spec" in text: - return "clj -M:spec" - return "clj -M:test" + return ["clj", "-M:spec"] + return ["clj", "-M:test"] def _project_interpreter(directory: Path) -> str: @@ -142,7 +164,7 @@ def _project_interpreter(directory: Path) -> str: return sys.executable -def _python_command(directory: Path) -> str: +def _python_command(directory: Path) -> list[str]: interpreter = _project_interpreter(directory) pyproject = _read(directory / "pyproject.toml") if ( @@ -151,12 +173,16 @@ def _python_command(directory: Path) -> str: or "pytest" in pyproject or (directory / "setup.cfg").is_file() and "pytest" in _read(directory / "setup.cfg") ): - return f"{interpreter} -m pytest" - return f"{interpreter} -m unittest discover" + return [interpreter, "-m", "pytest"] + return [interpreter, "-m", "unittest", "discover"] -def test_plan(root: Path, source: Path, language: str, override: str | None) -> tuple[str, Path]: - """The test command and the directory it runs in.""" +def test_plan(root: Path, source: Path, language: str, override: str | None) -> tuple[Command, Path]: + """The test command and the directory it runs in. + + A user override stays a shell string. Every generated command is an + argument vector, so a package path is one argument. + """ root = root.resolve() if override: @@ -166,18 +192,18 @@ def test_plan(root: Path, source: Path, language: str, override: str | None) -> return _clojure_command(directory), directory if language == "java": directory = nearest(source, "pom.xml", root) or root - return "mvn -q test -DexcludeTags=no-mutate", directory + return ["mvn", "-q", "test", "-DexcludeTags=no-mutate"], directory if language == "go": directory = nearest(source, "go.mod", root) or root relative = source.resolve().parent.relative_to(directory.resolve()) package = "." if relative == Path(".") else "./" + relative.as_posix() - return f"go test -count=1 {package}", directory + return ["go", "test", "-count=1", package], directory if language == "typescript": directory = nearest(source, "package.json", root) or root - return "npm test", directory + return ["npm", "test"], directory if language == "rust": directory = nearest(source, "Cargo.toml", root) or root - return "cargo test", directory + return ["cargo", "test"], directory if language == "python": directory = ( nearest(source, "pyproject.toml", root) @@ -186,4 +212,4 @@ def test_plan(root: Path, source: Path, language: str, override: str | None) -> or root ) return _python_command(directory), directory - return "false", root + return ["false"], root diff --git a/src/mutator/sites.py b/src/mutator/sites.py index 4bfee1e..26a1c11 100644 --- a/src/mutator/sites.py +++ b/src/mutator/sites.py @@ -227,12 +227,15 @@ def _clojure_node(node, data: bytes, found: list[RawSite]) -> None: def _visit_clojure(node, data: bytes, found: list[RawSite]) -> None: - _clojure_node(node, data, found) - for child in node.children: - _visit_clojure(child, data, found) + # An explicit stack: a 1,200-deep form must not raise RecursionError. + stack = [node] + while stack: + current = stack.pop() + _clojure_node(current, data, found) + stack.extend(reversed(current.children)) -def _visit(node, data: bytes, language: str, found: list[RawSite]) -> None: +def _visit_node(node, data: bytes, language: str, found: list[RawSite]) -> None: parent = node.parent.type if node.parent is not None else "" if node.child_count == 0: token = _text(data, node) @@ -257,8 +260,15 @@ def _visit(node, data: bytes, language: str, found: list[RawSite]) -> None: elif node.type == "bool_lit" and _text(data, node) in _BOOLEANS: text = _text(data, node) _add(found, node, text, _BOOLEANS[text], "boolean") - for child in node.children: - _visit(child, data, language, found) + + +def _visit(node, data: bytes, language: str, found: list[RawSite]) -> None: + # An explicit stack: a 1,200-term expression must not raise RecursionError. + stack = [node] + while stack: + current = stack.pop() + _visit_node(current, data, language, found) + stack.extend(reversed(current.children)) def discover_raw(source: str, language: str, path: str) -> list[RawSite]: diff --git a/src/mutator/workers.py b/src/mutator/workers.py index bff3bc1..b1f154c 100644 --- a/src/mutator/workers.py +++ b/src/mutator/workers.py @@ -18,6 +18,7 @@ from queue import Empty, Queue from threading import Lock from mutator.model import Site +from mutator.runner import Command from mutator.sites import apply_site # Build output and the worker tree itself must not be shared. A link to @@ -299,7 +300,7 @@ def _execute( relative: str, original: bytes, runner, - command: str, + command: Command, cwd: Path, timeout: float, file_key: str, @@ -332,7 +333,7 @@ def _drain( relative: str, original: bytes, runner, - command: str, + command: Command, cwd: Path, timeout: float, file_key: str, @@ -373,7 +374,7 @@ def _run_all( relative: str, original: bytes, runner, - command: str, + command: Command, cwd: Path, timeout: float, file_key: str, @@ -412,7 +413,7 @@ def run_mutants( sites: list[Site], max_workers: int | None, runner, - command: str, + command: Command, cwd: Path, timeout: float, file_key: str, diff --git a/tests/test_boundaries.py b/tests/test_boundaries.py index 0daa4cd..080f29b 100644 --- a/tests/test_boundaries.py +++ b/tests/test_boundaries.py @@ -6,6 +6,7 @@ alone when no input can tell the operators apart. """ import hashlib +import subprocess import sys import threading from pathlib import Path @@ -13,6 +14,7 @@ from pathlib import Path import pytest from mutator.cli import ( + GitStatusError, _changed_files, _is_source, _prepare_coverage, @@ -21,7 +23,7 @@ from mutator.cli import ( run, select_files, ) -from mutator.coverage import _candidates, _jacoco_index, _suffix_hit, covered_lines +from mutator.coverage import _candidates, _jacoco_index, covered_lines from mutator.crapper_link import ensure_crapper from mutator.edn import loads from mutator.engine import ( @@ -246,7 +248,6 @@ def test_grammar_guards_and_non_head_clojure_symbols(tmp_path): def test_coverage_indexes_use_package_names_form_counts_and_exact_paths(tmp_path): assert _candidates(lambda value: value, "") == [] assert _candidates(lambda _value: "same", "src/app.py") == ["same"] - assert _suffix_hit("demo/Board.java", ["Other.java"], lambda left, right: left.endswith(right)) is False report = tmp_path / "target" / "site" / "jacoco" / "jacoco.xml" report.parent.mkdir(parents=True) @@ -288,7 +289,7 @@ def test_commands_follow_the_nearest_project_and_a_killed_process(tmp_path): both.mkdir() (both / "bb.edn").write_text("{:tasks {test (clojure)}}\n", encoding="utf-8") (both / "deps.edn").write_text("{:deps {}}\n", encoding="utf-8") - assert _clojure_command(both) == "clj -M:test" + assert _clojure_command(both) == ["clj", "-M:test"] root = tmp_path / "proj" sub = root / "svc" @@ -299,28 +300,28 @@ def test_commands_follow_the_nearest_project_and_a_killed_process(tmp_path): clojure.write_text("(defn place [] 1)\n", encoding="utf-8") command, directory = command_for(root, clojure, "clojure", None) assert directory == sub - assert command == "bb test" + assert command == ["bb", "test"] (sub / "pom.xml").write_text("\n", encoding="utf-8") java = sub / "src" / "A.java" java.write_text("class A { int place(){ return 1; } }\n", encoding="utf-8") command, directory = command_for(root, java, "java", None) assert directory == sub - assert command == "mvn -q test -DexcludeTags=no-mutate" + assert command == ["mvn", "-q", "test", "-DexcludeTags=no-mutate"] (sub / "package.json").write_text("{}\n", encoding="utf-8") typescript = sub / "src" / "a.ts" typescript.write_text("export function place(){ return 1; }\n", encoding="utf-8") command, directory = command_for(root, typescript, "typescript", None) assert directory == sub - assert command == "npm test" + assert command == ["npm", "test"] (sub / "Cargo.toml").write_text("[package]\nname='demo'\n", encoding="utf-8") rust = sub / "src" / "lib.rs" rust.write_text("fn place() -> i32 { 1 }\n", encoding="utf-8") command, directory = command_for(root, rust, "rust", None) assert directory == sub - assert command == "cargo test" + assert command == ["cargo", "test"] (sub / "go.mod").write_text("module example.com/demo\n", encoding="utf-8") go = sub / "pkg" / "widget.go" @@ -328,14 +329,14 @@ def test_commands_follow_the_nearest_project_and_a_killed_process(tmp_path): go.write_text("package pkg\nfunc Run() int { return 1 }\n", encoding="utf-8") command, directory = command_for(root, go, "go", None) assert directory == sub - assert command == "go test -count=1 ./pkg" + assert command == ["go", "test", "-count=1", "./pkg"] (sub / "pyproject.toml").write_text("[project]\ndependencies=['pytest']\n", encoding="utf-8") python = sub / "app.py" python.write_text("def place():\n return 1\n", encoding="utf-8") command, directory = command_for(root, python, "python", None) assert directory == sub - assert command.endswith("-m pytest") + assert command[-2:] == ["-m", "pytest"] result = CommandRunner().run(f"{sys.executable} -c 'import time; time.sleep(30)'", tmp_path, 0.4) assert result.timed_out is True @@ -398,28 +399,93 @@ def test_help_and_conflicts_use_different_streams(capsys): assert captured.out == "" -def test_git_status_keeps_short_names_renames_and_the_failure_text(monkeypatch, tmp_path, capsys): - class Result: - def __init__(self, code, out="", err=""): - self.returncode = code - self.stdout = out - self.stderr = err +def _init_repo(path: Path) -> None: + subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True) + subprocess.run(["git", "config", "user.email", "dev@example.com"], cwd=path, check=True) + subprocess.run(["git", "config", "user.name", "Dev"], cwd=path, check=True) - def fail(*args, **kwargs): - return Result(1, err="boom\n") + +def _commit(path: Path, message: str) -> None: + subprocess.run(["git", "add", "-A"], cwd=path, check=True, capture_output=True) + subprocess.run(["git", "commit", "-m", message], cwd=path, check=True, capture_output=True) + + +def test_git_status_keeps_real_names_and_fails_closed(tmp_path, capsys): + outside = tmp_path / "outside" + outside.mkdir() + code = run(["--changed", "--root", str(outside)]) + captured = capsys.readouterr() + assert code == 128 + assert "not a git repository" in captured.err + assert "No source files" not in captured.out + + repo = tmp_path / "repo" + repo.mkdir() + _init_repo(repo) + source = "def place(x):\n return x > 0\n" + keep = repo / "src" / "café.py" + keep.parent.mkdir() + keep.write_text(source, encoding="utf-8") + nested = repo / "fresh" / "nested" / "new.py" + nested.parent.mkdir(parents=True) + nested.write_text(source, encoding="utf-8") + spaced = repo / "src" / "my file.py" + spaced.write_text(source, encoding="utf-8") + gone = repo / "gone.py" + gone.write_text(source, encoding="utf-8") + old = repo / "src" / "old.py" + old.write_text(source, encoding="utf-8") + _commit(repo, "base") + keep.write_text(source + "\n", encoding="utf-8") + nested.write_text(source + "\n", encoding="utf-8") + spaced.write_text(source + "\n", encoding="utf-8") + gone.unlink() + subprocess.run( + ["git", "mv", "src/old.py", "src/new.py"], cwd=repo, check=True, capture_output=True + ) + skipped = repo / "target" / "skip.py" + skipped.parent.mkdir() + skipped.write_text(source, encoding="utf-8") + + found = set(_changed_files(repo)) + assert keep.resolve() in found + assert nested.resolve() in found + assert spaced.resolve() in found + assert (repo / "src" / "new.py").resolve() in found + assert gone.resolve() not in found + assert old.resolve() not in found + + chosen = select_files(parse_args(["--root", str(repo), "--changed"])) + assert skipped.resolve() not in chosen + assert keep.resolve() in chosen + code = run(["--scan", "--no-coverage", "--root", str(repo), "--changed"]) + assert code == 0 + assert "café.py" in capsys.readouterr().out + + +def test_changed_from_a_subdirectory_stays_inside_it(tmp_path): + repo = tmp_path / "repo" + sub = repo / "sub" + (sub / "src").mkdir(parents=True) + (repo / "src").mkdir() + _init_repo(repo) + below = sub / "src" / "below.py" + above = repo / "src" / "above.py" + body = "def place(x):\n return x > 0\n" + below.write_text(body, encoding="utf-8") + above.write_text(body, encoding="utf-8") + chosen = select_files(parse_args(["--root", str(sub), "--changed"])) + assert chosen == [below.resolve()] + + +def test_an_empty_git_error_uses_the_fallback_text(monkeypatch, tmp_path): + def fail(args, **kwargs): + return subprocess.CompletedProcess(args, 128, b"", b" \n") monkeypatch.setattr("mutator.cli.subprocess.run", fail) - assert _changed_files(tmp_path) == [] - assert capsys.readouterr().err.strip() == "boom" - - def status(*args, **kwargs): - return Result(0, out="?? a\nR src/old.py -> src/new.py\n") - - monkeypatch.setattr("mutator.cli.subprocess.run", status) - found = _changed_files(tmp_path) - assert (tmp_path / "a").resolve() in found - assert (tmp_path / "src" / "new.py").resolve() in found - assert (tmp_path / "src" / "old.py").resolve() not in found + with pytest.raises(GitStatusError, match="git status failed") as caught: + _changed_files(tmp_path) + assert caught.value.code == 128 def test_selection_joins_the_root_and_skips_unknown_files(tmp_path, capsys): diff --git a/tests/test_cli.py b/tests/test_cli.py index f118ca2..aaa08ad 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -42,6 +42,8 @@ def test_help_and_conflicting_flags(): help_options = parse_args(["--help"]) assert help_options.exit_code == 0 assert "uml-viewer" in help_options.message + assert "__pycache__" in help_options.message + assert "testdata" in help_options.message conflict = parse_args(["--mutate-all", "--since-last-run", "src/demo.py"]) assert conflict.exit_code == 1 assert run(["--help"]) == 0 @@ -92,3 +94,28 @@ def test_real_python_mutants_update_the_snapshot(tmp_path, capsys): assert ":namespace \"demo\"" in text assert ":id \"defn/add\"" in text assert path.read_text(encoding="utf-8").startswith("def add") + + +def test_a_failed_coverage_command_does_not_read_the_report(tmp_path, capsys): + path = tmp_path / "src" / "app.py" + path.parent.mkdir() + path.write_text("def place(x):\n return x > 0\n", encoding="utf-8") + report = tmp_path / "target" / "coverage" / "python" / "lcov.info" + report.parent.mkdir(parents=True) + report.write_text("SF:src/app.py\nDA:1,1\nDA:2,1\nend_of_record\n", encoding="utf-8") + code = run( + [ + "--root", + str(tmp_path), + "--coverage-command", + "exit 3", + "--test-command", + "false", + "src/app.py", + ] + ) + captured = capsys.readouterr() + assert code == 3 + assert "will not be read" in captured.err + assert "Baseline failed" not in captured.err + assert path.read_text(encoding="utf-8").startswith("def place") diff --git a/tests/test_coverage.py b/tests/test_coverage.py index de2d5a2..c31eb4e 100644 --- a/tests/test_coverage.py +++ b/tests/test_coverage.py @@ -55,3 +55,44 @@ def test_lcov_marks_executed_lines(tmp_path): source.write_text("def place():\n return 1\n", encoding="utf-8") assert covered_lines(tmp_path, source, "python") == {1} assert covered_lines(tmp_path, source, "clojure") == {1} + + +def test_lcov_does_not_let_a_shorter_path_steal_hits(tmp_path): + report = tmp_path / "target" / "coverage" / "python" / "lcov.info" + report.parent.mkdir(parents=True) + report.write_text( + "SF:b/app.py\nDA:2,1\nend_of_record\n" + "SF:a/b/app.py\nDA:1,1\nend_of_record\n", + encoding="utf-8", + ) + short = tmp_path / "b" / "app.py" + long = tmp_path / "a" / "b" / "app.py" + short.parent.mkdir(parents=True) + long.parent.mkdir(parents=True) + short.write_text("def place():\n return 1\n", encoding="utf-8") + long.write_text("def place():\n return 1\n", encoding="utf-8") + assert covered_lines(tmp_path, long, "python") == {1} + assert covered_lines(tmp_path, short, "python") == {2} + + +def test_go_profile_does_not_cross_files(tmp_path): + profile = tmp_path / "target" / "coverage" / "go" / "coverage.out" + profile.parent.mkdir(parents=True) + profile.write_text( + "mode: set\n" + "b/widget.go:2.1,2.2 1 1\n" + "a/b/widget.go:3.1,3.2 1 0\n", + encoding="utf-8", + ) + short = tmp_path / "b" / "widget.go" + long = tmp_path / "a" / "b" / "widget.go" + short.parent.mkdir(parents=True) + long.parent.mkdir(parents=True) + short.write_text("package b\nfunc Run() {}\n", encoding="utf-8") + long.write_text("package b\nfunc Run() {}\n", encoding="utf-8") + assert covered_lines(tmp_path, short, "go") == {2} + assert covered_lines(tmp_path, long, "go") == set() + + profile.write_text("mode: set\na/b/widget.go:2.1,2.2 1 1\n", encoding="utf-8") + assert covered_lines(tmp_path, short, "go") is None + assert covered_lines(tmp_path, long, "go") == {2} diff --git a/tests/test_runner.py b/tests/test_runner.py index 6f40b03..796f2c9 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -1,6 +1,7 @@ import sys from mutator.runner import CommandRunner, _clojure_command, _python_command, nearest +from mutator.runner import test_plan as plan_command def test_nearest_walks_up_to_the_marker(tmp_path): @@ -19,29 +20,29 @@ def test_clojure_commands(tmp_path): bb = tmp_path / "bb-spec" bb.mkdir() (bb / "bb.edn").write_text('{:tasks {spec "spec"}}', encoding="utf-8") - assert _clojure_command(bb) == "bb spec --tag ~no-mutate" + assert _clojure_command(bb) == ["bb", "spec", "--tag", "~no-mutate"] bb_test = tmp_path / "bb-test" bb_test.mkdir() (bb_test / "bb.edn").write_text("{:tasks {test (clojure \"-M:test\")}}", encoding="utf-8") - assert _clojure_command(bb_test) == "bb test" + assert _clojure_command(bb_test) == ["bb", "test"] spec = tmp_path / "spec" spec.mkdir() (spec / "deps.edn").write_text( "{:aliases {:spec {:extra-deps {speclj/speclj {}}}}}", encoding="utf-8" ) - assert _clojure_command(spec) == "clj -M:spec --tag ~no-mutate" + assert _clojure_command(spec) == ["clj", "-M:spec", "--tag", "~no-mutate"] spec_only = tmp_path / "spec-only" spec_only.mkdir() (spec_only / "deps.edn").write_text("{:aliases {:spec {}}}", encoding="utf-8") - assert _clojure_command(spec_only) == "clj -M:spec" + assert _clojure_command(spec_only) == ["clj", "-M:spec"] plain = tmp_path / "plain" plain.mkdir() (plain / "deps.edn").write_text("{:deps {}}", encoding="utf-8") - assert _clojure_command(plain) == "clj -M:test" + assert _clojure_command(plain) == ["clj", "-M:test"] def _executable(path): @@ -55,15 +56,20 @@ def test_python_command_uses_an_absolute_project_interpreter(tmp_path): dot = project / ".venv" / "bin" / "python" _executable(dot) (project / "pytest.ini").write_text("[pytest]\n", encoding="utf-8") - assert _python_command(project) == f"{dot.absolute()} -m pytest" + assert _python_command(project) == [str(dot.absolute()), "-m", "pytest"] other = tmp_path / "other" plain = other / "venv" / "bin" / "python" _executable(plain) - assert _python_command(other) == f"{plain.absolute()} -m unittest discover" + assert _python_command(other) == [str(plain.absolute()), "-m", "unittest", "discover"] _executable(other / ".venv" / "bin" / "python") - assert _python_command(other) == f"{(other / '.venv' / 'bin' / 'python').absolute()} -m unittest discover" + assert _python_command(other) == [ + str((other / ".venv" / "bin" / "python").absolute()), + "-m", + "unittest", + "discover", + ] def test_python_command_keeps_the_virtualenv_symlink(tmp_path): @@ -74,7 +80,7 @@ def test_python_command_keeps_the_virtualenv_symlink(tmp_path): link.parent.mkdir(parents=True) link.symlink_to(target) (project / "pytest.ini").write_text("[pytest]\n", encoding="utf-8") - assert _python_command(project) == f"{link.absolute()} -m pytest" + assert _python_command(project) == [str(link.absolute()), "-m", "pytest"] def test_python_command_falls_back_when_the_virtualenv_cannot_run(tmp_path): @@ -82,37 +88,61 @@ def test_python_command_falls_back_when_the_virtualenv_cannot_run(tmp_path): binary = project / ".venv" / "bin" / "python" binary.parent.mkdir(parents=True) binary.write_text("", encoding="utf-8") - assert _python_command(project) == f"{sys.executable} -m unittest discover" + assert _python_command(project) == [sys.executable, "-m", "unittest", "discover"] missing = tmp_path / "missing" missing.mkdir() - assert _python_command(missing) == f"{sys.executable} -m unittest discover" + assert _python_command(missing) == [sys.executable, "-m", "unittest", "discover"] def test_python_commands(tmp_path): pytest_dir = tmp_path / "py" pytest_dir.mkdir() (pytest_dir / "pytest.ini").write_text("[pytest]\n", encoding="utf-8") - assert _python_command(pytest_dir).endswith("-m pytest") + assert _python_command(pytest_dir)[-2:] == ["-m", "pytest"] unit = tmp_path / "unit" unit.mkdir() - assert _python_command(unit).endswith("-m unittest discover") + assert _python_command(unit)[-3:] == ["-m", "unittest", "discover"] cfg = tmp_path / "cfg" cfg.mkdir() (cfg / "setup.cfg").write_text("[tool:pytest]\n", encoding="utf-8") - assert _python_command(cfg).endswith("-m pytest") + assert _python_command(cfg)[-2:] == ["-m", "pytest"] project = tmp_path / "proj" project.mkdir() (project / "pyproject.toml").write_text("[project]\ndependencies=['pytest']\n", encoding="utf-8") - assert _python_command(project).endswith("-m pytest") + assert _python_command(project)[-2:] == ["-m", "pytest"] conf = tmp_path / "conf" conf.mkdir() (conf / "conftest.py").write_text("", encoding="utf-8") - assert _python_command(conf).endswith("-m pytest") + assert _python_command(conf)[-2:] == ["-m", "pytest"] + + +def test_a_go_package_path_is_one_argument(tmp_path): + root = tmp_path / "proj" + package = root / "a;echo no" + package.mkdir(parents=True) + (root / "go.mod").write_text("module example.com/demo\n", encoding="utf-8") + source = package / "widget.go" + source.write_text("package main\nfunc Run() int { return 1 }\n", encoding="utf-8") + command, directory = plan_command(root, source, "go", None) + assert command == ["go", "test", "-count=1", "./a;echo no"] + assert directory == root + echoed = CommandRunner().run(["/bin/echo", command[-1]], directory, None) + assert echoed.code == 0 + assert echoed.output.strip() == "./a;echo no" + + spaced = root / "my dir" / "widget.go" + spaced.parent.mkdir() + spaced.write_text("package main\nfunc Run() int { return 1 }\n", encoding="utf-8") + command, _directory = plan_command(root, spaced, "go", None) + assert command == ["go", "test", "-count=1", "./my dir"] + override, override_dir = plan_command(root, source, "go", "go test ./...") + assert override == "go test ./..." + assert override_dir == root.resolve() def test_a_worker_overlay_is_imported_ahead_of_the_environment(tmp_path): diff --git a/tests/test_sites.py b/tests/test_sites.py index d4d3192..498b006 100644 --- a/tests/test_sites.py +++ b/tests/test_sites.py @@ -1,6 +1,7 @@ from pathlib import Path from mutator.functions import form_id, is_private, project_functions, sites_in_file +from mutator.sites import discover_raw def _sites(tmp_path: Path, relative: str, source: str): @@ -194,3 +195,11 @@ export function mount(app) { def test_a_quoted_nullish_operator_is_not_a_site(tmp_path): source = 'export function place() {\n return "a ?? b?.c";\n}\n' assert _sites(tmp_path, "src/demo/app.ts", source) == [] + + +def test_a_deep_expression_is_walked(tmp_path): + expr = " + ".join(["1"] * 1200) + python = f"def place():\n return {expr}\n" + assert discover_raw(python, "python", "app.py") + go = f"package p\nfunc place() int {{ return {expr} }}\n" + assert discover_raw(go, "go", "app.go")