404 lines
17 KiB
Python
404 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""diffkit self-test.
|
|
|
|
Three layers, in order of evidential strength:
|
|
|
|
1. hand cases -- targeted edge cases (no-newline EOF, create/delete,
|
|
fuzz, offset drift, malformed hunks, CRLF, multi-file)
|
|
2. property suite -- randomised base/target pairs, diffs produced by the
|
|
real GNU `diff`, applied by diffkit, compared
|
|
byte-for-byte against the true target; GNU `patch`
|
|
is run on the same input as a second oracle
|
|
3. mutation control -- three deliberately broken appliers must FAIL the same
|
|
suites. If a mutant passes, the suite is vacuous and
|
|
this script exits nonzero.
|
|
|
|
Exit code 0 iff every check holds. No third-party dependencies.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import random
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
import diffkit # noqa: E402
|
|
|
|
PLAIN_CASES = 120
|
|
DRIFT_CASES = 80
|
|
ALPHA = ["alpha", "beta", "gamma", "delta", "eps", "zeta", "", " indented",
|
|
"x = 1", "return", "}", "{", "TODO"]
|
|
|
|
_results: list[tuple[str, bool, str]] = []
|
|
|
|
|
|
def check(name: str, ok: bool, detail: str = "") -> bool:
|
|
_results.append((name, bool(ok), detail))
|
|
print("%s %s%s" % ("PASS" if ok else "FAIL", name,
|
|
"" if ok else (" :: " + detail)))
|
|
return bool(ok)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# helpers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def write(path: str, text: str) -> None:
|
|
with open(path, "w", encoding="utf-8", newline="") as fh:
|
|
fh.write(text)
|
|
|
|
|
|
def run_bytes(args: list[str], cwd: str) -> tuple[int, str, str]:
|
|
"""Run a command WITHOUT universal-newline translation (text=True would eat \\r)."""
|
|
r = subprocess.run(args, cwd=cwd, capture_output=True)
|
|
return (r.returncode,
|
|
r.stdout.decode("utf-8", "surrogateescape"),
|
|
r.stderr.decode("utf-8", "surrogateescape"))
|
|
|
|
|
|
def gnu_diff(tmp: str, a_name: str, b_name: str) -> str:
|
|
rc, out, err = run_bytes(["diff", "-u", a_name, b_name], tmp)
|
|
if rc not in (0, 1):
|
|
raise RuntimeError("diff failed: %s" % err)
|
|
return out
|
|
|
|
|
|
def gnu_patch_apply(tmp: str, diff_text: str, src_text: str,
|
|
fname: str = "base.txt") -> str | None:
|
|
"""Run GNU patch on src_text with diff_text; None if patch itself refused."""
|
|
d = os.path.join(tmp, "patchcopy")
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
os.makedirs(d)
|
|
write(os.path.join(d, fname), src_text)
|
|
write(os.path.join(d, "d.diff"), diff_text)
|
|
rc, _, _ = run_bytes(["patch", "--batch", "-p0", "-i", "d.diff"], d)
|
|
if rc != 0:
|
|
return None
|
|
with open(os.path.join(d, fname), "r", encoding="utf-8", newline="") as fh:
|
|
return fh.read()
|
|
|
|
|
|
def rand_text(rng: random.Random) -> str:
|
|
n = rng.randint(4, 40)
|
|
text = "".join(rng.choice(ALPHA) + "\n" for _ in range(n))
|
|
if rng.random() < 0.25: # sometimes no newline at EOF
|
|
text = text[:-1]
|
|
return text
|
|
|
|
|
|
def mutate(rng: random.Random, base: str) -> str:
|
|
lines = diffkit.split_lines(base)
|
|
for _ in range(rng.randint(1, 5)):
|
|
op = rng.choice(["ins", "del", "rep"])
|
|
if not lines:
|
|
op = "ins"
|
|
i = rng.randint(0, len(lines))
|
|
if op == "ins":
|
|
lines[i:i] = [rng.choice(ALPHA) + "\n" for _ in range(rng.randint(1, 3))]
|
|
elif op == "del":
|
|
del lines[i:min(len(lines), i + rng.randint(1, 3))]
|
|
else:
|
|
j = min(len(lines), i + rng.randint(1, 3))
|
|
if j > i:
|
|
lines[i:j] = [rng.choice(ALPHA) + "\n" for _ in range(j - i)]
|
|
return "".join(lines)
|
|
|
|
|
|
def rand_prefix(rng: random.Random) -> str:
|
|
return "".join("PREFIX-%d-%d\n" % (rng.randrange(10 ** 6), i)
|
|
for i in range(rng.randint(1, 4)))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 1. hand cases
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def hand_cases(tmp: str) -> None:
|
|
def apply_diff(a: str, b: str, src: str | None = None, **kw) -> str:
|
|
write(os.path.join(tmp, "base.txt"), a)
|
|
write(os.path.join(tmp, "target.txt"), b)
|
|
d = gnu_diff(tmp, "base.txt", "target.txt")
|
|
return diffkit.apply_one(d, a if src is None else src, **kw)
|
|
|
|
check("hand:simple replace", apply_diff("a\nb\nc\n", "a\nB\nc\n") == "a\nB\nc\n")
|
|
check("hand:insert + delete mix",
|
|
apply_diff("1\n2\n3\n4\n5\n", "1\nX\n3\n5\n6\n") == "1\nX\n3\n5\n6\n")
|
|
check("hand:no newline at EOF",
|
|
apply_diff("a\nb\n", "a\nB") == "a\nB")
|
|
check("hand:newline added at EOF",
|
|
apply_diff("a\nb", "a\nb\n") == "a\nb\n")
|
|
check("hand:empty context lines",
|
|
apply_diff("\n\nx\n\n", "\ny\n\n") == "\ny\n\n")
|
|
check("hand:CRLF preserved",
|
|
apply_diff("a\r\nb\r\nc\r\n", "a\r\nB\r\nc\r\n") == "a\r\nB\r\nc\r\n")
|
|
|
|
# insertion into an empty file -> hunk header @@ -0,0 +1,N @@
|
|
write(os.path.join(tmp, "empty.txt"), "")
|
|
write(os.path.join(tmp, "target.txt"), "x\ny\n")
|
|
d = gnu_diff(tmp, "empty.txt", "target.txt")
|
|
check("hand:insert into empty file", diffkit.apply_one(d, "") == "x\ny\n")
|
|
|
|
# creation from /dev/null
|
|
write(os.path.join(tmp, "target.txt"), "new1\nnew2\n")
|
|
_, out_txt, _ = run_bytes(["diff", "-u", "/dev/null", "target.txt"], tmp)
|
|
out = diffkit.apply_patch(diffkit.parse_unified_diff(out_txt), {})
|
|
check("hand:create from /dev/null", out == {"target.txt": "new1\nnew2\n"},
|
|
repr(out))
|
|
|
|
# deletion to /dev/null
|
|
write(os.path.join(tmp, "base.txt"), "gone1\ngone2\n")
|
|
_, out_txt, _ = run_bytes(["diff", "-u", "base.txt", "/dev/null"], tmp)
|
|
out = diffkit.apply_patch(diffkit.parse_unified_diff(out_txt),
|
|
{"base.txt": "gone1\ngone2\n"})
|
|
check("hand:delete to /dev/null", out == {"base.txt": None}, repr(out))
|
|
|
|
# malformed hunk: header counts disagree with body
|
|
bad = "--- a.txt\n+++ a.txt\n@@ -1,3 +1,3 @@\n a\n-b\n+c\n"
|
|
try:
|
|
diffkit.apply_one(bad, "a\nb\n")
|
|
check("hand:malformed hunk rejected", False, "no error raised")
|
|
except diffkit.DiffApplyError as e:
|
|
check("hand:malformed hunk rejected", "disagrees" in str(e), str(e))
|
|
|
|
# offset drift
|
|
base, target = "c1\nc2\nc3\na\nb\nc\nc4\nc5\nc6\n", "c1\nc2\nc3\na\nB\nc\nc4\nc5\nc6\n"
|
|
write(os.path.join(tmp, "base.txt"), base)
|
|
write(os.path.join(tmp, "target.txt"), target)
|
|
d = gnu_diff(tmp, "base.txt", "target.txt")
|
|
shifted = "P0\nP1\n" + base
|
|
got = diffkit.apply_one(d, shifted)
|
|
check("hand:offset drift located", got == "P0\nP1\n" + target, repr(got[:40]))
|
|
try:
|
|
diffkit.apply_one(d, shifted, max_offset=0)
|
|
check("hand:max_offset=0 rejects drift", False, "no error raised")
|
|
except diffkit.DiffApplyError:
|
|
check("hand:max_offset=0 rejects drift", True)
|
|
|
|
# fuzz: corrupt the first leading context line of the diff
|
|
lines = d.split("\n")
|
|
for i, ln in enumerate(lines):
|
|
if ln.startswith(" "):
|
|
lines[i] = " " + "c1-MUTATED"
|
|
break
|
|
fuzzed = "\n".join(lines)
|
|
corrupt_src = "c1-MUTATED\nc2\nc3\na\nb\nc\nc4\nc5\nc6\n"
|
|
expected = "c1-MUTATED\nc2\nc3\na\nB\nc\nc4\nc5\nc6\n"
|
|
try:
|
|
diffkit.apply_one(fuzzed, corrupt_src, max_fuzz=0)
|
|
check("hand:fuzz=0 rejects mismatch", False, "no error raised")
|
|
except diffkit.DiffApplyError:
|
|
check("hand:fuzz=0 rejects mismatch", True)
|
|
check("hand:fuzz=1 recovers", diffkit.apply_one(fuzzed, corrupt_src, max_fuzz=1) == expected)
|
|
|
|
# round trip: forward then reverse
|
|
write(os.path.join(tmp, "base.txt"), base)
|
|
write(os.path.join(tmp, "target.txt"), target)
|
|
fwd = gnu_diff(tmp, "base.txt", "target.txt")
|
|
rev = gnu_diff(tmp, "target.txt", "base.txt")
|
|
check("hand:round trip forward+reverse",
|
|
diffkit.apply_one(rev, diffkit.apply_one(fwd, base)) == base)
|
|
|
|
# multi-file in one text
|
|
write(os.path.join(tmp, "f1.txt"), "one\ntwo\nthree\n")
|
|
write(os.path.join(tmp, "f1new.txt"), "one\nTWO\nthree\n")
|
|
write(os.path.join(tmp, "f2.txt"), "red\ngreen\nblue\n")
|
|
write(os.path.join(tmp, "f2new.txt"), "red\nblue\n")
|
|
d1 = run_bytes(["diff", "-u", "f1.txt", "f1new.txt"], tmp)[1]
|
|
d2 = run_bytes(["diff", "-u", "f2.txt", "f2new.txt"], tmp)[1]
|
|
out = diffkit.apply_patch(diffkit.parse_unified_diff(d1 + d2),
|
|
{"f1.txt": "one\ntwo\nthree\n", "f2.txt": "red\ngreen\nblue\n"})
|
|
check("hand:multi-file patch",
|
|
out == {"f1new.txt": "one\nTWO\nthree\n", "f2new.txt": "red\nblue\n"}, repr(out))
|
|
|
|
# path stripping
|
|
p = diffkit.parse_unified_diff("--- a/x/y.txt\n+++ b/x/y.txt\n@@ -1 +1 @@\n-a\n+b\n")
|
|
check("hand:-p1 path strip",
|
|
diffkit.apply_patch(p, {"x/y.txt": "a\n"}, strip=1) == {"x/y.txt": "b\n"})
|
|
check("hand:timestamps stripped",
|
|
p[0].old_path == "a/x/y.txt" and p[0].new_path == "b/x/y.txt")
|
|
|
|
# split_lines must not split on \v \f \x85 (str.splitlines would)
|
|
weird = "a\x0bb\x0cc\n"
|
|
check("hand:split_lines only splits on \\n",
|
|
diffkit.split_lines(weird) == ["a\x0bb\x0cc\n"], repr(diffkit.split_lines(weird)))
|
|
|
|
|
|
def cli_case(tmp: str) -> None:
|
|
"""End-to-end: the CLI must patch real files on disk, for -p0 and git-style -p1."""
|
|
gen = os.path.join(tmp, "cli-gen") # real a/ and b/ trees, to generate the diff
|
|
flat = os.path.join(tmp, "cli-flat") # single working file, git-style prefixes
|
|
for d in (gen, flat):
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
os.makedirs(os.path.join(gen, "a"))
|
|
os.makedirs(os.path.join(gen, "b"))
|
|
os.makedirs(flat)
|
|
|
|
src = "alpha\nbeta\ngamma\n"
|
|
dst = "alpha\nBETA\ngamma\ndelta\n"
|
|
write(os.path.join(gen, "a", "f.txt"), src)
|
|
write(os.path.join(gen, "b", "f.txt"), dst)
|
|
_, d, _ = run_bytes(["diff", "-u", "a/f.txt", "b/f.txt"], gen)
|
|
write(os.path.join(flat, "d.diff"), d)
|
|
write(os.path.join(gen, "d.diff"), d)
|
|
|
|
# -p0: paths as written (a/f.txt -> b/f.txt)
|
|
rc, out, err = run_bytes([sys.executable, os.path.join(HERE, "diffkit.py"), "-p0", "d.diff"], gen)
|
|
got = open(os.path.join(gen, "b", "f.txt"), encoding="utf-8", newline="").read()
|
|
check("cli:-p0 patches a file on disk", rc == 0 and got == dst,
|
|
"rc=%s out=%r err=%r got=%r" % (rc, out, err, got))
|
|
|
|
# -p1: git-style virtual prefixes, patching f.txt in place
|
|
write(os.path.join(flat, "f.txt"), src)
|
|
rc, out, err = run_bytes([sys.executable, os.path.join(HERE, "diffkit.py"), "-p1", "d.diff"], flat)
|
|
got = open(os.path.join(flat, "f.txt"), encoding="utf-8", newline="").read()
|
|
check("cli:-p1 patches a file on disk in place", rc == 0 and got == dst,
|
|
"rc=%s out=%r err=%r got=%r" % (rc, out, err, got))
|
|
|
|
# a hunk that cannot apply must exit nonzero and leave the file untouched
|
|
write(os.path.join(gen, "b", "f.txt"), src) # reset target to its pre-patch state
|
|
write(os.path.join(gen, "a", "f.txt"), "totally\ndifferent\ncontent\n")
|
|
rc2, _, err2 = run_bytes([sys.executable, os.path.join(HERE, "diffkit.py"), "-p0", "d.diff"], gen)
|
|
after = open(os.path.join(gen, "b", "f.txt"), encoding="utf-8", newline="").read()
|
|
check("cli:unplaceable hunk exits nonzero, target intact",
|
|
rc2 != 0 and after == src, "rc=%s err=%r after=%r" % (rc2, err2, after))
|
|
|
|
# a missing patch file must be a clean error, not a traceback
|
|
rc3, _, err3 = run_bytes([sys.executable, os.path.join(HERE, "diffkit.py"), "-p0", "nope.diff"], flat)
|
|
check("cli:missing patch file is a clean error",
|
|
rc3 == 1 and "Traceback" not in err3 and "cannot read patch file" in err3,
|
|
"rc=%s err=%r" % (rc3, err3))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 2. property suite
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def property_suite(apply_fn, tmp: str, n_cases: int, drift: bool, seed0: int
|
|
) -> tuple[list[str], int, int, int]:
|
|
"""Return (failures, patch_agreements, patch_skips, cases_actually_run)."""
|
|
fails: list[str] = []
|
|
agree = skipped = ran = 0
|
|
for k in range(n_cases):
|
|
rng = random.Random(seed0 + k)
|
|
base = rand_text(rng)
|
|
target = mutate(rng, base)
|
|
if base == target:
|
|
continue
|
|
write(os.path.join(tmp, "base.txt"), base)
|
|
write(os.path.join(tmp, "target.txt"), target)
|
|
d = gnu_diff(tmp, "base.txt", "target.txt")
|
|
if not d.strip():
|
|
continue
|
|
|
|
pre = rand_prefix(rng) if drift else ""
|
|
src, expected = pre + base, pre + target
|
|
|
|
if drift:
|
|
hunks = diffkit.parse_unified_diff(d)[0].hunks
|
|
if any(h.old_count == 0 for h in hunks):
|
|
continue # pure insertion: no content anchor, drift undefined
|
|
|
|
ran += 1
|
|
try:
|
|
got = apply_fn(d, src)
|
|
except Exception as e: # noqa: BLE001
|
|
fails.append("case %d raised %s: %s" % (k, type(e).__name__, e))
|
|
continue
|
|
if got != expected:
|
|
fails.append("case %d mismatch (drift=%s)" % (k, drift))
|
|
|
|
if pre == "": # GNU patch cross-check only where the diff is exact
|
|
pt = gnu_patch_apply(tmp, d, src)
|
|
if pt is None:
|
|
skipped += 1
|
|
elif pt == expected:
|
|
agree += 1
|
|
else:
|
|
fails.append("case %d: GNU patch disagrees with the true target" % k)
|
|
return fails, agree, skipped, ran
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 3. mutants (the suite must kill them)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def mutant_no_search(diff_text: str, base_text: str) -> str:
|
|
"""Naive applier: trust the hunk header position, no drift, no fuzz."""
|
|
p = diffkit.parse_unified_diff(diff_text)[0]
|
|
return diffkit.apply_file_patch(p, base_text, max_fuzz=0, max_offset=0)
|
|
|
|
|
|
def mutant_first_hunk_only(diff_text: str, base_text: str) -> str:
|
|
p = diffkit.parse_unified_diff(diff_text)[0]
|
|
p.hunks = p.hunks[:1]
|
|
return diffkit.apply_file_patch(p, base_text)
|
|
|
|
|
|
def mutant_swaps_sides(diff_text: str, base_text: str) -> str:
|
|
p = diffkit.parse_unified_diff(diff_text)[0]
|
|
for h in p.hunks:
|
|
h.lines = [("+" if k == "-" else "-" if k == "+" else k, c) for k, c in h.lines]
|
|
h.old_count, h.new_count = h.new_count, h.old_count
|
|
return diffkit.apply_file_patch(p, base_text)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def main() -> int:
|
|
tmp = tempfile.mkdtemp(prefix="diffkit-selftest-")
|
|
try:
|
|
hand_cases(tmp)
|
|
cli_case(tmp)
|
|
|
|
real = lambda d, s: diffkit.apply_one(d, s) # noqa: E731
|
|
|
|
fails, agree, skipped, ran = property_suite(real, tmp, PLAIN_CASES, False, 1000)
|
|
check("property: %d randomised exact diffs apply (generated %d)" % (ran, PLAIN_CASES),
|
|
not fails, "; ".join(fails[:3]))
|
|
check("property: GNU patch oracle agrees (%d/%d, %d skipped)"
|
|
% (agree, agree + skipped, skipped), agree > 0 and skipped == 0)
|
|
|
|
dfails, dagree, _, dran = property_suite(real, tmp, DRIFT_CASES, True, 9000)
|
|
check("property: %d randomised offset-drift diffs apply (generated %d)"
|
|
% (dran, DRIFT_CASES), not dfails, "; ".join(dfails[:3]))
|
|
|
|
# determinism
|
|
again = property_suite(real, tmp, 20, False, 1000)[0]
|
|
check("property: suite is deterministic", not again)
|
|
|
|
# mutation control
|
|
m_fails = property_suite(mutant_no_search, tmp, 30, True, 9000)[0]
|
|
check("mutation: no-search applier is killed (%d failures)" % len(m_fails),
|
|
len(m_fails) > 0)
|
|
|
|
m2_fails = property_suite(mutant_first_hunk_only, tmp, PLAIN_CASES, False, 1000)[0]
|
|
check("mutation: first-hunk-only applier is killed (%d failures)" % len(m2_fails),
|
|
len(m2_fails) > 0)
|
|
|
|
m3_fails = property_suite(mutant_swaps_sides, tmp, PLAIN_CASES, False, 1000)[0]
|
|
check("mutation: sides-swapped applier is killed (%d failures)" % len(m3_fails),
|
|
len(m3_fails) > 0)
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
passed = sum(1 for _, ok, _ in _results if ok)
|
|
total = len(_results)
|
|
print("\n%d/%d checks passed" % (passed, total))
|
|
if passed != total:
|
|
print("FAILURES:")
|
|
for name, ok, detail in _results:
|
|
if not ok:
|
|
print(" - %s :: %s" % (name, detail))
|
|
return 1
|
|
print("ALL CHECKS PASSED")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|