forge: append_only_audit DRY collapse — extract _subseq_match_count from duplicated subsequence scan in line_test + is_line_subsequence
This commit is contained in:
forge committed
1 parent
59f3ce2a5c
commit
1468811853
2 files changed
+335
-9
No files matched your search
@@ -167,10 +167,7 @@ def line_test(parent_bytes, child_bytes):
|
||||
return "IDENTICAL", None, "", 0, 0
|
||||
missing = sum((Counter(pl) - Counter(cl)).values())
|
||||
added = sum((Counter(cl) - Counter(pl)).values())
|
||||
i = 0
|
||||
for ln in cl:
|
||||
if i < len(pl) and ln == pl[i]:
|
||||
i += 1
|
||||
i = _subseq_match_count(pl, cl)
|
||||
if i == len(pl):
|
||||
return "SUBSEQUENCE", None, "", 0, added
|
||||
return "LOSS", i + 1, pl[i].decode("utf-8", "replace")[:120], missing, added
|
||||
@@ -187,14 +184,26 @@ def escape_unescape(b):
|
||||
return ESCAPE_RE.sub(lambda m: chr(int(m.group(1), 16)).encode("utf-8"), b)
|
||||
|
||||
|
||||
def _subseq_match_count(parent_lines, child_lines):
|
||||
"""Greedy count of parent_lines that match, in order, within child_lines.
|
||||
|
||||
Returns the number of parent_lines consumed by the forward scan:
|
||||
len(parent_lines) means the parent is a subsequence of the child; a
|
||||
smaller value means the scan broke after that many matches, and
|
||||
parent_lines[i] is the first parent line that did not match.
|
||||
"""
|
||||
i = 0
|
||||
for ln in child_lines:
|
||||
if i < len(parent_lines) and ln == parent_lines[i]:
|
||||
i += 1
|
||||
return i
|
||||
|
||||
|
||||
def is_line_subsequence(a_bytes, b_bytes):
|
||||
"""True iff a_bytes's lines are an order-preserving subsequence of b_bytes's."""
|
||||
al = a_bytes.split(b"\n")
|
||||
bl = b_bytes.split(b"\n")
|
||||
i = 0
|
||||
for ln in bl:
|
||||
if i < len(al) and ln == al[i]:
|
||||
i += 1
|
||||
return i == len(al)
|
||||
return _subseq_match_count(al, bl) == len(al)
|
||||
|
||||
|
||||
def loss_kind(parent_bytes, child_bytes):
|
||||
|
||||
@@ -0,0 +1,317 @@
|
||||
#!/usr/bin/env python3
|
||||
"""verify_append_only_audit_dry_refactor.py
|
||||
|
||||
Equivalence proof for the DRY collapse in tools/append_only_audit.py
|
||||
(forge lane: Refactor).
|
||||
|
||||
THE SMELL
|
||||
The greedy subsequence-walk loop was hand-duplicated, verbatim, in two
|
||||
functions that do the same scan with different variable names:
|
||||
|
||||
line_test() -- inlined in the audit engine:
|
||||
i = 0
|
||||
for ln in cl:
|
||||
if i < len(pl) and ln == pl[i]:
|
||||
^-- the inlined loop (4 lines)
|
||||
|
||||
is_line_subsequence() -- the standalone predicate used by loss_kind():
|
||||
i = 0
|
||||
for ln in bl:
|
||||
if i < len(al) and ln == al[i]:
|
||||
^-- the SAME loop, verbatim, only variable names differ
|
||||
return i == len(al)
|
||||
|
||||
Two copies of one 4-line statement is the canonical DRY smell; each edit to
|
||||
the scan logic would have to touch both, and a third call site would be a
|
||||
third freehand copy.
|
||||
|
||||
THE REFACTOR
|
||||
One module-level helper replaces both inlined loops:
|
||||
|
||||
def _subseq_match_count(parent_lines, child_lines):
|
||||
... # the single greedy scan
|
||||
return i # number of parent_lines consumed
|
||||
|
||||
line_test(): i = _subseq_match_count(pl, cl)
|
||||
is_line_subsequence(): return _subseq_match_count(al, bl) == len(al)
|
||||
|
||||
WHAT THIS VERIFIER PROVES (post-commit: HEAD~1 = original, HEAD = refactored)
|
||||
|
||||
[1] selftest: ORIGINAL (HEAD~1) 18/18 rc 0
|
||||
[2] selftest: REFACTORED (HEAD) 18/18 rc 0
|
||||
[3] selftest: stdout BYTE-IDENTICAL (original == refactored)
|
||||
[4] code-level: refactored HAS _subseq_match_count; original has NONE
|
||||
[5] code-level: refactored inlined-loops = 0; original inlined-loops >= 1
|
||||
[6] non-vacuity: live selftest produced a real 18-cell census
|
||||
[7] line_test direct equivalence on 8 edge-case inputs
|
||||
[8] is_line_subsequence direct equivalence on 6 inputs
|
||||
[9] loss_kind direct equivalence on 4 inputs (calls is_line_subsequence)
|
||||
[10] report() on a real synthetic audit: BYTE-IDENTICAL stdout
|
||||
[11] verifier itself committed to HEAD
|
||||
|
||||
Exit 0 iff all checks pass, else 1.
|
||||
"""
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
REPO = "/root/troupe-standup"
|
||||
TOOL = "tools/append_only_audit.py"
|
||||
VERIFIER_NAME = "verify_append_only_audit_dry_refactor.py"
|
||||
VERIFIER_PATH = "wizard-work/forge/" + VERIFIER_NAME
|
||||
SUMMARY = "selftest: 18 cells, 18 passed, 0 failed"
|
||||
HELPER = "def _subseq_match_count("
|
||||
INLINED_CL = "for ln in cl:"
|
||||
INLINED_BL = "for ln in bl:"
|
||||
TMPDIR = os.environ.get("TMPDIR", tempfile.gettempdir())
|
||||
|
||||
|
||||
class Mismatch(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def git_show(rev, path):
|
||||
"""Return the text of `path` at git revision `rev` in REPO."""
|
||||
proc = subprocess.run(["git", "-C", REPO, "show", f"{rev}:{path}"],
|
||||
capture_output=True, text=True, timeout=60)
|
||||
if proc.returncode != 0:
|
||||
raise Mismatch(f"cannot read {rev}:{path}: {proc.stderr.strip()}")
|
||||
return proc.stdout
|
||||
|
||||
|
||||
def materialize_original():
|
||||
"""Write the pre-refactor source (HEAD~1) to a temp file; return path."""
|
||||
src = git_show("HEAD~1", TOOL)
|
||||
d = tempfile.mkdtemp(prefix="aoa_orig_")
|
||||
p = os.path.join(d, "append_only_audit.py")
|
||||
with open(p, "w", encoding="utf-8") as fh:
|
||||
fh.write(src)
|
||||
os.chmod(p, 0o755)
|
||||
return p
|
||||
|
||||
|
||||
def load_module(path, name):
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
def run_script(source_text, args):
|
||||
"""Write source_text to an isolated temp file and run with args."""
|
||||
fd, path = tempfile.mkstemp(suffix=".py", dir=TMPDIR)
|
||||
try:
|
||||
with os.fdopen(fd, "w") as fh:
|
||||
fh.write(source_text)
|
||||
proc = subprocess.run([sys.executable, path] + list(args),
|
||||
capture_output=True, text=True, timeout=120)
|
||||
return proc.returncode, proc.stdout, proc.stderr
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
def count_inlined_loops(text):
|
||||
"""Count bare `for ln in cl:` / `for ln in bl:` loop headers in code (not comments)."""
|
||||
code_lines = [ln for ln in text.splitlines()
|
||||
if not ln.lstrip().startswith("#")]
|
||||
return sum(1 for ln in code_lines if ln.strip() in (INLINED_CL, INLINED_BL))
|
||||
|
||||
|
||||
def sha(text):
|
||||
return hashlib.sha256(text.encode()).hexdigest()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Direct equivalence cases: (parent_bytes, child_bytes, description)
|
||||
# ---------------------------------------------------------------------------
|
||||
EDGE_CASES_LINE_TEST = [
|
||||
(b"", b"hello\n", "NO_PARENT (empty parent)"),
|
||||
(b"same\n", b"same\n", "IDENTICAL"),
|
||||
(b"a\nb\n", b"a\nb\nc\n", "APPEND (byte prefix)"),
|
||||
(b"a\nb\n", b"a\nX\nb\nc\n", "SUBSEQUENCE (mid insertion)"),
|
||||
(b"a\nb\nc\n", b"a\nb\n", "LOSS (parent longer)"),
|
||||
(b"\n\n\n", b"x\ny\nz\n", "LOSS (no match)"),
|
||||
(b"dash \\u2014 here\nplain\n", b"dash \u2014 here\nplain\n",
|
||||
"ESCAPE_NORMALISATION path"),
|
||||
(b"alpha\nbeta\n", b"ALPHA\nbeta\n", "LOSS (case change)"),
|
||||
]
|
||||
|
||||
EDGE_CASES_SUBSEQ = [
|
||||
(b"a\nb\n", b"a\nb\nc\n", True, "a is subsequence of abc"),
|
||||
(b"a\nb\n", b"a\nX\nb\nc\n", True, "a is subsequence of aXb"),
|
||||
(b"a\nb\nc\n", b"a\nb\n", False, "abc is not subsequence of ab"),
|
||||
(b"alpha\n", b"ALPHA\n", False, "alpha != ALPHA"),
|
||||
(b"", b"anything\n", True, "empty parent is trivially subsequence"),
|
||||
(b"x\ny\n", b"y\nx\n", False, "NOT order-preserving"),
|
||||
]
|
||||
|
||||
EDGE_CASES_LOSS_KIND = [
|
||||
(b"dash \\u2014 here\nplain\n", b"dash \u2014 here\nplain\n",
|
||||
"ESCAPE_NORMALISATION", "escape rewrite"),
|
||||
(b"alpha\nbeta\n", b"ALPHA\nbeta\n",
|
||||
"REWRITE", "case change"),
|
||||
(b"a\nb\nc\n", b"a\nb\nx\n",
|
||||
"REWRITE", "last line changed"),
|
||||
(b"foo\nbar\n", b"FOO\nbaz\n",
|
||||
"REWRITE", "both lines changed"),
|
||||
]
|
||||
|
||||
|
||||
def make_fixture_repo(tmp):
|
||||
"""A synthetic git repo with append / mid-insertion / loss history."""
|
||||
repo = os.path.join(tmp, "repo")
|
||||
os.makedirs(repo)
|
||||
env = dict(os.environ, GIT_AUTHOR_NAME="t", GIT_AUTHOR_EMAIL="t@e",
|
||||
GIT_COMMITTER_NAME="t", GIT_COMMITTER_EMAIL="t@e")
|
||||
subprocess.run(["git", "init", "-q", "-b", "main", repo], check=True, env=env,
|
||||
capture_output=True)
|
||||
|
||||
def w(path, data):
|
||||
p = os.path.join(repo, path)
|
||||
os.makedirs(os.path.dirname(p), exist_ok=True)
|
||||
with open(p, "w") as fh:
|
||||
fh.write(data)
|
||||
|
||||
def g(*a):
|
||||
subprocess.run(["git", "-C", repo] + list(a), check=True, env=env,
|
||||
capture_output=True)
|
||||
|
||||
w("a.txt", "one\ntwo\n")
|
||||
w("mid.txt", "head\nbody\ntail\n")
|
||||
g("add", "-A")
|
||||
g("commit", "-qm", "c1")
|
||||
|
||||
p = os.path.join(repo, "a.txt")
|
||||
with open(p, "a") as fh:
|
||||
fh.write("three\n")
|
||||
g("add", "-A"); g("commit", "-qm", "c2 append")
|
||||
|
||||
p = os.path.join(repo, "mid.txt")
|
||||
with open(p, "w") as fh:
|
||||
fh.write("head\nINSERTED\nbody\ntail\n")
|
||||
g("add", "-A"); g("commit", "-qm", "c3 mid insertion")
|
||||
|
||||
p = os.path.join(repo, "mid.txt")
|
||||
with open(p, "w") as fh:
|
||||
fh.write("head\nINSERTED\ntail\n")
|
||||
g("add", "-A"); g("commit", "-qm", "c4 loss")
|
||||
return repo
|
||||
|
||||
|
||||
def main():
|
||||
ok = True
|
||||
cells, failed = [], []
|
||||
|
||||
def check(label, passed, detail=""):
|
||||
cells.append(label)
|
||||
status = "ok " if passed else "FAIL"
|
||||
print(f" [{status}] {label}" + (f" [{detail}]" if detail and not passed else ""))
|
||||
if not passed:
|
||||
failed.append(f"{label}: {detail}" if detail else label)
|
||||
nonlocal ok
|
||||
ok = False
|
||||
|
||||
orig_path = materialize_original()
|
||||
try:
|
||||
orig_src = git_show("HEAD~1", TOOL)
|
||||
with open(os.path.join(REPO, TOOL), encoding="utf-8") as fh:
|
||||
ref_src = fh.read()
|
||||
|
||||
mod_o = load_module(orig_path, "aoa_orig")
|
||||
mod_r = load_module(os.path.join(REPO, TOOL), "aoa_ref")
|
||||
|
||||
# ---- [1-3] selftest equivalence ---------------------------------------
|
||||
rc_o, out_o, _ = run_script(orig_src, ["--selftest"])
|
||||
rc_r, out_r, _ = run_script(ref_src, ["--selftest"])
|
||||
check("[1] selftest: ORIGINAL (HEAD~1) 18/18 rc 0",
|
||||
rc_o == 0 and SUMMARY in out_o, f"rc={rc_o} {out_o[-120:]}")
|
||||
check("[2] selftest: REFACTORED (HEAD) 18/18 rc 0",
|
||||
rc_r == 0 and SUMMARY in out_r, f"rc={rc_r} {out_r[-120:]}")
|
||||
check("[3] selftest: stdout BYTE-IDENTICAL (original == refactored)",
|
||||
out_o == out_r, f"len_o={len(out_o)} len_r={len(out_r)}")
|
||||
|
||||
# ---- [4-5] code-level: smell existed and was collapsed ----------------
|
||||
check("[4] refactored HAS _subseq_match_count; original NONE",
|
||||
HELPER in ref_src and HELPER not in orig_src,
|
||||
f"ref={HELPER in ref_src} orig={HELPER in orig_src}")
|
||||
check("[5] refactored inlined-loops = 0; original >= 1",
|
||||
count_inlined_loops(ref_src) == 0 and count_inlined_loops(orig_src) >= 1,
|
||||
f"ref={count_inlined_loops(ref_src)} orig={count_inlined_loops(orig_src)}")
|
||||
|
||||
# ---- [6] non-vacuity --------------------------------------------------
|
||||
check("[6] non-vacuity: live selftest produced a real 18-cell census",
|
||||
"c01" in out_r and "PASS" in out_r and out_r.count("PASS") >= 17)
|
||||
|
||||
# ---- [7-9] direct function equivalence --------------------------------
|
||||
mismatches = []
|
||||
for pb, cb, desc in EDGE_CASES_LINE_TEST:
|
||||
o = mod_o.line_test(pb, cb)
|
||||
r = mod_r.line_test(pb, cb)
|
||||
if o != r:
|
||||
mismatches.append(f"line_test {desc}: {o!r} != {r!r}")
|
||||
check("[7] line_test direct equivalence (8 edge cases)",
|
||||
not mismatches, "; ".join(mismatches)[:300])
|
||||
|
||||
subseq_mismatches = []
|
||||
for pb, cb, want, desc in EDGE_CASES_SUBSEQ:
|
||||
o = mod_o.is_line_subsequence(pb, cb)
|
||||
r = mod_r.is_line_subsequence(pb, cb)
|
||||
if o != r or o != want:
|
||||
subseq_mismatches.append(f"{desc}: orig={o} ref={r} want={want}")
|
||||
check("[8] is_line_subsequence direct equivalence (6 inputs)",
|
||||
not subseq_mismatches, "; ".join(subseq_mismatches)[:300])
|
||||
|
||||
lk_mismatches = []
|
||||
for pb, cb, want, desc in EDGE_CASES_LOSS_KIND:
|
||||
o = mod_o.loss_kind(pb, cb)
|
||||
r = mod_r.loss_kind(pb, cb)
|
||||
if o != r or o != want:
|
||||
lk_mismatches.append(f"{desc}: orig={o} ref={r} want={want}")
|
||||
check("[9] loss_kind direct equivalence (4 inputs, via is_line_subsequence)",
|
||||
not lk_mismatches, "; ".join(lk_mismatches)[:300])
|
||||
|
||||
# ---- [10] report() byte-identical on a real synthetic audit -----------
|
||||
with tempfile.TemporaryDirectory(prefix="aoa_report_") as td:
|
||||
repo = make_fixture_repo(td)
|
||||
res_o = mod_o.audit(repo)
|
||||
res_r = mod_r.audit(repo)
|
||||
check("[10a] audit() rows match structurally",
|
||||
len(res_o["rows"]) == len(res_r["rows"])
|
||||
and all(r_o == r_r for r_o, r_r in zip(res_o["rows"], res_r["rows"])),
|
||||
f"n_o={len(res_o['rows'])} n_r={len(res_r['rows'])}")
|
||||
buf_o, buf_r = io.StringIO(), io.StringIO()
|
||||
mod_o.report(res_o, out=buf_o)
|
||||
mod_r.report(res_r, out=buf_r)
|
||||
check("[10b] report() stdout BYTE-IDENTICAL",
|
||||
buf_o.getvalue() == buf_r.getvalue(),
|
||||
f"sha_o={sha(buf_o.getvalue()[:12])} sha_r={sha(buf_r.getvalue()[:12])}")
|
||||
|
||||
# ---- [11] verifier committed to HEAD ----------------------------------
|
||||
rc_gc = subprocess.run(
|
||||
["git", "-C", REPO, "cat-file", "-e", f"HEAD:{VERIFIER_PATH}"],
|
||||
capture_output=True).returncode
|
||||
check("[11] verifier committed to HEAD", rc_gc == 0, f"cat-file rc={rc_gc}")
|
||||
|
||||
finally:
|
||||
shutil.rmtree(os.path.dirname(orig_path), ignore_errors=True)
|
||||
|
||||
print(f"\nverify: {len(cells)} checks, {len(cells) - len(failed)} passed, {len(failed)} failed")
|
||||
for f in failed:
|
||||
print(f" FAIL {f}")
|
||||
verdict = ("ALL GREEN -- append_only_audit DRY collapse behavior preserved"
|
||||
if ok else "MISMATCH -- behavior NOT preserved")
|
||||
print(f"\nVERDICT: {verdict}")
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except Mismatch as e:
|
||||
print(f"\nVERDICT: MISMATCH -- {e}")
|
||||
sys.exit(1)
|
||||
Reference in new issue
Block a user