mirror of
https://github.com/unclebob/mutator.git
synced 2026-11-06 18:23:50 +00:00
Add a multi-language mutation tester that writes .metrics/mutate for uml-viewer.
This commit is contained in:
Robert C. Martin
@@ -0,0 +1,9 @@
|
||||
.venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.pytest_cache/
|
||||
.metrics/
|
||||
target/
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
@@ -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/<namespace>.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
|
||||
```
|
||||
@@ -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 "$@"
|
||||
@@ -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"]
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Mutation testing for the languages crapper scores."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,4 @@
|
||||
from mutator.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -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/<name>` or, for a private operation, `defn-/<name>`.
|
||||
uml-viewer joins that name to the operation and joins `:namespace` to the
|
||||
class.
|
||||
|
||||
Options:
|
||||
-h, --help Print this help and exit.
|
||||
--root <path> Project root. Metrics are written here.
|
||||
Default: the current directory.
|
||||
-s, --source-root <path> 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 <n,n,...> 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 <cmd> Run this command instead of the per-language
|
||||
coverage tools, then read the reports it wrote.
|
||||
--test-command <cmd> Run this command for the baseline and every
|
||||
mutant. The working directory is the project
|
||||
root.
|
||||
--timeout-factor <number> Mutant timeout, as a multiple of the baseline
|
||||
duration. Default: 10. The timeout is at least
|
||||
2 seconds.
|
||||
--mutation-warning <number> 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))
|
||||
@@ -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))
|
||||
@@ -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
|
||||
@@ -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())
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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/<name>`
|
||||
or `defn-/<name>`, 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()
|
||||
@@ -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 = ""
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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")
|
||||
@@ -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")
|
||||
@@ -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(
|
||||
"""<?xml version="1.0"?>
|
||||
<report name="demo">
|
||||
<package name="demo">
|
||||
<sourcefile name="Board.java">
|
||||
<line nr="3" mi="0" ci="2" mb="0" cb="0"/>
|
||||
<line nr="4" mi="4" ci="0" mb="1" cb="0"/>
|
||||
<line nr="5" mi="0" ci="0" mb="0" cb="1"/>
|
||||
</sourcefile>
|
||||
</package>
|
||||
</report>
|
||||
""",
|
||||
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}
|
||||
@@ -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"
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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")
|
||||
@@ -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) == []
|
||||
Reference in New Issue
Block a user