mirror of
https://github.com/unclebob/mutator.git
synced 2026-11-06 18:23:50 +00:00
303 lines
10 KiB
Python
303 lines
10 KiB
Python
import os
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
|
|
from mutator.engine import mutate_file, restore_backups
|
|
from mutator.edn import loads
|
|
from mutator.metrics import snapshot_path
|
|
from mutator.runner import CommandResult
|
|
from mutator.workers import create_workers, delete_tree, new_run_dir, worker_count
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self):
|
|
self.verbose = False
|
|
self.calls = 0
|
|
self._lock = threading.Lock()
|
|
|
|
def run(self, command, cwd, timeout):
|
|
with self._lock:
|
|
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_site_inside_a_covered_multi_line_statement_is_run(tmp_path):
|
|
# LCOV marks only the line a statement starts on.
|
|
path = tmp_path / "src" / "demo.py"
|
|
path.parent.mkdir()
|
|
path.write_text("def add(a, b):\n return (a\n + b)\n", encoding="utf-8")
|
|
runner = FakeRunner()
|
|
result = mutate_file(
|
|
path,
|
|
tmp_path,
|
|
runner=runner,
|
|
covered_lines={2},
|
|
ignore_coverage=False,
|
|
mutate_all=True,
|
|
lines=None,
|
|
test_command="fake",
|
|
timeout_factor=10,
|
|
mutation_warning=50,
|
|
baselines={},
|
|
)
|
|
assert result.forms[0].uncovered == 0
|
|
assert result.forms[0].killed == 1
|
|
|
|
|
|
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_worker_count_follows_sites_cores_and_the_requested_cap():
|
|
cores = os.cpu_count() or 1
|
|
assert worker_count(10, None) == min(10, cores)
|
|
assert worker_count(10, 3) == min(3, cores)
|
|
assert worker_count(2, 8) == min(2, cores)
|
|
|
|
|
|
def test_workers_keep_a_private_copy_and_link_the_rest(tmp_path):
|
|
source = tmp_path / "src" / "demo.py"
|
|
source.parent.mkdir()
|
|
source.write_text("def place(x):\n return x > 0\n", encoding="utf-8")
|
|
sibling = tmp_path / "src" / "other.py"
|
|
sibling.write_text("kept = True\n", encoding="utf-8")
|
|
marker = tmp_path / "tests" / "keep.txt"
|
|
marker.parent.mkdir()
|
|
marker.write_text("keep", encoding="utf-8")
|
|
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n", encoding="utf-8")
|
|
original = source.read_bytes()
|
|
|
|
base = new_run_dir(tmp_path)
|
|
try:
|
|
workers = create_workers(base, tmp_path, "src/demo.py", original, 2)
|
|
assert len(workers) == 2
|
|
for worker in workers:
|
|
private = worker / "src" / "demo.py"
|
|
assert private.is_file()
|
|
assert not private.is_symlink()
|
|
assert private.read_bytes() == original
|
|
assert (worker / "src" / "other.py").is_symlink()
|
|
assert (worker / "tests").is_symlink()
|
|
assert (worker / "tests" / "keep.txt").read_text(encoding="utf-8") == "keep"
|
|
assert (worker / "pyproject.toml").is_file()
|
|
assert not (worker / "pyproject.toml").is_symlink()
|
|
(workers[0] / "src" / "demo.py").write_text("changed\n", encoding="utf-8")
|
|
assert workers[1].joinpath("src/demo.py").read_bytes() == original
|
|
assert source.read_bytes() == original
|
|
finally:
|
|
delete_tree(base)
|
|
|
|
assert source.read_bytes() == original
|
|
assert sibling.read_text(encoding="utf-8") == "kept = True\n"
|
|
assert marker.read_text(encoding="utf-8") == "keep"
|
|
assert not base.exists()
|
|
|
|
|
|
def test_node_imports_the_worker_copy_through_a_relative_specifier(tmp_path):
|
|
source = tmp_path / "src" / "find.mjs"
|
|
source.parent.mkdir()
|
|
source.write_text('export const value = "real"\n', encoding="utf-8")
|
|
beside = tmp_path / "src" / "find.test.mjs"
|
|
beside.write_text('import { value } from "./find.mjs";\nconsole.log(value)\n', encoding="utf-8")
|
|
(tmp_path / "src" / "notes.mjs").write_text("export const notes = 1\n", encoding="utf-8")
|
|
main = tmp_path / "src" / "main.mjs"
|
|
main.write_text('import { value } from "./find.mjs";\nexport const main = value\n', encoding="utf-8")
|
|
(tmp_path / "src" / "app.test.mjs").write_text('import { main } from "./main.mjs";\n', encoding="utf-8")
|
|
distant = tmp_path / "tests" / "use.mjs"
|
|
distant.parent.mkdir()
|
|
distant.write_text('import { value } from "../src/find.mjs";\nconsole.log(value)\n', encoding="utf-8")
|
|
(tmp_path / "tests" / "other.test.mjs").write_text("export const n = 1\n", encoding="utf-8")
|
|
|
|
base = new_run_dir(tmp_path)
|
|
try:
|
|
worker = create_workers(base, tmp_path, "src/find.mjs", source.read_bytes(), 1)[0]
|
|
private = worker / "src" / "find.mjs"
|
|
private.write_text('export const value = "worker"\n', encoding="utf-8")
|
|
for relative in ("src/find.test.mjs", "src/main.mjs", "src/app.test.mjs", "tests/use.mjs"):
|
|
copied = worker / relative
|
|
assert copied.is_file()
|
|
assert not copied.is_symlink()
|
|
assert (worker / "src" / "notes.mjs").is_symlink()
|
|
assert not (worker / "tests").is_symlink()
|
|
assert (worker / "tests" / "other.test.mjs").is_symlink()
|
|
if shutil.which("node"):
|
|
for script in ("src/find.test.mjs", "tests/use.mjs"):
|
|
completed = subprocess.run(
|
|
["node", script],
|
|
cwd=worker,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert completed.returncode == 0
|
|
assert completed.stdout.strip() == "worker"
|
|
finally:
|
|
delete_tree(base)
|
|
|
|
assert beside.read_text(encoding="utf-8").startswith("import")
|
|
assert distant.read_text(encoding="utf-8").startswith("import")
|
|
assert source.read_text(encoding="utf-8") == 'export const value = "real"\n'
|
|
|
|
|
|
def test_a_mutation_run_leaves_the_project_tree_unchanged(tmp_path):
|
|
path = tmp_path / "src" / "demo.py"
|
|
path.parent.mkdir()
|
|
path.write_text("def place(x):\n return x > 0\n", encoding="utf-8")
|
|
original = path.read_text(encoding="utf-8")
|
|
|
|
class Record:
|
|
verbose = False
|
|
|
|
def run(self, command, cwd, timeout):
|
|
if timeout is not None:
|
|
assert "mutation-workers" in cwd.as_posix()
|
|
assert (cwd / "src" / "demo.py").read_text(encoding="utf-8") != original
|
|
return CommandResult(code=0 if timeout is None else 1, timed_out=False, seconds=0.01, output="")
|
|
|
|
mutate_file(
|
|
path,
|
|
tmp_path,
|
|
runner=Record(),
|
|
covered_lines=None,
|
|
ignore_coverage=True,
|
|
mutate_all=True,
|
|
lines=None,
|
|
test_command="fake",
|
|
timeout_factor=10,
|
|
mutation_warning=50,
|
|
baselines={},
|
|
max_workers=2,
|
|
)
|
|
assert path.read_text(encoding="utf-8") == original
|
|
assert [item for item in (tmp_path / "target").rglob("*") if item.is_file()] == []
|
|
|
|
|
|
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()
|