From bf42bd22e759b113846eab1f09c50eb64a1a7071 Mon Sep 17 00:00:00 2001 From: "Robert C. Martin" Date: Wed, 30 Sep 2026 10:40:54 -0500 Subject: [PATCH] Add a multi-language mutation tester that writes .metrics/mutate for uml-viewer. --- .gitignore | 9 + README.md | 116 +++++++++ mutator | 9 + pyproject.toml | 26 ++ src/mutator/__init__.py | 3 + src/mutator/__main__.py | 4 + src/mutator/cli.py | 462 ++++++++++++++++++++++++++++++++++++ src/mutator/coverage.py | 161 +++++++++++++ src/mutator/crapper_link.py | 36 +++ src/mutator/edn.py | 206 ++++++++++++++++ src/mutator/engine.py | 370 +++++++++++++++++++++++++++++ src/mutator/functions.py | 195 +++++++++++++++ src/mutator/metrics.py | 344 +++++++++++++++++++++++++++ src/mutator/model.py | 81 +++++++ src/mutator/report.py | 74 ++++++ src/mutator/runner.py | 133 +++++++++++ src/mutator/sites.py | 254 ++++++++++++++++++++ tests/test_cli.py | 94 ++++++++ tests/test_coverage.py | 57 +++++ tests/test_edn.py | 55 +++++ tests/test_engine.py | 147 ++++++++++++ tests/test_metrics.py | 132 +++++++++++ tests/test_runner.py | 68 ++++++ tests/test_sites.py | 165 +++++++++++++ 24 files changed, 3201 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100755 mutator create mode 100644 pyproject.toml create mode 100644 src/mutator/__init__.py create mode 100644 src/mutator/__main__.py create mode 100644 src/mutator/cli.py create mode 100644 src/mutator/coverage.py create mode 100644 src/mutator/crapper_link.py create mode 100644 src/mutator/edn.py create mode 100644 src/mutator/engine.py create mode 100644 src/mutator/functions.py create mode 100644 src/mutator/metrics.py create mode 100644 src/mutator/model.py create mode 100644 src/mutator/report.py create mode 100644 src/mutator/runner.py create mode 100644 src/mutator/sites.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_coverage.py create mode 100644 tests/test_edn.py create mode 100644 tests/test_engine.py create mode 100644 tests/test_metrics.py create mode 100644 tests/test_runner.py create mode 100644 tests/test_sites.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5533635 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ +.metrics/ +target/ +*.egg-info/ +dist/ +build/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..4641eec --- /dev/null +++ b/README.md @@ -0,0 +1,116 @@ +# mutator + +Mutation testing for Clojure, Java, Go, TypeScript, Rust, and Python. One run detects the language of each source file, applies that language's mutation rules, and writes the snapshot [uml-viewer](https://github.com/unclebob/uml-viewer) already reads. + +Clojure follows [clj-mutate](https://github.com/unclebob/clj-mutate). Go follows [mutate4go](https://github.com/unclebob/mutate4go). Java follows [mutate4java](https://github.com/unclebob/mutate4java). TypeScript, Rust, and Python use the same decisions as Java, spelled in that language. Function names and namespaces come from [crapper](https://github.com/unclebob/crapper), which is what uml-viewer joins to `.metrics/crap.edn`. + +A mutant is killed when the tests fail or time out. It survives when the tests still pass. A site on a line the coverage report does not hit is uncovered and is not run. The score uml-viewer paints is killed / (killed + survived). + +## Run + +```bash +./mutator +``` + +The first run creates `.venv` and installs the tool. crapper has to be checked out next to this directory (`../crapper`), because that is where the namespace and function names come from. From a project root the tool walks the tree, skips `test`, `spec`, `vendor`, `node_modules`, and `target`, and writes `.metrics/mutate/.edn`. + +The first run of a tree executes every covered mutant. Start with the file you are working on. + +```bash +./mutator src/demo/core.clj # one file, or a directory +./mutator --scan src/demo # list sites, do not run tests +./mutator --changed # git additions and edits +./mutator --mutate-all # rerun killed mutants too +./mutator --no-coverage # do not skip uncovered lines +./mutator --use-existing-coverage # read reports already on disk +./mutator --test-command "pytest -q tests/test_demo.py" +``` + +Once a snapshot exists, the next run is differential. It reruns survivors and every site in a function whose text changed. Killed mutants in an unchanged function stay killed. `--since-last-run` is that same selection. `--mutate-all` ignores it. + +Exit `0` when every executed mutant was killed. Exit `2` when the baseline tests fail. Exit `3` when a mutant survives. A failed baseline does not rewrite the snapshot. + +Mutants run one at a time in the project tree. The original file is restored after each one. An interrupted run leaves a copy under `target/mutator-backup/`, and the next run puts it back before reading source. + +## Snapshot + +uml-viewer loads every `*.edn` file under `.metrics/mutate`. Each file is one namespace. A Java file with an inner class, or a Go file with a function and a method, therefore writes more than one file. Two Go files in the same package share one snapshot, and rerunning one of them keeps the other's forms. + +```clojure +{:version 1 + :source "src/demo/board.py" + :namespace "demo.board.Board" + :outcomes {"[\"src/demo/board.py\",\"demo.board.Board\",\"defn/place\",10,11,\">\",\">=\"]" :survived} + :forms [{:id "defn/place" + :kind "defn" + :file "src/demo/board.py" + :line 2 + :end-line 4 + :hash "…" + :killed 1 + :survived 1 + :uncovered 0 + :sites 2} + {:id "defn-/_hide" + :kind "defn-" + :killed 0 + :survived 0 + :uncovered 1 + :sites 1}]} +``` + +`:namespace` is the class key crapper uses. The form id is what the viewer splits: + +| Form id | Operation | Private | +| --- | --- | --- | +| `defn/place` | `place` | no | +| `defn-/hide` | `hide` | yes | + +`:sites` is every mutation in the function, including uncovered ones. The viewer prints `---no mutation sites---` when that count is zero. + +| Language | `:namespace` | `:name` | +| --- | --- | --- | +| Clojure | the `ns` | the `defn` / `defn-` name | +| Java | `package.Class`, or `package.Outer.Inner` | the method name | +| Go | the package import path, or `import/path.Receiver` | the function or method name | +| TypeScript | the dotted module path, or `module.Class` | the function or method name | +| Rust | `crate::module`, or `crate::module::Type` | the function or method name | +| Python | the dotted module path, or `module.Class` | the function or method name | + +`defn-` is Clojure's private form. Elsewhere a private method, an unexported Go function, a Python name that starts with `_`, a Rust function that is not `pub`, or a TypeScript `private` / `#` name uses `defn-/`. + +## What each language mutates + +Tokens inside strings, comments, and quoted Clojure forms are left alone. A replacement that does not compile fails the test run and counts as killed. + +| Language | Mutations | +| --- | --- | +| Clojure | clj-mutate's head symbols: `+`/`-`, `*` to `/`, `inc`/`dec`, comparisons, `=`/`not=`, `if`/`if-not`, `when`/`when-not`, `and`/`or`, and the seq, predicate, and coercion pairs. `true`/`false` and `0`/`1` anywhere. | +| Java | `true`/`false`, `==`/`!=`, `<`/`<=`, `>`/`>=`, `+`/`-`, `*`/`/`, `&&`/`||`, delete `!` and unary `-`, `0`/`1`. Constructors are not entries, matching crapper. | +| Go | mutate4go: `+`/`-`, `*` to `/` only, comparisons, `==`/`!=`, `true`/`false`, `&&`/`||`, `0`/`1`. | +| TypeScript | the Java set, plus `===`/`!==`. | +| Rust | the Java set (`&&`/`||`, `!`, unary `-`). | +| Python | the Java set, spelled `and`/`or`, `not`, and `True`/`False`. | + +## Tests and coverage + +The baseline command has to pass before any mutant runs. A mutant's timeout is ten times that baseline, and at least two seconds. `--timeout-factor` changes the multiple. `--test-command` replaces the default. Defaults: + +| Language | Command | Where | +| --- | --- | --- | +| Clojure | `clj -M:spec --tag ~no-mutate` when Speclj is on the classpath, otherwise `clj -M:test` or `bb spec` | nearest `deps.edn` or `bb.edn` | +| Java | `mvn -q test -DexcludeTags=no-mutate` | nearest `pom.xml` | +| Go | `go test -count=1` of the file's package | nearest `go.mod` | +| TypeScript | `npm test` | nearest `package.json` | +| Rust | `cargo test` | nearest `Cargo.toml` | +| Python | `python -m pytest`, or `unittest discover` | nearest project file | + +Coverage is generated with crapper's commands unless `--use-existing-coverage` or `--no-coverage` is set. A file missing from the report is treated as uncovered, and the run says so. + +## Development + +```bash +python3 -m venv .venv +.venv/bin/pip install -e '.[dev]' +.venv/bin/pytest +``` diff --git a/mutator b/mutator new file mode 100755 index 0000000..5cd7ca8 --- /dev/null +++ b/mutator @@ -0,0 +1,9 @@ +#!/bin/sh +# Run the mutation tool, creating a local virtualenv on first use. +ROOT=$(CDPATH= cd -- "$(dirname "$0")" && pwd) +PY="$ROOT/.venv/bin/python" +if [ ! -x "$PY" ]; then + python3 -m venv "$ROOT/.venv" + "$ROOT/.venv/bin/pip" install -e "$ROOT" +fi +exec "$PY" -m mutator "$@" diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ac155a7 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[project] +name = "mutator" +version = "0.1.0" +description = "Mutation testing for Clojure, Java, Go, TypeScript, Rust, and Python, in the snapshot uml-viewer reads." +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "tree-sitter>=0.24", + "tree-sitter-language-pack>=0.7", +] + +[project.scripts] +mutator = "mutator.cli:main" + +[project.optional-dependencies] +dev = ["pytest>=8"] + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/src/mutator/__init__.py b/src/mutator/__init__.py new file mode 100644 index 0000000..89d0345 --- /dev/null +++ b/src/mutator/__init__.py @@ -0,0 +1,3 @@ +"""Mutation testing for the languages crapper scores.""" + +__version__ = "0.1.0" diff --git a/src/mutator/__main__.py b/src/mutator/__main__.py new file mode 100644 index 0000000..ae8e3af --- /dev/null +++ b/src/mutator/__main__.py @@ -0,0 +1,4 @@ +from mutator.cli import main + +if __name__ == "__main__": + main() diff --git a/src/mutator/cli.py b/src/mutator/cli.py new file mode 100644 index 0000000..2c435c4 --- /dev/null +++ b/src/mutator/cli.py @@ -0,0 +1,462 @@ +"""Command line for the multi-language mutation tool.""" + +from __future__ import annotations + +import subprocess +import sys +from dataclasses import dataclass, field +from pathlib import Path + +from mutator.crapper_link import ensure_crapper +from mutator.coverage import covered_lines +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 = """\ +Usage: mutator [options] [path-or-filter ...] + +Discover mutation sites in Clojure, Java, Go, TypeScript, Rust, and Python, +run the project's tests against each one, and write `.metrics/mutate` for +uml-viewer. Namespaces and function names are the ones crapper writes into +`.metrics/crap.edn`. + +A form id is `defn/` or, for a private operation, `defn-/`. +uml-viewer joins that name to the operation and joins `:namespace` to the +class. + +Options: + -h, --help Print this help and exit. + --root Project root. Metrics are written here. + Default: the current directory. + -s, --source-root Walk this tree instead of the project root. + May be repeated. + --changed Mutate added and modified source files from + git status. + --scan List mutation sites. Do not run tests or + write metrics. + --mutate-all Run every covered site, including ones the + last snapshot already killed. + --since-last-run Run survivors and sites in new or rewritten + functions. This is the default when a snapshot + exists. + --lines Run only mutations on these source lines. + --no-coverage Treat every site as covered. + --use-existing-coverage Read coverage already on disk. Same as + --reuse-coverage. + --reuse-coverage Read coverage already on disk. + --coverage-command Run this command instead of the per-language + coverage tools, then read the reports it wrote. + --test-command Run this command for the baseline and every + mutant. The working directory is the project + root. + --timeout-factor Mutant timeout, as a multiple of the baseline + duration. Default: 10. The timeout is at least + 2 seconds. + --mutation-warning Warn when a file selects more covered sites + than this. Default: 50. + --verbose Print each test command and each mutant. + +Arguments: + path File or directory to mutate. Test paths are skipped. + filter When the argument is not a path, only source files whose + 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. + +The default, once a snapshot exists, reruns survivors and sites in functions +whose text changed. Killed mutants in unchanged functions are kept. + +Exit codes: + 0 every executed mutant was killed, or there was nothing to run + 1 usage error + 2 baseline tests failed + 3 at least one mutant survived + +Coverage, when it is produced, uses the same commands as crapper: + Clojure clj -M:cov --lcov + Java Maven JaCoCo + Go go test ./... -coverprofile=... + TypeScript npm run coverage, Vitest, or c8 + Rust cargo llvm-cov or cargo tarpaulin + Python coverage.py LCOV + +A site on a line the report does not mark as hit is uncovered and is not run. +""" + + +@dataclass +class Options: + action: str + message: str = "" + exit_code: int = 0 + project_root: Path = field(default_factory=lambda: Path(".")) + source_roots: list[str] = field(default_factory=list) + positionals: list[str] = field(default_factory=list) + scan: bool = False + mutate_all: bool = False + since_last_run: bool = False + lines: set[int] | None = None + no_coverage: bool = False + use_existing_coverage: bool = False + coverage_command: str | None = None + test_command: str | None = None + timeout_factor: float = 10.0 + mutation_warning: int = 50 + changed: bool = False + verbose: bool = False + + +def _take(args: list[str], index: int, option: str) -> str: + if index + 1 >= len(args) or not args[index + 1] or args[index + 1].startswith("-"): + raise ValueError(f"{option} requires a value") + return args[index + 1] + + +def _lines(value: str) -> set[int]: + found = set() + for piece in value.split(","): + piece = piece.strip() + if not piece: + continue + number = int(piece) + if number < 1: + raise ValueError("--lines requires positive line numbers") + found.add(number) + if not found: + raise ValueError("--lines requires a line number") + return found + + +def parse_args(argv: list[str] | None = None) -> Options: + args = list(sys.argv[1:] if argv is None else argv) + if any(arg in {"-h", "--help"} for arg in args): + return Options(action="help", message=HELP, exit_code=0) + options = Options(action="mutate") + index = 0 + try: + while index < len(args): + arg = args[index] + if arg in {"-s", "--source-root"}: + options.source_roots.append(_take(args, index, arg)) + index += 2 + continue + if arg == "--root": + options.project_root = Path(_take(args, index, arg)) + index += 2 + continue + if arg == "--coverage-command": + options.coverage_command = _take(args, index, arg) + index += 2 + continue + if arg == "--test-command": + options.test_command = _take(args, index, arg) + index += 2 + continue + if arg == "--timeout-factor": + options.timeout_factor = float(_take(args, index, arg)) + if options.timeout_factor <= 0: + raise ValueError("--timeout-factor requires a positive number") + index += 2 + continue + if arg == "--mutation-warning": + options.mutation_warning = int(_take(args, index, arg)) + index += 2 + continue + if arg == "--lines": + options.lines = _lines(_take(args, index, arg)) + index += 2 + continue + if arg == "--no-coverage": + options.no_coverage = True + index += 1 + continue + if arg in {"--use-existing-coverage", "--reuse-coverage"}: + options.use_existing_coverage = True + index += 1 + continue + if arg == "--changed": + options.changed = True + index += 1 + continue + if arg == "--scan": + options.scan = True + index += 1 + continue + if arg == "--mutate-all": + options.mutate_all = True + index += 1 + continue + if arg == "--since-last-run": + options.since_last_run = True + index += 1 + continue + if arg == "--verbose": + options.verbose = True + index += 1 + continue + if arg.startswith("-"): + raise ValueError(f"Unknown option: {arg}") + options.positionals.append(arg) + index += 1 + except ValueError as exc: + return Options(action="help", message=f"{exc}\n\n{HELP}", exit_code=1) + if options.mutate_all and options.since_last_run: + return Options( + action="help", + message=f"--mutate-all cannot be combined with --since-last-run\n\n{HELP}", + exit_code=1, + ) + if options.no_coverage and options.coverage_command: + return Options( + action="help", + message=f"--no-coverage cannot be combined with --coverage-command\n\n{HELP}", + exit_code=1, + ) + if options.scan and options.mutate_all: + return Options( + action="help", + message=f"--scan cannot be combined with --mutate-all\n\n{HELP}", + exit_code=1, + ) + return options + + +def _changed_files(root: Path) -> list[Path]: + result = subprocess.run( + ["git", "status", "--porcelain"], + cwd=root, + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + print(result.stderr.strip() or "git status failed", file=sys.stderr) + return [] + found: list[Path] = [] + for line in result.stdout.splitlines(): + if len(line) < 4: + 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()) + return found + + +def _positionals(root: Path, args: list[str]) -> tuple[list[Path], list[str]]: + existing: list[Path] = [] + filters: list[str] = [] + for arg in args: + candidate = Path(arg) + if not candidate.is_absolute(): + candidate = root / arg + if candidate.exists(): + existing.append(candidate.resolve()) + else: + filters.append(arg) + return existing, filters + + +def _changed_sources(root: Path, crapper) -> list[Path]: + files = [] + for path in _changed_files(root): + if crapper.discover.language_of(path) is None: + continue + if crapper.discover.is_test_file(path): + continue + files.append(path) + return files + + +def _is_source(path: Path, crapper) -> bool: + if crapper.discover.language_of(path) is None: + return False + return not crapper.discover.is_test_file(path) + + +def _explicit_sources(existing: list[Path], crapper) -> list[Path]: + files = [] + for path in existing: + if path.is_dir(): + files.extend(crapper.discover.iter_source_files([path])) + continue + if _is_source(path, crapper): + files.append(path) + continue + if crapper.discover.is_test_file(path): + print(f"Skipping test file {path}", file=sys.stderr) + return files + + +def _choose_files(options: Options, root: Path, existing: list[Path], crapper) -> list[Path]: + if options.changed: + return _changed_sources(root, crapper) + if options.source_roots: + return crapper.discover.iter_source_files([(root / path).resolve() for path in options.source_roots]) + if existing: + return _explicit_sources(existing, crapper) + return crapper.discover.iter_source_files([root]) + + +def _matching(files: list[Path], filters: list[str]) -> list[Path]: + if not filters: + return files + return [path for path in files if any(item in path.as_posix() for item in filters)] + + +def _inside_root(files: list[Path], root: Path) -> list[Path]: + inside = [] + for path in files: + try: + path.resolve().relative_to(root) + except ValueError: + print(f"Skipping {path}; it is outside {root}", file=sys.stderr) + continue + inside.append(path.resolve()) + return inside + + +def select_files(options: Options) -> list[Path]: + crapper = ensure_crapper() + root = options.project_root.resolve() + existing, filters = _positionals(root, options.positionals) + files = _choose_files(options, root, existing, crapper) + 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 + crapper = ensure_crapper() + crapper.runners.run_coverage(root, files, options.coverage_command) + + +def _coverage_for(options: Options, root: Path, path: Path) -> set[int] | None: + if options.no_coverage: + return None + crapper = ensure_crapper() + language = crapper.discover.language_of(path) + if language is None: + return None + return covered_lines(root, path, language) + + +def _print_help(options: Options) -> int: + stream = sys.stdout if options.exit_code == 0 else sys.stderr + ending = "" if options.message.endswith("\n") else "\n" + print(options.message, file=stream, end=ending) + return options.exit_code + + +def _require_crapper() -> str | None: + try: + ensure_crapper() + except ImportError as exc: + return str(exc) + return None + + +def _restore(root: Path) -> None: + for path in restore_backups(root): + print(f"Restored {path} from an interrupted mutation.", file=sys.stderr) + + +def _scan(options: Options, root: Path, files: list[Path]) -> int: + for path in files: + print( + scan_file( + path, + root, + covered_lines=_coverage_for(options, root, path), + ignore_coverage=options.no_coverage, + lines=options.lines, + ), + end="", + ) + return 0 + + +def _record(result, forms: list, written: list[str]) -> str: + if result.skipped: + print(f"Skipping {result.path}: {result.skipped}", file=sys.stderr) + return "skip" + if result.baseline_failed: + print(result.baseline_message, file=sys.stderr) + return "baseline" + print(format_site_log(result.sites, result.statuses), end="") + forms.extend(result.forms) + written.extend(result.written) + if any(form.survived for form in result.forms): + return "survived" + return "ok" + + +def _finish(baseline_failed: bool, survived: bool) -> int: + if baseline_failed: + return 2 + if survived: + return 3 + return 0 + + +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]] = {} + forms = [] + written: list[str] = [] + baseline_failed = False + survived = False + for path in files: + result = mutate_file( + path, + root, + runner=runner, + covered_lines=_coverage_for(options, root, path), + ignore_coverage=options.no_coverage, + mutate_all=options.mutate_all, + lines=options.lines, + test_command=options.test_command, + timeout_factor=options.timeout_factor, + mutation_warning=options.mutation_warning, + baselines=baselines, + ) + outcome = _record(result, forms, written) + if outcome == "baseline": + baseline_failed = True + elif outcome == "survived": + survived = True + if forms: + print(format_results(forms), end="") + for path in written: + print(f"Wrote {path}", file=sys.stderr) + return _finish(baseline_failed, survived) + + +def run(argv: list[str] | None = None) -> int: + options = parse_args(argv) + if options.action == "help": + return _print_help(options) + missing = _require_crapper() + if missing: + print(missing, file=sys.stderr) + return 2 + root = options.project_root.resolve() + files = select_files(options) + if not files: + print("No source files to mutate.") + return 0 + if not options.scan: + _restore(root) + _prepare_coverage(options, root, files) + if options.scan: + return _scan(options, root, files) + return _mutate_files(options, root, files) + + +def main(argv: list[str] | None = None) -> None: + sys.exit(run(argv)) diff --git a/src/mutator/coverage.py b/src/mutator/coverage.py new file mode 100644 index 0000000..c158e76 --- /dev/null +++ b/src/mutator/coverage.py @@ -0,0 +1,161 @@ +"""Decide which lines the tests already exercise. + +Java uses JaCoCo source lines. Go uses statement profiles. Clojure prefers +Cloverage's per-line form counts, then LCOV. The other languages use LCOV. +A file that does not appear in a report has no coverage data. +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +from pathlib import Path + +from mutator.crapper_link import ensure_crapper + + +def _candidates(normalize, source_path: str) -> list[str]: + absolute = normalize(str(Path(source_path).resolve())) if source_path else "" + candidates = [] + for value in (normalize(source_path), absolute): + if value and value not in candidates: + candidates.append(value) + return candidates + + +def _exact(normalized: dict, candidates: list[str]): + for candidate in candidates: + if candidate in normalized: + return normalized[candidate] + 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 _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 _lookup(index: dict, source_path: str): + if not index: + return None + crapper = ensure_crapper() + normalize = crapper.coverage.normalize_path + normalized = {normalize(key): value for key, value in index.items()} + candidates = _candidates(normalize, source_path) + found = _exact(normalized, candidates) + if found is not None: + return found + return _by_suffix(normalized, candidates, crapper.coverage.suffix_match) + + +def _covered_from_pairs(lines: dict[int, tuple[int, int]] | None) -> set[int] | None: + if lines is None: + return None + return {number for number, (covered, _total) in lines.items() if covered > 0} + + +def _jacoco_paths(root: Path) -> list[Path]: + paths = [root / "target" / "site" / "jacoco" / "jacoco.xml"] + paths.extend(root.glob("*/target/site/jacoco/jacoco.xml")) + paths.extend(root.glob("*/*/target/site/jacoco/jacoco.xml")) + return paths + + +def _line_number(line) -> int | None: + try: + number = int(line.get("nr") or "0") + instructions = int(line.get("ci") or "0") + branches = int(line.get("cb") or "0") + except ValueError: + return None + if instructions > 0 or branches > 0: + return number + return None + + +def _source_lines(source) -> set[int]: + covered = set() + for line in source.findall("line"): + number = _line_number(line) + if number is not None: + covered.add(number) + return covered + + +def _package_lines(xml_root) -> dict[str, set[int]]: + found: dict[str, set[int]] = {} + for package in xml_root.iter("package"): + package_name = (package.get("name") or "").strip("/") + for source in package.findall("sourcefile"): + filename = source.get("name") or "" + key = f"{package_name}/{filename}" if package_name else filename + found[key] = _source_lines(source) + return found + + +def _read_jacoco(path: Path, crapper): + text = path.read_text(encoding="utf-8", errors="replace") + cleaned = crapper.coverage._DOCTYPE.sub("", text) + try: + return ET.fromstring(cleaned) + except ET.ParseError: + return None + + +def _jacoco_index(root: Path) -> dict[str, set[int]]: + crapper = ensure_crapper() + found: dict[str, set[int]] = {} + seen: set[Path] = set() + for path in _jacoco_paths(root): + if not path.is_file(): + continue + resolved = path.resolve() + if resolved in seen: + continue + seen.add(resolved) + xml_root = _read_jacoco(path, crapper) + if xml_root is None: + continue + found.update(_package_lines(xml_root)) + return found + + +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) + if segments is None: + return None + covered: set[int] = set() + for start, end, _statements, hits in segments: + if hits <= 0: + continue + for line in range(start, end + 1): + covered.add(line) + return covered + + +def covered_lines(root: Path, source_path: Path, language: str) -> set[int] | None: + """Lines with a hit, or None when this file is absent from coverage.""" + + crapper = ensure_crapper() + path = source_path.resolve().as_posix() + if language == "java": + return _lookup(_jacoco_index(root), path) + bundle = crapper.coverage.load_bundle(root) + if language == "go": + return _go_lines(bundle.go_profile, path) + if language == "clojure": + html = _covered_from_pairs(_lookup(bundle.form_html, path)) + if html is not None: + return html + return _covered_from_pairs(_lookup(bundle.lcov, path)) diff --git a/src/mutator/crapper_link.py b/src/mutator/crapper_link.py new file mode 100644 index 0000000..7f045fb --- /dev/null +++ b/src/mutator/crapper_link.py @@ -0,0 +1,36 @@ +"""Import crapper from the environment or the sibling checkout. + +Function names and namespaces come from crapper so `.metrics/mutate` joins +the same operations uml-viewer joins to `.metrics/crap.edn`. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + + +def _sibling_src() -> Path: + project = Path(__file__).resolve().parents[2] + return project.parent / "crapper" / "src" + + +def ensure_crapper(): + try: + import crapper + except ImportError: + sibling = _sibling_src() + if not (sibling / "crapper").is_dir(): + raise ImportError( + "crapper is not installed and was not found at " + f"{sibling}. Clone github.com/unclebob/crapper next to mutator." + ) from None + sys.path.insert(0, str(sibling)) + import crapper + import crapper.coverage + import crapper.discover + import crapper.languages + import crapper.languages.treesitter + import crapper.runners + + return crapper diff --git a/src/mutator/edn.py b/src/mutator/edn.py new file mode 100644 index 0000000..9111fc4 --- /dev/null +++ b/src/mutator/edn.py @@ -0,0 +1,206 @@ +"""Read and write the EDN subset uml-viewer loads with clojure.edn.""" + +from __future__ import annotations + +import re + + +def _escape(value: str) -> str: + return value.replace("\\", "\\\\").replace('"', '\\"') + + +def _keyword_key(key: str) -> bool: + return bool(re.fullmatch(r"[A-Za-z*+!_?][A-Za-z0-9*+!_?\-./]*", key)) + + +def dumps(value) -> str: + """Render a Python value as EDN. + + Dict keys that are plain identifiers become keywords. Other keys and all + ordinary strings become EDN strings. `True` and `False` are not used; + booleans here are the strings `true` and `false` only when passed through + `keyword`. + """ + + return _render(value) + + +def keyword(name: str) -> tuple: + """A value that renders as an EDN keyword.""" + + return ("keyword", name) + + +def _render(value) -> str: + if value is None: + return "nil" + if isinstance(value, tuple) and len(value) == 2 and value[0] == "keyword": + return f":{value[1]}" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, int) and not isinstance(value, bool): + return str(value) + if isinstance(value, float): + text = f"{value:.4f}".rstrip("0").rstrip(".") + if "." not in text: + text += ".0" + return text + if isinstance(value, str): + return f'"{_escape(value)}"' + if isinstance(value, (list, tuple)): + body = " ".join(_render(item) for item in value) + return f"[{body}]" if body else "[]" + if isinstance(value, dict): + parts = [] + for key, item in value.items(): + rendered_key = f":{key}" if isinstance(key, str) and _keyword_key(key) else _render(key) + parts.append(f"{rendered_key} {_render(item)}") + body = " ".join(parts) + return "{" + body + "}" + raise TypeError(f"cannot render {type(value).__name__} as EDN") + + +class _Reader: + def __init__(self, text: str): + self.text = text + self.length = len(text) + self.index = 0 + + def parse(self): + self._skip() + if self.index >= self.length: + raise ValueError("empty EDN") + value = self._value() + self._skip() + if self.index < self.length: + raise ValueError(f"trailing EDN at {self.index}") + return value + + def _peek(self) -> str: + if self.index >= self.length: + raise ValueError("unexpected end of EDN") + return self.text[self.index] + + def _skip(self) -> None: + while self.index < self.length: + ch = self.text[self.index] + if ch.isspace() or ch == ",": + self.index += 1 + continue + if ch == ";": + while self.index < self.length and self.text[self.index] != "\n": + self.index += 1 + continue + break + + def _value(self): + self._skip() + ch = self._peek() + if ch == '"': + return self._string() + if ch == ":": + return self._keyword() + if ch == "{": + return ("map", self._collection("}", self._map_item)) + if ch == "[": + return ("vec", self._collection("]", self._element)) + if ch == "(": + return ("list", self._collection(")", self._element)) + if ch == "#": + self.index += 1 + if self._peek() == "{": + return set(self._collection("}", self._element)) + raise ValueError("unsupported EDN dispatch") + return self._atom() + + def _collection(self, closing: str, item): + self.index += 1 + values = [] + while True: + self._skip() + if self._peek() == closing: + self.index += 1 + return values + values.append(item()) + + def _element(self): + return self._value() + + def _map_item(self): + key = self._value() + self._skip() + val = self._value() + return key, val + + def _string(self) -> str: + self.index += 1 + chars: list[str] = [] + while self.index < self.length: + ch = self.text[self.index] + self.index += 1 + if ch == '"': + return "".join(chars) + if ch == "\\": + if self.index >= self.length: + raise ValueError("unterminated string escape") + escaped = self.text[self.index] + self.index += 1 + chars.append({"n": "\n", "t": "\t", "r": "\r", '"': '"', "\\": "\\"}.get(escaped, escaped)) + continue + chars.append(ch) + raise ValueError("unterminated string") + + def _keyword(self) -> str: + self.index += 1 + start = self.index + while self.index < self.length and _symbol_char(self.text[self.index]): + self.index += 1 + if self.index == start: + raise ValueError("empty keyword") + return self.text[start : self.index] + + def _atom(self): + start = self.index + while self.index < self.length and _symbol_char(self.text[self.index]): + self.index += 1 + token = self.text[start : self.index] + if token == "": + raise ValueError(f"unexpected character {self._peek()!r}") + if token == "nil": + return None + if token == "true": + return True + if token == "false": + return False + if re.fullmatch(r"-?\d+", token): + return int(token) + if re.fullmatch(r"-?\d+\.\d+", token): + return float(token) + return token + + +def _symbol_char(ch: str) -> bool: + return ch not in ' \t\r\n,{}[]()"\\;' + + +def _maps(value): + if isinstance(value, tuple) and len(value) == 2 and value[0] in {"map", "vec", "list"}: + kind, items = value + if kind == "map": + return {_key(key): _maps(item) for key, item in items} + return [_maps(item) for item in items] + if isinstance(value, set): + return {_maps(item) for item in value} + return value + + +def _key(value): + if isinstance(value, (str, int, float)) or value is None: + return value + return str(value) + + +def loads(text: str): + """Parse one EDN value. Keywords become their names, without the colon.""" + + return _maps(_Reader(text).parse()) diff --git a/src/mutator/engine.py b/src/mutator/engine.py new file mode 100644 index 0000000..9a13a8f --- /dev/null +++ b/src/mutator/engine.py @@ -0,0 +1,370 @@ +"""Apply each selected mutant, run tests, and record the snapshot.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +from mutator.crapper_link import ensure_crapper +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.sites import apply_site + + +def restore_backups(root: Path) -> list[Path]: + """Put back sources left mutated by a killed run.""" + + base = root / "target" / "mutator-backup" + restored: list[Path] = [] + if not base.is_dir(): + return restored + for backup in sorted(path for path in base.rglob("*") if path.is_file()): + relative = backup.relative_to(base) + target = root / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(backup.read_bytes()) + backup.unlink() + restored.append(target) + return restored + + +def _drop_bytecode(path: Path) -> None: + """A same-second .pyc would hide the mutant from Python.""" + + cache = path.parent / "__pycache__" + if not cache.is_dir(): + return + for item in cache.glob(path.stem + ".*.pyc"): + item.unlink(missing_ok=True) + + +def _backup(root: Path, path: Path, original: bytes) -> Path: + relative = path.resolve().relative_to(root.resolve()) + backup = root / "target" / "mutator-backup" / relative + backup.parent.mkdir(parents=True, exist_ok=True) + if not backup.exists(): + backup.write_bytes(original) + return backup + + +def _covered(site: Site, lines: set[int] | None, ignore_coverage: bool) -> bool: + if ignore_coverage: + return True + if lines is None: + return False + return site.line in lines + + +def _select( + sites: list[Site], + covered: dict[str, bool], + history, + digests: dict[tuple[str, str], str], + mutate_all: bool, + lines: set[int] | None, +) -> list[Site]: + chosen = [] + for site in sites: + if not covered[site.mutation_id]: + continue + if lines is not None and site.line not in lines: + continue + key = (site.namespace, site.form_id) + prior = history.forms.get(key) + unchanged = prior is not None and prior.digest == digests.get(key) + if mutate_all or not unchanged: + chosen.append(site) + continue + if history.outcomes.get(site.mutation_id) == "killed": + continue + chosen.append(site) + return chosen + + +def _carry_forward(sites: list[Site], covered: dict[str, bool], history, digests) -> dict[str, str]: + by_id = {site.mutation_id: site for site in sites} + carried = {} + for mutation, status in history.outcomes.items(): + site = by_id.get(mutation) + if site is None or not covered[site.mutation_id]: + continue + key = (site.namespace, site.form_id) + prior = history.forms.get(key) + if prior is not None and prior.digest == digests.get(key): + carried[mutation] = status + return carried + + +def _forms( + source: str, + path: Path, + root: Path, + file_key: str, + sites: list[Site], + covered: dict[str, bool], + outcomes: dict[str, str], + lines: set[int] | None, +) -> list[FormResult]: + spans = form_spans(source, path, root) + digests = form_digests(source, path, root) + grouped: dict[tuple[str, str], list[Site]] = {} + for site in sites: + grouped.setdefault((site.namespace, site.form_id), []).append(site) + forms = [] + for key, (start, end, name, private) in sorted(spans.items(), key=lambda item: item[1][0]): + namespace, form = key + owned = grouped.get(key, []) + killed = survived = uncovered = 0 + for site in owned: + if not covered[site.mutation_id]: + uncovered += 1 + continue + status = outcomes.get(site.mutation_id) + if status == "killed": + killed += 1 + elif status == "survived": + survived += 1 + elif lines is None: + survived += 1 + forms.append( + FormResult( + id=form, + namespace=namespace, + name=name, + private=private, + file=file_key, + line=start, + end_line=end, + digest=digests[key], + killed=killed, + survived=survived, + uncovered=uncovered, + sites=len(owned), + ) + ) + return forms + + +def _run_mutant(path: Path, root: Path, original: bytes, site: Site, run) -> str: + current = original[site.start : site.end].decode("utf-8") + if current != site.original: + print(f"Skipped {path}:{site.line} {site.description}; source bytes moved", file=sys.stderr) + return "survived" + backup = _backup(root, path, original) + mutated = apply_site(original.decode("utf-8"), site.start, site.end, site.mutant) + path.write_text(mutated, encoding="utf-8") + _drop_bytecode(path) + try: + return run() + finally: + path.write_bytes(original) + if path.read_bytes() == original and backup.is_file(): + backup.unlink() + + +def _decode(path: Path) -> tuple[bytes, str] | None: + original = path.read_bytes() + try: + return original, original.decode("utf-8") + except UnicodeError: + return None + + +def _coverage_map( + found: list[Site], covered_lines: set[int] | None, ignore_coverage: bool +) -> dict[str, bool]: + return {site.mutation_id: _covered(site, covered_lines, ignore_coverage) for site in found} + + +def _warn_uncovered(file_key: str, found, covered_lines, ignore_coverage: bool) -> None: + if covered_lines is not None or ignore_coverage or not found: + return + print( + f"No coverage data for {file_key}; its sites are uncovered. " + "Pass --no-coverage to run them anyway.", + file=sys.stderr, + ) + + +def _warn_many(file_key: str, selected, mutation_warning: int) -> None: + if len(selected) > mutation_warning: + print( + f"WARNING: {len(selected)} covered mutations selected in {file_key}.", + file=sys.stderr, + ) + + +def _remember_baseline(baselines, cache_key, runner, command: str, cwd: Path) -> None: + if cache_key in baselines: + return + baseline = runner.run(command, cwd, None) + if baseline.code != 0: + tail = "\n".join(baseline.output.splitlines()[-20:]) + baselines[cache_key] = (False, baseline.seconds, tail) + return + 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}" + if tail: + message = f"{message}\n{tail}" + return RunResult(path=file_key, forms=[], written=[], baseline_failed=True, baseline_message=message) + + +def _mutant_status(runner, command: str, cwd: Path, timeout: float) -> str: + result = runner.run(command, cwd, timeout) + if result.timed_out or result.code != 0: + return "killed" + return "survived" + + +def _run_selected(path, root, original, selected, runner, command, cwd, timeout, file_key, outcomes) -> None: + for site in selected: + if runner.verbose: + print(f"{file_key}:{site.line} {site.description}", file=sys.stderr) + outcomes[site.mutation_id] = _run_mutant( + path, + root, + original, + site, + lambda: _mutant_status(runner, command, cwd, timeout), + ) + + +def _apply_selected( + path, + root, + original, + selected, + runner, + language, + test_command, + timeout_factor, + file_key, + baselines, + outcomes, +) -> RunResult | None: + if not selected: + return None + command, cwd = test_plan(root, path, language, test_command) + cache_key = (command, str(cwd)) + _remember_baseline(baselines, cache_key, runner, command, cwd) + ok, seconds, tail = baselines[cache_key] + if not ok: + return _baseline_failure(file_key, command, tail) + timeout = max(2.0, seconds * timeout_factor) + _run_selected(path, root, original, selected, runner, command, cwd, timeout, file_key, outcomes) + return None + + +def _statuses(found: list[Site], covered: dict[str, bool], outcomes: dict[str, str]) -> dict[str, str]: + statuses = {} + for site in found: + if not covered[site.mutation_id]: + statuses[site.mutation_id] = "uncovered" + elif site.mutation_id in outcomes: + statuses[site.mutation_id] = outcomes[site.mutation_id] + return statuses + + +def _restore_if_dirty(path: Path, original: bytes) -> None: + if path.read_bytes() != original: + path.write_bytes(original) + + +def mutate_file( + path: Path, + root: Path, + *, + runner: CommandRunner, + covered_lines: set[int] | None, + ignore_coverage: bool, + mutate_all: bool, + lines: set[int] | None, + test_command: str | None, + timeout_factor: float, + mutation_warning: int, + baselines: dict[tuple[str, str], tuple[bool, float, str]], +) -> RunResult: + """Mutate one file and write its namespaces into `.metrics/mutate`.""" + + root = root.resolve() + path = path.resolve() + crapper = ensure_crapper() + language = crapper.discover.language_of(path) + file_key = source_key(path, root) + if language is None: + return RunResult(path=file_key, forms=[], written=[], skipped="unsupported file") + decoded = _decode(path) + if decoded is None: + return RunResult(path=file_key, forms=[], written=[], skipped="file is not UTF-8") + original, source = decoded + found = sites_in_file(source, path, root, file_key) + covered = _coverage_map(found, covered_lines, ignore_coverage) + _warn_uncovered(file_key, found, covered_lines, ignore_coverage) + history = load_history(root, file_key) + digests = form_digests(source, path, root) + selected = _select(found, covered, history, digests, mutate_all, lines) + outcomes = _carry_forward(found, covered, history, digests) + _warn_many(file_key, selected, mutation_warning) + failure = _apply_selected( + path, + root, + original, + selected, + runner, + language, + test_command, + timeout_factor, + file_key, + baselines, + outcomes, + ) + if failure is not None: + return failure + forms = _forms(source, path, root, file_key, found, covered, outcomes, lines) + written = write_results(root, file_key, forms, outcomes) + _restore_if_dirty(path, original) + return RunResult( + path=file_key, + forms=forms, + written=[item.as_posix() for item in written], + sites=found, + statuses=_statuses(found, covered, outcomes), + ) + + +def scan_file( + path: Path, + root: Path, + *, + covered_lines: set[int] | None, + ignore_coverage: bool, + lines: set[int] | None, +) -> str: + """Inventory mutation sites without running tests or writing metrics.""" + + root = root.resolve() + path = path.resolve() + file_key = source_key(path, root) + source = path.read_text(encoding="utf-8") + found = sites_in_file(source, path, root, file_key) + if lines is not None: + found = [site for site in found if site.line in lines] + covered = None + if not ignore_coverage and covered_lines is not None: + covered = {site.mutation_id: site.line in covered_lines for site in found} + history = load_history(root, file_key) + digests = form_digests(source, path, root) + changed = set() + if history.forms: + changed = { + key + for key, digest in digests.items() + if key not in history.forms or history.forms[key].digest != digest + } + return format_scan(file_key, found, changed, covered) diff --git a/src/mutator/functions.py b/src/mutator/functions.py new file mode 100644 index 0000000..3f6ef8a --- /dev/null +++ b/src/mutator/functions.py @@ -0,0 +1,195 @@ +"""Bind mutation sites to the functions crapper reports.""" + +from __future__ import annotations + +import hashlib +import json +import re +from pathlib import Path + +from mutator.crapper_link import ensure_crapper +from mutator.model import Site +from mutator.sites import discover_raw + + +def _lines(source: str) -> list[str]: + return source.splitlines() + + +def _slice(source: str, start: int, end: int) -> str: + lines = _lines(source) + return "\n".join(lines[start - 1 : end]) + + +def digest(source: str, start: int, end: int) -> str: + chunk = "\n".join(line.rstrip() for line in _lines(source)[start - 1 : end]) + return hashlib.sha256(chunk.encode("utf-8")).hexdigest() + + +def _header(source: str, start: int, end: int) -> str: + text = _slice(source, start, min(end, start + 6)).lstrip() + while text.startswith("#["): + close = text.find("]") + if close < 0: + break + text = text[close + 1 :].lstrip() + return text + + +def _python_private(name: str) -> bool: + if name.startswith("__") and name.endswith("__"): + return False + return name.startswith("_") + + +def _typescript_private(name: str, header: str) -> bool: + return name.startswith("#") or bool(re.search(r"\bprivate\b", header)) + + +def is_private(language: str, name: str, source: str, start: int, end: int) -> bool: + """True when uml-viewer should treat the operation as private.""" + + header = _header(source, start, end) + if language == "clojure": + return bool(re.match(r"\(defn-", header)) + if language == "java": + return bool(re.search(r"\bprivate\b", header)) + if language == "go": + return name[:1].islower() + if language == "python": + return _python_private(name) + if language == "rust": + return re.match(r"pub\s", header) is None + if language == "typescript": + return _typescript_private(name, header) + return False + + +def form_id(name: str, private: bool) -> str: + """The id uml-viewer splits into an operation name. + + `defn/place` is public. `defn-/hide` is private. The viewer checks the + private prefix first. + """ + + prefix = "defn-" if private else "defn" + return f"{prefix}/{name}" + + +def mutation_id( + file: str, namespace: str, form: str, start: int, end: int, original: str, mutant: str +) -> str: + return json.dumps( + [file, namespace, form, start, end, original, mutant], separators=(",", ":") + ) + + +def _mutation_field(mutation: str, index: int) -> str | None: + try: + value = json.loads(mutation) + except json.JSONDecodeError: + return None + if isinstance(value, list) and len(value) > index and isinstance(value[index], str): + return value[index] + return None + + +def mutation_file(mutation: str) -> str | None: + return _mutation_field(mutation, 0) + + +def mutation_namespace(mutation: str) -> str | None: + return _mutation_field(mutation, 1) + + +def project_functions(source: str, path: Path, root: Path): + crapper = ensure_crapper() + language = crapper.discover.language_of(path) + if language is None: + return [] + return crapper.languages.functions_in_file( + language, + source, + path.resolve().as_posix(), + root.resolve().as_posix(), + ) + + +def _owner(functions, line: int): + matches = [fn for fn in functions if fn.start_line <= line <= fn.end_line] + if not matches: + return None + return min(matches, key=lambda fn: (fn.end_line - fn.start_line, fn.start_line)) + + +def sites_in_file(source: str, path: Path, root: Path, file_key: str) -> list[Site]: + crapper = ensure_crapper() + language = crapper.discover.language_of(path) + if language is None: + return [] + functions = project_functions(source, path, root) + privacy = { + id(fn): is_private(language, fn.name, source, fn.start_line, fn.end_line) for fn in functions + } + found: list[Site] = [] + for raw in discover_raw(source, language, path.as_posix()): + owner = _owner(functions, raw.line) + if owner is None: + continue + private = privacy[id(owner)] + form = form_id(owner.name, private) + found.append( + Site( + file=file_key, + namespace=owner.namespace, + form_id=form, + name=owner.name, + line=raw.line, + start=raw.start, + end=raw.end, + original=raw.original, + mutant=raw.mutant, + category=raw.category, + mutation_id=mutation_id( + file_key, owner.namespace, form, raw.start, raw.end, raw.original, raw.mutant + ), + ) + ) + return found + + +def form_digests(source: str, path: Path, root: Path) -> dict[tuple[str, str], str]: + """Map `(namespace, form id)` to a hash of the function text.""" + + crapper = ensure_crapper() + language = crapper.discover.language_of(path) + if language is None: + return {} + grouped: dict[tuple[str, str], list] = {} + for fn in project_functions(source, path, root): + private = is_private(language, fn.name, source, fn.start_line, fn.end_line) + key = (fn.namespace, form_id(fn.name, private)) + grouped.setdefault(key, []).append(fn) + digests = {} + for key, fns in grouped.items(): + ordered = sorted(fns, key=lambda fn: (fn.start_line, fn.end_line)) + parts = [digest(source, fn.start_line, fn.end_line) for fn in ordered] + digests[key] = hashlib.sha256("".join(parts).encode("utf-8")).hexdigest() + return digests + + +def form_spans(source: str, path: Path, root: Path) -> dict[tuple[str, str], tuple[int, int, str, bool]]: + """`(namespace, form id)` -> start line, end line, name, private.""" + + crapper = ensure_crapper() + language = crapper.discover.language_of(path) + spans = {} + for fn in project_functions(source, path, root): + private = is_private(language or "", fn.name, source, fn.start_line, fn.end_line) + key = (fn.namespace, form_id(fn.name, private)) + if key not in spans: + spans[key] = (fn.start_line, fn.end_line, fn.name, private) + continue + start, end, name, private_flag = spans[key] + spans[key] = (min(start, fn.start_line), max(end, fn.end_line), name, private_flag) + return spans diff --git a/src/mutator/metrics.py b/src/mutator/metrics.py new file mode 100644 index 0000000..daf18ff --- /dev/null +++ b/src/mutator/metrics.py @@ -0,0 +1,344 @@ +"""Read and write `.metrics/mutate` snapshots. + +uml-viewer loads every `*.edn` file under `.metrics/mutate`. Each file is one +namespace. It joins a form to an operation when the form id is `defn/` +or `defn-/`, and it joins the file to a class on `:namespace`. +""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path + +from mutator.edn import dumps, keyword, loads +from mutator.functions import mutation_file, mutation_namespace +from mutator.model import FormResult, History, PriorForm + + +def _relative(path: Path, root: Path) -> str: + return path.resolve().relative_to(root.resolve()).as_posix() + + +def source_key(path: Path, root: Path) -> str: + return _relative(path, root) + + +def snapshot_path(root: Path, namespace: str) -> Path: + parts = namespace.replace("\\", "/").replace("::", "/").split("/") + safe = [] + for part in parts: + if part in {"", ".", ".."}: + raise ValueError(f"unsafe namespace {namespace!r}") + safe.append(part) + return root.joinpath(".metrics", "mutate", *safe[:-1], safe[-1] + ".edn") + + +def _edn_files(root: Path) -> list[Path]: + directory = root / ".metrics" / "mutate" + if not directory.is_dir(): + return [] + return sorted(path for path in directory.rglob("*.edn") if path.is_file()) + + +def _read(path: Path) -> dict | None: + try: + data = loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return data if isinstance(data, dict) else None + + +def _forms_of(data: dict) -> list[dict]: + forms = data.get("forms") + if not isinstance(forms, list): + return [] + return [form for form in forms if isinstance(form, dict)] + + +def _outcomes_of(data: dict) -> dict[str, str]: + raw = data.get("outcomes") + if not isinstance(raw, dict): + return {} + found = {} + for key, value in raw.items(): + if not isinstance(key, str): + continue + status = value[1:] if isinstance(value, str) and value.startswith(":") else value + if status in {"killed", "survived"}: + found[key] = status + return found + + +def _form_file(form: dict, data: dict) -> str | None: + file_name = form.get("file") + if isinstance(file_name, str) and file_name: + return file_name + source = data.get("source") + return source if isinstance(source, str) else None + + +def _remember_form(history: History, namespace: str, form: dict, data: dict, file_key: str) -> None: + if _form_file(form, data) != file_key: + return + form_id = form.get("id") + digest = form.get("hash") + if not isinstance(form_id, str) or not isinstance(digest, str): + return + history.forms[(namespace, form_id)] = PriorForm( + namespace=namespace, + id=form_id, + digest=digest, + file=file_key, + ) + + +def _remember_forms(history: History, data: dict, namespace: str, file_key: str) -> None: + for form in _forms_of(data): + _remember_form(history, namespace, form, data, file_key) + + +def _remember_outcomes(history: History, data: dict, file_key: str) -> None: + for mutation, status in _outcomes_of(data).items(): + if mutation_file(mutation) == file_key: + history.outcomes[mutation] = status + + +def load_history(root: Path, file_key: str) -> History: + """Prior form hashes and outcomes recorded for `file_key`.""" + + history = History() + for path in _edn_files(root): + data = _read(path) + if data is None: + continue + namespace = data.get("namespace") + if not isinstance(namespace, str): + continue + _remember_forms(history, data, namespace, file_key) + _remember_outcomes(history, data, file_key) + return history + + +def _render_form(form: FormResult) -> dict: + return { + "id": form.id, + "kind": "defn-" if form.private else "defn", + "file": form.file, + "line": form.line, + "end-line": form.end_line, + "hash": form.digest, + "killed": form.killed, + "survived": form.survived, + "uncovered": form.uncovered, + "sites": form.sites, + } + + +def _render_snapshot(namespace: str, source: str, forms: list[dict], outcomes: dict[str, str]) -> str: + payload = { + "version": 1, + "tested-at": datetime.now().astimezone().isoformat(timespec="seconds"), + "source": source, + "namespace": namespace, + "outcomes": {key: keyword(outcomes[key]) for key in sorted(outcomes)}, + "forms": forms, + } + text = dumps(payload) + # Keep the file readable. dumps is one line; break after the top-level pairs + # is unnecessary because clojure.edn reads either shape. + return text + "\n" + + +def _kept_forms(data: dict, file_key: str) -> list[dict]: + return [form for form in _forms_of(data) if _form_file(form, data) != file_key] + + +def _kept_outcomes(data: dict, file_key: str) -> dict[str, str]: + return { + mutation: status + for mutation, status in _outcomes_of(data).items() + if mutation_file(mutation) != file_key + } + + +def _group_forms(forms: list[FormResult]) -> dict[str, list[FormResult]]: + incoming: dict[str, list[FormResult]] = {} + for form in forms: + incoming.setdefault(form.namespace, []).append(form) + return incoming + + +def _file_outcomes(outcomes: dict[str, str], file_key: str) -> dict[str, str]: + found = {} + for mutation, status in outcomes.items(): + if mutation_file(mutation) != file_key: + continue + if status not in {"killed", "survived"}: + continue + found[mutation] = status + return found + + +def _read_snapshots(root: Path) -> list[tuple[Path, dict]]: + existing = [] + for path in _edn_files(root): + data = _read(path) + if data is not None: + existing.append((path, data)) + return existing + + +def _mentions_file(data: dict, file_key: str) -> bool: + for form in _forms_of(data): + if _form_file(form, data) == file_key: + return True + return False + + +def _keep(retained: dict, namespace: str, data: dict, file_key: str) -> None: + kept_forms = _kept_forms(data, file_key) + kept_outcomes = _kept_outcomes(data, file_key) + if not kept_forms and not kept_outcomes: + return + forms_so_far, outcomes_so_far = retained.get(namespace, ([], {})) + forms_so_far.extend(kept_forms) + outcomes_so_far.update(kept_outcomes) + retained[namespace] = (forms_so_far, outcomes_so_far) + + +def _retain(existing, file_key: str): + retained: dict[str, tuple[list[dict], dict[str, str]]] = {} + previous: set[str] = set() + for _path, data in existing: + namespace = data.get("namespace") + if not isinstance(namespace, str): + continue + if _mentions_file(data, file_key): + previous.add(namespace) + _keep(retained, namespace, data, file_key) + return retained, previous + + +def _outcomes_for(outcomes: dict[str, str], namespace: str) -> dict[str, str]: + found = {} + for mutation, status in outcomes.items(): + if mutation_namespace(mutation) == namespace: + found[mutation] = status + return found + + +def _form_order(form: dict) -> tuple: + return (str(form.get("file")), int(form.get("line") or 0), str(form.get("id"))) + + +def _snapshot_source(fresh: list[FormResult], kept_forms: list[dict], file_key: str) -> str: + if fresh: + return fresh[0].file + return str(kept_forms[0].get("file") or file_key) + + +def _write_namespace( + root: Path, + namespace: str, + kept_forms: list[dict], + kept_outcomes: dict[str, str], + fresh: list[FormResult], + outcomes: dict[str, str], + file_key: str, +) -> Path | None: + rendered = _dedupe(kept_forms + [_render_form(form) for form in fresh]) + rendered.sort(key=_form_order) + merged = dict(kept_outcomes) + if fresh: + merged.update(_outcomes_for(outcomes, namespace)) + path = snapshot_path(root, namespace) + if not rendered and not merged: + path.unlink(missing_ok=True) + return None + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + _render_snapshot(namespace, _snapshot_source(fresh, kept_forms, file_key), rendered, merged), + encoding="utf-8", + ) + return path + + +def _stale(path: Path, data: dict, touched: set[str], canonical: set[Path], file_key: str) -> bool: + if path.resolve() in canonical: + return False + namespace = data.get("namespace") + if isinstance(namespace, str) and namespace in touched: + return True + return _form_file_only(data, file_key) + + +def _drop_stale(existing, touched: set[str], canonical: set[Path], file_key: str) -> None: + """Remove a per-file snapshot so the viewer cannot prefer it over the namespace file.""" + + for path, data in existing: + if _stale(path, data, touched, canonical, file_key): + path.unlink(missing_ok=True) + + +def write_results( + root: Path, file_key: str, forms: list[FormResult], outcomes: dict[str, str] +) -> list[Path]: + """Replace this file's forms, preserving other files that share a namespace. + + One snapshot file per namespace. Older per-file snapshots for a namespace + this write touches are folded in and removed, so uml-viewer sees a single + map for that class. + """ + + root = root.resolve() + incoming = _group_forms(forms) + incoming_outcomes = _file_outcomes(outcomes, file_key) + existing = _read_snapshots(root) + retained, previous = _retain(existing, file_key) + touched = set(incoming) | previous | set(retained) + written: list[Path] = [] + for namespace in sorted(touched): + kept_forms, kept_outcomes = retained.get(namespace, ([], {})) + path = _write_namespace( + root, + namespace, + kept_forms, + kept_outcomes, + incoming.get(namespace, []), + incoming_outcomes, + file_key, + ) + if path is not None: + written.append(path) + canonical = {snapshot_path(root, namespace).resolve() for namespace in touched} + _drop_stale(existing, touched, canonical, file_key) + _remove_empty_dirs(root / ".metrics" / "mutate") + return written + + +def _form_file_only(data: dict, file_key: str) -> bool: + forms = _forms_of(data) + if not forms: + return data.get("source") == file_key + files = {_form_file(form, data) for form in forms} + return files == {file_key} + + +def _dedupe(forms: list[dict]) -> list[dict]: + seen = set() + unique = [] + for form in forms: + key = (form.get("file"), form.get("id"), form.get("line"), form.get("hash")) + if key in seen: + continue + seen.add(key) + unique.append(form) + return unique + + +def _remove_empty_dirs(directory: Path) -> None: + if not directory.is_dir(): + return + for child in sorted(directory.rglob("*"), reverse=True): + if child.is_dir() and not any(child.iterdir()): + child.rmdir() diff --git a/src/mutator/model.py b/src/mutator/model.py new file mode 100644 index 0000000..37f5745 --- /dev/null +++ b/src/mutator/model.py @@ -0,0 +1,81 @@ +"""Mutation sites and the per-function counts uml-viewer reads.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Site: + """One replacement inside a function crapper would score.""" + + file: str + namespace: str + form_id: str + name: str + line: int + start: int + end: int + original: str + mutant: str + category: str + mutation_id: str + + @property + def description(self) -> str: + if self.mutant == "": + return f"delete {self.original}" + return f"{self.original} -> {self.mutant}" + + +@dataclass +class FormResult: + """Killed, survived, and uncovered totals for one operation.""" + + id: str + namespace: str + name: str + private: bool + file: str + line: int + end_line: int + digest: str + killed: int = 0 + survived: int = 0 + uncovered: int = 0 + sites: int = 0 + + @property + def score(self) -> float | None: + executed = self.killed + self.survived + if executed == 0: + return None + return self.killed / executed + + +@dataclass +class PriorForm: + namespace: str + id: str + digest: str + file: str + + +@dataclass +class History: + """Previous results for one source file, used for differential runs.""" + + forms: dict[tuple[str, str], PriorForm] = field(default_factory=dict) + outcomes: dict[str, str] = field(default_factory=dict) + + +@dataclass +class RunResult: + path: str + forms: list[FormResult] + written: list[str] + sites: list[Site] = field(default_factory=list) + statuses: dict[str, str] = field(default_factory=dict) + baseline_failed: bool = False + baseline_message: str = "" + skipped: str = "" diff --git a/src/mutator/report.py b/src/mutator/report.py new file mode 100644 index 0000000..e861390 --- /dev/null +++ b/src/mutator/report.py @@ -0,0 +1,74 @@ +"""Text reports for a mutation run and for --scan.""" + +from __future__ import annotations + +from mutator.model import FormResult, Site + + +def _score(form: FormResult) -> str: + if form.score is None: + return " n/a" + return f"{form.score * 100:5.1f}%" + + +def format_results(forms: list[FormResult]) -> str: + if not forms: + return "No functions to mutate.\n" + name_width = max(8, min(48, max(len(form.name) for form in forms))) + namespace_width = max(9, min(72, max(len(form.namespace) for form in forms))) + header = ( + f"{'Function':<{name_width}} {'Namespace':<{namespace_width}} " + f"{'Killed':>7} {'Survived':>9} {'Uncovered':>10} {'Sites':>6} {'Score':>7}" + ) + ordered = sorted( + forms, + key=lambda form: (form.score is not None, form.score or 0, -form.survived, form.name), + ) + lines = [header, "-" * len(header)] + killed = survived = uncovered = sites = 0 + for form in ordered: + lines.append( + f"{form.name:<{name_width}} {form.namespace:<{namespace_width}} " + f"{form.killed:7d} {form.survived:9d} {form.uncovered:10d} {form.sites:6d} {_score(form)}" + ) + killed += form.killed + survived += form.survived + uncovered += form.uncovered + sites += form.sites + executed = killed + survived + score = "n/a" if executed == 0 else f"{(killed / executed) * 100:.1f}%" + lines.append("-" * len(header)) + lines.append( + f"{'Total':<{name_width}} {'':<{namespace_width}} " + f"{killed:7d} {survived:9d} {uncovered:10d} {sites:6d} {score:>7}" + ) + lines.append("") + return "\n".join(lines) + + +def format_site_log(sites: list[Site], status: dict[str, str]) -> str: + lines = [] + for site in sites: + label = status.get(site.mutation_id) + if label is None: + continue + lines.append(f"{label.upper():9} {site.file}:{site.line} {site.description}") + if not lines: + return "" + return "\n".join(lines) + "\n" + + +def format_scan( + path: str, sites: list[Site], changed: set[tuple[str, str]], covered: dict[str, bool] | None +) -> str: + lines = [f"Scan: {len(sites)} mutation sites in {path}"] + for site in sites: + mark = "*" if (site.namespace, site.form_id) in changed else " " + coverage = "" + if covered is not None and not covered.get(site.mutation_id, True): + coverage = " uncovered" + lines.append(f"{mark} {path}:{site.line} {site.description}{coverage} [{site.form_id}]") + if changed: + lines.append("* marks a function whose text changed since the last snapshot.") + lines.append("") + return "\n".join(lines) diff --git a/src/mutator/runner.py b/src/mutator/runner.py new file mode 100644 index 0000000..66ccaba --- /dev/null +++ b/src/mutator/runner.py @@ -0,0 +1,133 @@ +"""Run a project's tests, and decide the command for each language.""" + +from __future__ import annotations + +import os +import signal +import subprocess +import sys +import time +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class CommandResult: + code: int + timed_out: bool + seconds: float + output: str + + +class CommandRunner: + def __init__(self, verbose: bool = False): + self.verbose = verbose + + def run(self, command: str, cwd: Path, timeout: float | None) -> CommandResult: + if self.verbose: + print(f"+ ({cwd}) {command}", file=sys.stderr) + started = time.monotonic() + environment = os.environ.copy() + environment["PYTHONDONTWRITEBYTECODE"] = "1" + process = subprocess.Popen( + command, + cwd=cwd, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + start_new_session=True, + text=True, + env=environment, + ) + try: + output, _err = process.communicate(timeout=timeout) + code = process.returncode if process.returncode is not None else 1 + timed_out = False + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + output, _err = process.communicate() + code = 124 + timed_out = True + seconds = time.monotonic() - started + return CommandResult(code=code, timed_out=timed_out, seconds=seconds, output=output or "") + + +def nearest(start: Path, marker: str, stop: Path) -> Path | None: + current = start if start.is_dir() else start.parent + stop = stop.resolve() + current = current.resolve() + while True: + if (current / marker).is_file(): + return current + if current == stop or current.parent == current: + return None + current = current.parent + + +def _read(path: Path) -> str: + try: + return path.read_text(encoding="utf-8") + except OSError: + return "" + + +def _clojure_command(directory: Path) -> 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" + text = _read(deps) + if ":spec" in text and "speclj" in text: + return "clj -M:spec --tag ~no-mutate" + if ":spec" in text: + return "clj -M:spec" + return "clj -M:test" + + +def _python_command(directory: Path) -> str: + pyproject = _read(directory / "pyproject.toml") + if ( + (directory / "pytest.ini").is_file() + or (directory / "conftest.py").is_file() + or "pytest" in pyproject + or (directory / "setup.cfg").is_file() and "pytest" in _read(directory / "setup.cfg") + ): + return f"{sys.executable} -m pytest" + return f"{sys.executable} -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.""" + + root = root.resolve() + if override: + return override, root + if language == "clojure": + directory = nearest(source, "deps.edn", root) or nearest(source, "bb.edn", root) or root + return _clojure_command(directory), directory + if language == "java": + directory = nearest(source, "pom.xml", root) or root + 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 + if language == "typescript": + directory = nearest(source, "package.json", root) or root + return "npm test", directory + if language == "rust": + directory = nearest(source, "Cargo.toml", root) or root + return "cargo test", directory + if language == "python": + directory = ( + nearest(source, "pyproject.toml", root) + or nearest(source, "pytest.ini", root) + or nearest(source, "setup.cfg", root) + or root + ) + return _python_command(directory), directory + return "false", root diff --git a/src/mutator/sites.py b/src/mutator/sites.py new file mode 100644 index 0000000..43ea7c5 --- /dev/null +++ b/src/mutator/sites.py @@ -0,0 +1,254 @@ +"""Find mutation sites. + +Clojure follows clj-mutate's symbol rules. Go follows mutate4go, including +one-way `*` to `/`. Java follows mutate4java. TypeScript, Rust, and Python +use the same decisions as Java, spelled in that language. +""" + +from __future__ import annotations + +from mutator.crapper_link import ensure_crapper + +_BINARY = { + "+": "-", + "-": "+", + "*": "/", + "/": "*", + ">": ">=", + ">=": ">", + "<": "<=", + "<=": "<", + "==": "!=", + "!=": "==", + "===": "!==", + "!==": "===", + "&&": "||", + "||": "&&", + "and": "or", + "or": "and", +} + +_BINARY_PARENTS = { + "binary_expression", + "binary_operator", + "comparison_operator", + "boolean_operator", +} + +_UNARY_PARENTS = {"unary_expression", "unary_operator", "not_operator"} + +_BOOLEANS = {"true": "false", "false": "true", "True": "False", "False": "True"} + +_NUMBERS = { + "decimal_integer_literal", + "integer_literal", + "int_literal", + "integer", + "number", + "num_lit", +} + +_CLOJURE_HEAD = { + "+": ("-", "arithmetic"), + "-": ("+", "arithmetic"), + "*": ("/", "arithmetic"), + "inc": ("dec", "arithmetic"), + "dec": ("inc", "arithmetic"), + ">": (">=", "comparison"), + ">=": (">", "comparison"), + "<": ("<=", "comparison"), + "<=": ("<", "comparison"), + "=": ("not=", "equality"), + "not=": ("=", "equality"), + "if": ("if-not", "conditional"), + "if-not": ("if", "conditional"), + "when": ("when-not", "conditional"), + "when-not": ("when", "conditional"), + "and": ("or", "logical"), + "or": ("and", "logical"), + "double": ("int", "coercion"), + "int": ("double", "coercion"), + "first": ("second", "seq"), + "second": ("first", "seq"), + "filter": ("remove", "seq"), + "remove": ("filter", "seq"), + "take": ("drop", "seq"), + "drop": ("take", "seq"), + "rest": ("next", "seq"), + "next": ("rest", "seq"), + "every?": ("some", "seq"), + "some": ("every?", "seq"), + "min": ("max", "numeric"), + "max": ("min", "numeric"), + "pos?": ("neg?", "predicate"), + "neg?": ("pos?", "predicate"), + "even?": ("odd?", "predicate"), + "odd?": ("even?", "predicate"), + "nil?": ("some?", "predicate"), + "some?": ("nil?", "predicate"), +} + + +class RawSite: + def __init__(self, line: int, start: int, end: int, original: str, mutant: str, category: str): + self.line = line + self.start = start + self.end = end + self.original = original + self.mutant = mutant + self.category = category + + +def _grammar(language: str, path: str) -> str: + if language == "typescript" and path.endswith(".tsx"): + return "tsx" + if language == "go": + return "go" + return language + + +def _text(data: bytes, node) -> str: + return data[node.start_byte : node.end_byte].decode("utf-8") + + +def _category(token: str) -> str: + if token in {"+", "-", "*", "/"}: + return "arithmetic" + if token in {">", ">=", "<", "<="}: + return "comparison" + if token in {"==", "!=", "===", "!=="}: + return "equality" + if token in {"&&", "||", "and", "or"}: + return "logical" + return "arithmetic" + + +def _binary_mutant(language: str, token: str) -> str | None: + if language == "go" and token == "/": + return None + return _BINARY.get(token) + + +def _add(found: list[RawSite], node, original: str, mutant: str, category: str) -> None: + if mutant == original: + return + found.append( + RawSite( + line=node.start_point[0] + 1, + start=node.start_byte, + end=node.end_byte, + original=original, + mutant=mutant, + category=category, + ) + ) + + +def _clojure_head(node, data: bytes) -> str | None: + for child in node.children: + if child.type == "sym_lit": + return _text(data, child) + return None + + +def _is_head(node) -> bool: + parent = node.parent + if parent is None or parent.type != "list_lit": + return False + for child in parent.children: + if child.type == "sym_lit": + return child.id == node.id + return False + + +def _clojure_skipped(node, data: bytes) -> bool: + current = node + while current is not None: + if any(part in current.type for part in ("comment", "quot", "string", "str_lit", "char_lit")): + return True + if current.type == "list_lit": + head = _clojure_head(current, data) + if head in {"comment", "quote"}: + return True + current = current.parent + return False + + +def _clojure_text(node, data: bytes) -> str: + if node.child_count == 0 or node.type in {"sym_lit", "bool_lit", "num_lit"}: + return _text(data, node) + return "" + + +def _flip01(text: str) -> str: + if text == "0": + return "1" + return "0" + + +def _clojure_node(node, data: bytes, found: list[RawSite]) -> None: + if _clojure_skipped(node, data): + return + text = _clojure_text(node, data) + if node.type == "sym_lit" and _is_head(node) and text in _CLOJURE_HEAD: + mutant, category = _CLOJURE_HEAD[text] + _add(found, node, text, mutant, category) + return + if node.type == "bool_lit" and text in _BOOLEANS: + _add(found, node, text, _BOOLEANS[text], "boolean") + return + if node.type == "num_lit" and text in {"0", "1"}: + _add(found, node, text, _flip01(text), "constant") + + +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) + + +def _visit(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) + # A literal sits inside the same expression as an operator, so these + # checks are independent. `0` is a child of `a > 0`, not an operator. + if parent in _BINARY_PARENTS: + mutant = _binary_mutant(language, token) + if mutant is not None: + _add(found, node, token, mutant, _category(token)) + if token in {"!", "not"} and parent in _UNARY_PARENTS | {"comparison_operator"}: + _add(found, node, token, "", "unary") + if token == "-" and parent in _UNARY_PARENTS: + _add(found, node, token, "", "unary") + if node.type in {"true", "false"} and token in _BOOLEANS: + _add(found, node, token, _BOOLEANS[token], "boolean") + if node.type in _NUMBERS and token in {"0", "1"}: + _add(found, node, token, "1" if token == "0" else "0", "constant") + 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 discover_raw(source: str, language: str, path: str) -> list[RawSite]: + """Mutation sites in source order. Offsets are UTF-8 byte offsets.""" + + crapper = ensure_crapper() + data = source.encode("utf-8") + grammar = _grammar(language, path) + _data, tree = crapper.languages.treesitter.parse(source, grammar) + found: list[RawSite] = [] + if language == "clojure": + _visit_clojure(tree.root_node, data, found) + else: + _visit(tree.root_node, data, language, found) + found.sort(key=lambda site: (site.line, site.start, site.end)) + return found + + +def apply_site(source: str, start: int, end: int, mutant: str) -> str: + data = source.encode("utf-8") + updated = data[:start] + mutant.encode("utf-8") + data[end:] + return updated.decode("utf-8") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..f118ca2 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,94 @@ +import subprocess +import sys + +from mutator.cli import parse_args, run +from mutator.runner import CommandResult + + +def test_lines_option_parses_positive_numbers(): + options = parse_args(["--lines", "2, 4", "src/demo.py"]) + assert options.lines == {2, 4} + + +def test_changed_selects_a_dirty_source_file(tmp_path, monkeypatch, capsys): + path = tmp_path / "src" / "demo.py" + path.parent.mkdir() + path.write_text("def place(x):\n return x > 0\n", encoding="utf-8") + subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) + subprocess.run(["git", "add", "src/demo.py"], cwd=tmp_path, check=True, capture_output=True) + monkeypatch.chdir(tmp_path) + code = run(["--scan", "--no-coverage", "--changed", "--root", str(tmp_path)]) + captured = capsys.readouterr() + assert code == 0 + assert "demo.py:2" in captured.out + + +def test_a_directory_argument_skips_a_named_test_file(tmp_path, capsys): + src = tmp_path / "pkg" / "app.py" + src.parent.mkdir() + src.write_text("def place(x):\n return x > 0\n", encoding="utf-8") + test = tmp_path / "pkg" / "test_app.py" + test.write_text("def test_place():\n assert True\n", encoding="utf-8") + code = run( + ["--scan", "--no-coverage", "--root", str(tmp_path), str(tmp_path / "pkg"), str(test)] + ) + captured = capsys.readouterr() + assert code == 0 + assert "Skipping test file" in captured.err + assert "app.py:2" in captured.out + + +def test_help_and_conflicting_flags(): + help_options = parse_args(["--help"]) + assert help_options.exit_code == 0 + assert "uml-viewer" in help_options.message + conflict = parse_args(["--mutate-all", "--since-last-run", "src/demo.py"]) + assert conflict.exit_code == 1 + assert run(["--help"]) == 0 + + +def test_scan_lists_sites_without_writing_metrics(tmp_path, monkeypatch): + path = tmp_path / "src" / "demo.py" + path.parent.mkdir() + path.write_text("def place(x):\n return x > 0\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + code = run(["--scan", "--no-coverage", "--root", str(tmp_path), "src/demo.py"]) + assert code == 0 + assert not (tmp_path / ".metrics").exists() + + +def test_real_python_mutants_update_the_snapshot(tmp_path, capsys): + path = tmp_path / "src" / "demo.py" + path.parent.mkdir() + path.write_text( + "def add(a, b):\n" + " if a > 0:\n" + " return a + b\n" + " return 0\n", + encoding="utf-8", + ) + command = ( + f"{sys.executable} -c \"import sys; sys.path.insert(0, 'src'); " + "from demo import add; raise SystemExit(0 if add(2, 3) == 5 else 1)\"" + ) + code = run( + [ + "--no-coverage", + "--mutate-all", + "--root", + str(tmp_path), + "--test-command", + command, + "src/demo.py", + ] + ) + captured = capsys.readouterr() + assert code == 3 + assert "SURVIVED" in captured.out + assert "KILLED" in captured.out + snapshot = tmp_path / ".metrics" / "mutate" / "demo.edn" + assert snapshot.is_file() + text = snapshot.read_text(encoding="utf-8") + assert ":namespace \"demo\"" in text + assert ":id \"defn/add\"" in text + assert path.read_text(encoding="utf-8").startswith("def add") diff --git a/tests/test_coverage.py b/tests/test_coverage.py new file mode 100644 index 0000000..de2d5a2 --- /dev/null +++ b/tests/test_coverage.py @@ -0,0 +1,57 @@ +from mutator.coverage import covered_lines + + +def test_jacoco_marks_lines_with_covered_instructions(tmp_path): + report = tmp_path / "target" / "site" / "jacoco" / "jacoco.xml" + report.parent.mkdir(parents=True) + report.write_text( + """ + + + + + + + + + +""", + encoding="utf-8", + ) + source = tmp_path / "src" / "demo" / "Board.java" + source.parent.mkdir(parents=True) + source.write_text("package demo;\nclass Board {}\n", encoding="utf-8") + assert covered_lines(tmp_path, source, "java") == {3, 5} + + +def test_go_profile_marks_hit_lines(tmp_path): + profile = tmp_path / "target" / "coverage" / "go" / "coverage.out" + profile.parent.mkdir(parents=True) + profile.write_text( + "mode: set\n" + "demo/widget.go:2.1,4.2 1 1\n" + "demo/widget.go:5.1,6.2 1 0\n", + encoding="utf-8", + ) + source = tmp_path / "demo" / "widget.go" + source.parent.mkdir() + source.write_text("package demo\nfunc Run() {}\n", encoding="utf-8") + assert covered_lines(tmp_path, source, "go") == {2, 3, 4} + assert covered_lines(tmp_path, tmp_path / "other" / "missing.go", "go") is None + empty = tmp_path / "empty" + empty.mkdir() + assert covered_lines(empty, empty / "demo" / "widget.go", "go") is None + + +def test_lcov_marks_executed_lines(tmp_path): + report = tmp_path / "target" / "coverage" / "python" / "lcov.info" + report.parent.mkdir(parents=True) + report.write_text( + "SF:src/demo/app.py\nDA:1,3\nDA:2,0\nend_of_record\n", + encoding="utf-8", + ) + source = tmp_path / "src" / "demo" / "app.py" + source.parent.mkdir(parents=True) + 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} diff --git a/tests/test_edn.py b/tests/test_edn.py new file mode 100644 index 0000000..ace328e --- /dev/null +++ b/tests/test_edn.py @@ -0,0 +1,55 @@ +from mutator.edn import dumps, keyword, loads + + +def test_round_trip_matches_the_snapshot_shape(): + payload = { + "version": 1, + "tested-at": "2026-09-30T12:00:00-05:00", + "source": 'src/demo/say "hi".py', + "namespace": "demo.Board", + "outcomes": { + '["src/a.py","demo.Board","defn/place",1,2,">",">="]': keyword("killed"), + '["src/a.py","demo.Board","defn-/hide",3,4,"true","false"]': keyword("survived"), + }, + "forms": [ + { + "id": "defn/place", + "kind": "defn", + "file": "src/demo/Board.java", + "line": 4, + "end-line": 9, + "hash": "abc", + "killed": 1, + "survived": 0, + "uncovered": 2, + "sites": 3, + } + ], + } + text = dumps(payload) + assert text.startswith("{") + assert ":namespace" in text + assert ":killed" in text + parsed = loads(text) + assert parsed["namespace"] == "demo.Board" + assert parsed["forms"][0]["id"] == "defn/place" + assert parsed["forms"][0]["end-line"] == 9 + assert parsed["forms"][0]["killed"] == 1 + assert set(parsed["outcomes"].values()) == {"killed", "survived"} + + +def test_empty_collections_and_nil(): + assert loads("{:forms [] :outcomes {} :note nil}") == { + "forms": [], + "outcomes": {}, + "note": None, + } + + +def test_atoms_cover_booleans_numbers_and_symbols(): + assert loads("true") is True + assert loads("false") is False + assert loads("12") == 12 + assert loads("-3") == -3 + assert loads("1.50") == 1.5 + assert loads("foo") == "foo" diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..d15b79b --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,147 @@ +from mutator.engine import mutate_file, restore_backups +from mutator.edn import loads +from mutator.metrics import snapshot_path +from mutator.runner import CommandResult + + +class FakeRunner: + def __init__(self): + self.verbose = False + self.calls = 0 + + def run(self, command, cwd, timeout): + self.calls += 1 + text = (cwd / "src" / "demo.py").read_text(encoding="utf-8") + namespace: dict = {} + exec(text, namespace) + ok = namespace["add"](1, 2) == 3 + return CommandResult(code=0 if ok else 1, timed_out=False, seconds=0.01, output="") + + +SOURCE = """def add(a, b): + if a > 0: + return a + b + return a + b +""" + + +def test_mutants_are_killed_or_kept_and_the_snapshot_is_differential(tmp_path): + path = tmp_path / "src" / "demo.py" + path.parent.mkdir() + path.write_text(SOURCE, encoding="utf-8") + original = path.read_bytes() + runner = FakeRunner() + first = mutate_file( + path, + tmp_path, + runner=runner, + covered_lines=None, + ignore_coverage=True, + mutate_all=True, + lines=None, + test_command="fake", + timeout_factor=10, + mutation_warning=50, + baselines={}, + ) + assert path.read_bytes() == original + assert first.baseline_failed is False + form = first.forms[0] + assert form.id == "defn/add" + assert form.namespace == "demo" + # `>` and the unexecuted `+` survive. The executed `+` is killed. `0` -> `1` + # still returns 3 for add(1, 2). + assert form.killed == 1 + assert form.survived == 3 + assert form.sites == 4 + assert form.killed + form.survived + form.uncovered == form.sites + data = loads(snapshot_path(tmp_path, "demo").read_text(encoding="utf-8")) + assert data["namespace"] == "demo" + assert data["forms"][0]["id"] == "defn/add" + assert data["forms"][0]["killed"] == 1 + assert data["forms"][0]["survived"] == 3 + first_calls = runner.calls + + second = mutate_file( + path, + tmp_path, + runner=runner, + covered_lines=None, + ignore_coverage=True, + mutate_all=False, + lines=None, + test_command="fake", + timeout_factor=10, + mutation_warning=50, + baselines={}, + ) + assert second.forms[0].killed == 1 + assert second.forms[0].survived == 3 + # Baseline plus the three survivors. The killed mutant stays in the snapshot. + assert runner.calls - first_calls == 4 + + +def test_uncovered_sites_are_not_executed(tmp_path): + path = tmp_path / "src" / "demo.py" + path.parent.mkdir() + path.write_text(SOURCE, encoding="utf-8") + runner = FakeRunner() + result = mutate_file( + path, + tmp_path, + runner=runner, + covered_lines=set(), + ignore_coverage=False, + mutate_all=True, + lines=None, + test_command="fake", + timeout_factor=10, + mutation_warning=50, + baselines={}, + ) + assert runner.calls == 0 + assert result.forms[0].uncovered == 4 + assert result.forms[0].killed == 0 + assert result.forms[0].sites == 4 + + +def test_a_failed_baseline_does_not_rewrite_metrics(tmp_path): + path = tmp_path / "src" / "demo.py" + path.parent.mkdir() + path.write_text(SOURCE, encoding="utf-8") + + class Red: + verbose = False + + def run(self, command, cwd, timeout): + return CommandResult(code=1, timed_out=False, seconds=0.01, output="nope") + + result = mutate_file( + path, + tmp_path, + runner=Red(), + covered_lines=None, + ignore_coverage=True, + mutate_all=True, + lines=None, + test_command="fake", + timeout_factor=10, + mutation_warning=50, + baselines={}, + ) + assert result.baseline_failed + assert not snapshot_path(tmp_path, "demo").exists() + + +def test_restore_backups_puts_an_interrupted_mutant_back(tmp_path): + source = tmp_path / "src" / "demo.py" + source.parent.mkdir() + source.write_text("original\n", encoding="utf-8") + backup = tmp_path / "target" / "mutator-backup" / "src" / "demo.py" + backup.parent.mkdir(parents=True) + backup.write_text("original\n", encoding="utf-8") + source.write_text("mutated\n", encoding="utf-8") + restored = restore_backups(tmp_path) + assert restored == [source] + assert source.read_text(encoding="utf-8") == "original\n" + assert not backup.exists() diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..f53397e --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,132 @@ +from mutator.edn import dumps, loads +from mutator.metrics import load_history, snapshot_path, write_results +from mutator.model import FormResult + + +def _form(namespace, name, file_name, digest, killed=1, private=False): + form_id = f"defn-/{name}" if private else f"defn/{name}" + return FormResult( + id=form_id, + namespace=namespace, + name=name, + private=private, + file=file_name, + line=2, + end_line=4, + digest=digest, + killed=killed, + survived=0, + uncovered=1, + sites=2, + ) + + +def _viewer_name(form_id): + """The same split uml-viewer.application.overlay/form-name performs.""" + + if form_id.startswith("defn-/"): + return form_id[6:], True + if form_id.startswith("defn/"): + return form_id[5:], False + return None, False + + +def test_one_snapshot_per_namespace_preserves_the_other_file(tmp_path): + write_results( + tmp_path, + "a.go", + [_form("demo", "Run", "a.go", "hash-a")], + {}, + ) + write_results( + tmp_path, + "b.go", + [_form("demo", "Stop", "b.go", "hash-b", killed=4)], + {}, + ) + path = snapshot_path(tmp_path, "demo") + data = loads(path.read_text(encoding="utf-8")) + assert data["namespace"] == "demo" + names = [] + for form in data["forms"]: + operation, private = _viewer_name(form["id"]) + names.append((operation, private, form["killed"], form["uncovered"], form["sites"])) + assert ("Run", False, 1, 1, 2) in names + assert ("Stop", False, 4, 1, 2) in names + assert list(tmp_path.joinpath(".metrics", "mutate").rglob("*.edn")) == [path] + + write_results( + tmp_path, + "a.go", + [_form("demo", "Run", "a.go", "hash-a", killed=9)], + {}, + ) + again = loads(path.read_text(encoding="utf-8")) + by_id = {form["id"]: form for form in again["forms"]} + assert by_id["defn/Run"]["killed"] == 9 + assert by_id["defn/Stop"]["killed"] == 4 + + +def test_private_form_id_and_history_round_trip(tmp_path): + outcome = '["hide.clj","demo.core","defn-/hide",8,12,"false","true"]' + write_results( + tmp_path, + "hide.clj", + [_form("demo.core", "hide", "hide.clj", "digest", private=True)], + {outcome: "killed"}, + ) + data = loads(snapshot_path(tmp_path, "demo.core").read_text(encoding="utf-8")) + assert _viewer_name(data["forms"][0]["id"]) == ("hide", True) + history = load_history(tmp_path, "hide.clj") + assert history.forms[("demo.core", "defn-/hide")].digest == "digest" + assert history.outcomes[outcome] == "killed" + + +def test_snapshot_paths_follow_the_namespace(tmp_path): + assert snapshot_path(tmp_path, "example.com/demo.Widget") == ( + tmp_path / ".metrics" / "mutate" / "example.com" / "demo.Widget.edn" + ) + assert snapshot_path(tmp_path, "crate::foo::Bar") == ( + tmp_path / ".metrics" / "mutate" / "crate" / "foo" / "Bar.edn" + ) + + +def test_a_non_canonical_snapshot_is_folded_into_the_namespace_file(tmp_path): + stale = tmp_path / ".metrics" / "mutate" / "legacy.edn" + stale.parent.mkdir(parents=True) + stale.write_text( + dumps( + { + "version": 1, + "namespace": "demo", + "source": "b.go", + "forms": [ + { + "id": "defn/Stop", + "file": "b.go", + "line": 3, + "hash": "hash-b", + "killed": 4, + "survived": 0, + "uncovered": 1, + "sites": 2, + } + ], + "outcomes": {}, + } + ) + + "\n", + encoding="utf-8", + ) + write_results(tmp_path, "a.go", [_form("demo", "Run", "a.go", "hash-a")], {}) + assert not stale.exists() + data = loads(snapshot_path(tmp_path, "demo").read_text(encoding="utf-8")) + by_id = {form["id"]: form for form in data["forms"]} + assert by_id["defn/Stop"]["killed"] == 4 + assert by_id["defn/Run"]["file"] == "a.go" + + +def test_dropping_a_file_removes_its_namespace_when_nothing_else_remains(tmp_path): + write_results(tmp_path, "a.go", [_form("demo", "Run", "a.go", "hash-a")], {}) + write_results(tmp_path, "a.go", [], {}) + assert not snapshot_path(tmp_path, "demo").exists() diff --git a/tests/test_runner.py b/tests/test_runner.py new file mode 100644 index 0000000..2fde75d --- /dev/null +++ b/tests/test_runner.py @@ -0,0 +1,68 @@ +from mutator.runner import _clojure_command, _python_command, nearest + + +def test_nearest_walks_up_to_the_marker(tmp_path): + root = tmp_path / "proj" + nested = root / "src" / "demo" + nested.mkdir(parents=True) + (root / "pyproject.toml").write_text("", encoding="utf-8") + source = nested / "app.py" + source.write_text("x = 1\n", encoding="utf-8") + assert nearest(source, "pyproject.toml", root) == root + assert nearest(nested, "pyproject.toml", root) == root + assert nearest(source, "missing.toml", root) is None + + +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" + + 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" + + 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" + + 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" + + plain = tmp_path / "plain" + plain.mkdir() + (plain / "deps.edn").write_text("{:deps {}}", encoding="utf-8") + assert _clojure_command(plain) == "clj -M:test" + + +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") + + unit = tmp_path / "unit" + unit.mkdir() + assert _python_command(unit).endswith("-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") + + 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") + + conf = tmp_path / "conf" + conf.mkdir() + (conf / "conftest.py").write_text("", encoding="utf-8") + assert _python_command(conf).endswith("-m pytest") diff --git a/tests/test_sites.py b/tests/test_sites.py new file mode 100644 index 0000000..1a36f1a --- /dev/null +++ b/tests/test_sites.py @@ -0,0 +1,165 @@ +from pathlib import Path + +from mutator.functions import form_id, is_private, project_functions, sites_in_file + + +def _sites(tmp_path: Path, relative: str, source: str): + path = tmp_path / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8") + return sites_in_file(source, path, tmp_path, relative) + + +def test_clojure_rules_follow_clj_mutate_and_mark_private_defns(tmp_path): + source = """(ns demo.core) +(defn place [x] + (if (> x 0) + (+ x 1) + 0)) +(defn- hide [] + false) +""" + sites = _sites(tmp_path, "src/demo/core.clj", source) + described = {(site.form_id, site.original, site.mutant) for site in sites} + assert ("defn/place", "if", "if-not") in described + assert ("defn/place", ">", ">=") in described + assert ("defn/place", "+", "-") in described + assert ("defn/place", "0", "1") in described + assert ("defn/place", "1", "0") in described + assert ("defn-/hide", "false", "true") in described + assert {site.namespace for site in sites} == {"demo.core"} + functions = project_functions(source, tmp_path / "src/demo/core.clj", tmp_path) + assert form_id("hide", True) == "defn-/hide" + assert is_private("clojure", "hide", source, 6, 7) + assert {fn.namespace for fn in functions} == {"demo.core"} + + +def test_clojure_ignores_strings_comments_and_quotes(tmp_path): + source = """(ns demo.core) +(defn place [] + (comment (+ 1 0)) + (quote (+ 1 0)) + (str "true + 1")) +""" + sites = _sites(tmp_path, "src/demo/core.clj", source) + assert sites == [] + + +def test_java_rules_follow_mutate4java(tmp_path): + source = """package demo; +class Board { + int place(int x) { + if (x > 0 && ready) + return x + 1; + return !ok ? -x : 0; + } + private int hide() { return 1; } +} +""" + sites = _sites(tmp_path, "src/demo/Board.java", source) + public = [site for site in sites if site.form_id == "defn/place"] + private = [site for site in sites if site.form_id == "defn-/hide"] + assert ("+", "-") in {(site.original, site.mutant) for site in public} + assert ("/", "*") not in {(site.original, site.mutant) for site in public} + assert ("*", "/") not in {(site.original, site.mutant) for site in public} + assert (">", ">=") in {(site.original, site.mutant) for site in public} + assert ("&&", "||") in {(site.original, site.mutant) for site in public} + assert ("!", "") in {(site.original, site.mutant) for site in public} + assert ("-", "") in {(site.original, site.mutant) for site in public} + assert ("0", "1") in {(site.original, site.mutant) for site in public} + assert private == [] or {site.original for site in private} == {"1"} + assert {site.namespace for site in sites} == {"demo.Board"} + + +def test_java_swaps_division_both_ways(tmp_path): + source = """package demo; +class Math { + int place(int a, int b) { return a * b + a / b; } +} +""" + sites = _sites(tmp_path, "src/demo/Math.java", source) + pairs = {(site.original, site.mutant) for site in sites} + assert ("*", "/") in pairs + assert ("/", "*") in pairs + assert ("+", "-") in pairs + + +def test_go_matches_mutate4go_and_splits_receivers(tmp_path): + source = """package demo +func Run(a int, b int) int { + if a > 0 && b == 1 { + return a * b + } + return a / b +} +type Widget struct{} +func (w Widget) Run() bool { return true } +""" + sites = _sites(tmp_path, "demo.go", source) + free = [site for site in sites if site.namespace == "demo"] + method = [site for site in sites if site.namespace == "demo.Widget"] + pairs = {(site.original, site.mutant) for site in free} + assert ("*", "/") in pairs + assert ("/", "*") not in pairs + assert ("&&", "||") in pairs + assert ("==", "!=") in pairs + assert (">", ">=") in pairs + assert ("0", "1") in pairs + assert ("1", "0") in pairs + assert {site.form_id for site in method} == {"defn/Run"} + assert ("true", "false") in {(site.original, site.mutant) for site in method} + assert {site.form_id for site in free} == {"defn/Run"} + + +def test_go_unexported_function_is_private(tmp_path): + source = """package demo +func hide() bool { return false } +""" + sites = _sites(tmp_path, "demo.go", source) + assert sites[0].form_id == "defn-/hide" + + +def test_python_typescript_and_rust_use_the_java_decisions(tmp_path): + python = _sites( + tmp_path, + "src/demo/app.py", + "def place(x):\n if x > 0 and ready:\n return x + 1\n if not ok:\n return 0\n", + ) + typescript = _sites( + tmp_path, + "src/demo/app.ts", + "export function place(x: number) {\n if (x > 0 && ready) return x + 1;\n return true === false;\n}\n", + ) + rust = _sites( + tmp_path, + "src/lib.rs", + "fn place(x: i32) -> i32 {\n if x > 0 && ready { return x + 1; }\n if !ok { return -x; }\n 0\n}\n", + ) + assert ("and", "or") in {(site.original, site.mutant) for site in python} + assert ("not", "") in {(site.original, site.mutant) for site in python} + assert ("True", "False") not in {(site.original, site.mutant) for site in python} + assert {site.namespace for site in python} == {"demo.app"} + assert ("===", "!==") in {(site.original, site.mutant) for site in typescript} + assert ("&&", "||") in {(site.original, site.mutant) for site in typescript} + assert {site.namespace for site in typescript} == {"demo.app"} + assert ("&&", "||") in {(site.original, site.mutant) for site in rust} + assert ("!", "") in {(site.original, site.mutant) for site in rust} + assert ("-", "") in {(site.original, site.mutant) for site in rust} + + +def test_python_private_and_class_namespace(tmp_path): + source = """class Board: + def place(self, x): + return x > 0 + def _hide(self): + return False +""" + sites = _sites(tmp_path, "src/demo/board.py", source) + assert {site.form_id for site in sites if site.name == "place"} == {"defn/place"} + assert {site.form_id for site in sites if site.name == "_hide"} == {"defn-/_hide"} + assert {site.namespace for site in sites} == {"demo.board.Board"} + + +def test_string_operators_are_not_sites(tmp_path): + source = 'def place():\n return "x > 0 and true"\n' + assert _sites(tmp_path, "src/demo/app.py", source) == []