forge: refactor _scan_dry DRY scanner - eliminate import-time side effect, redundant or, dead sys import, resource leak; 11-check verifier, byte-identical output
This commit is contained in:
1 parent
6aaa1266a1
commit
d7e6e8e50c
5 files changed
+679
No files matched your search
@@ -0,0 +1,110 @@
|
||||
# _scan_dry refactor verification (forge lane: Refactor)
|
||||
|
||||
## The shipped artifact
|
||||
|
||||
Three files under `wizard-work/forge/`:
|
||||
|
||||
- `_scan_dry.py` -- original probe: a DRY smell-scanner that itself runs its full scan
|
||||
(and prints a 4811-byte report) at module top level.
|
||||
- `_scan_dry_refactored.py` -- refactored probe with smells removed.
|
||||
- `verify_scan_dry_refactor.py` -- equivalence + smell-absence + non-vacuity harness
|
||||
(11 checks).
|
||||
|
||||
Evidence files: `_scan_dry_out.txt` (original stdout+stderr, cksum
|
||||
`2478253836 4811`), `_scan_dry_verifier_out.txt` (full verifier run).
|
||||
|
||||
## The smell (what the refactor fixes)
|
||||
|
||||
The probe *is* a DRY scanner, so this is dogfooding -- the fixes are the smells a
|
||||
detector should flag on its own author:
|
||||
|
||||
1. **Import-time side effect.** The `for f in sorted(os.listdir(TOOLS))` loop and
|
||||
every `print(...)` run at module top level. Importing the module (for a test,
|
||||
harness, or monkeypatch) executes the whole scan and emits ~4.8 KB of output
|
||||
-- a side effect masquerading as init.
|
||||
2. **Redundant `or` clause.** `"len([" in l or "len([" in l` -- the second
|
||||
operand is identical to the first, so `or` adds no match; dead logic that
|
||||
only reads as "len([...]) vs len([...]) patterns" confusion.
|
||||
3. **Unused `sys` import.** `import ast, os, hashlib, collections, sys` -- `sys`
|
||||
is never referenced anywhere in the body.
|
||||
4. **Resource leak.** `src = open(path, encoding="utf-8").read()` -- file
|
||||
handle never closed (no `with` block).
|
||||
5. **No `__main__` guard.** The scan runs on import.
|
||||
|
||||
## The refactor
|
||||
|
||||
`_scan_dry_refactored.py`:
|
||||
|
||||
- Moved the scan loop into `main()` and guarded it with
|
||||
`if __name__ == "__main__": main()` -- import is now side-effect-free.
|
||||
- Collapsed `"len([" in l or "len([" in l` to `"len([" in l` -- identical count,
|
||||
zero behavior change.
|
||||
- Dropped the unused `sys` import.
|
||||
- Replaced the bare `open()` with a `with open(...) as fh:` context manager.
|
||||
- Extracted two pure, testable helpers:
|
||||
- `scan_file(path)` -- returns `(info, dup_funcs, repeated_lines, nlines, funcs)`
|
||||
(Counter of hits + duplicate-function detection + repeated-line report).
|
||||
- `format_report(...)` -- returns the per-file report as a list of strings
|
||||
(pure: no I/O), so the formatting is unit-testable without running the probe.
|
||||
- Replaced magic literals (`3`, `15`, `12`) with named constants
|
||||
`REPEATED_MIN`, `REPEATED_MIN_LEN`, `BODY_HASH_LEN`.
|
||||
|
||||
Output is byte-identical to the original: 4811 bytes, cksum
|
||||
`2478253836`.
|
||||
|
||||
## Verifier bug found and fixed in this pass
|
||||
|
||||
The first non-vacuity guard (`check_original_still_smells`) had inverted logic on
|
||||
the `__main__` leg: it appended a "problem" when the original *lacked* a main
|
||||
guard -- but lacking the guard *is* the smell we fixed, not a problem. Corrected
|
||||
so the guard passes iff the original still exhibits the removed smells (no guard,
|
||||
doubled `or`, bare `open`, module-level scan loop). Without this, a vacuous
|
||||
"refactor" that deleted nothing would pass -- the guard now fails closed:
|
||||
the original **must** still smell.
|
||||
|
||||
## Evidence (measured, not claimed)
|
||||
|
||||
Byte-identical stdout+stderr (original vs refactored), captured by the verifier's
|
||||
`check_identical_output`:
|
||||
```
|
||||
$ cksum _scan_dry_out.txt
|
||||
2478253836 4811 _scan_dry_out.txt
|
||||
OK: identical output (exit 0, 4811 bytes)
|
||||
```
|
||||
|
||||
Verifier run (11/11):
|
||||
```
|
||||
$ python3 verify_scan_dry_refactor.py; echo "VERIFIER_EXIT=$?"
|
||||
OK: py_compile passes
|
||||
OK: identical output (exit 0, 4811 bytes)
|
||||
OK: __main__ guard present
|
||||
OK: no 'global' statements in code
|
||||
OK: unused 'sys' import removed
|
||||
OK: redundant doubled 'or' clause removed; single 'len([ in l' present
|
||||
OK: bare open() replaced with 'with' context manager
|
||||
OK: scan_file, format_report, main extracted; main() calls scan_file
|
||||
OK: import is side-effect-free (no probe output at import)
|
||||
OK: scan_file real (/root/troupe-standup/tools/check_lint.py: 574 lines, 11 funcs); dup-body detection works
|
||||
OK: original still exhibits the removed smells (refactor is non-vacuous)
|
||||
|
||||
=== Verification: 11/11 checks passed ===
|
||||
VERIFIER_EXIT=0
|
||||
```
|
||||
|
||||
## Metrics (smell counts, before -> after)
|
||||
| smell | original | refactored |
|
||||
|------------------------------------------------|----------|------------|
|
||||
| import-time side effects (module-level scan+print) | yes | none (guarded `main()`) |
|
||||
| `__main__` guard | absent | present |
|
||||
| redundant `or` clause (`"len([" in l or "len([" in l`) | present | removed (single `"len([" in l`) |
|
||||
| unused `sys` import | present | removed |
|
||||
| resource-leaking bare `open()` | present | `with` context manager |
|
||||
| module-level mutable control flow | yes (`for`+`if`+`print` at top level) | none (pure `scan_file`/`format_report` + `main`) |
|
||||
| byte-identical output (cksum) | `2478253836 4811` | `2478253836 4811` |
|
||||
|
||||
## Verdict
|
||||
Behavior-preserving and measurable: py_compile clean, byte-identical output
|
||||
(4811 bytes, identical cksum), all 11 verifier checks pass (exit 0) including the
|
||||
non-vacuity guard that the original still smelled. The scanner's own
|
||||
import-time-side-effect, redundant-`or`, dead-import, and resource-leak smells
|
||||
are eliminated, and the scan logic is decoupled into pure helpers.
|
||||
@@ -0,0 +1,138 @@
|
||||
|
||||
=== append_only_audit.py (498 lines, 20 funcs) ===
|
||||
listcomp with if-filter x1
|
||||
manual +=1 count x8
|
||||
repeated x5: print("", file=out)
|
||||
|
||||
=== census_row_delta.py (510 lines, 14 funcs) ===
|
||||
manual +=1 count x2
|
||||
repeated x3: for k in d["removed"]:
|
||||
repeated x3: for k in d["added"]:
|
||||
|
||||
=== check_lint.py (574 lines, 11 funcs) ===
|
||||
manual +=1 count x12
|
||||
repeated x10: # --------------------------------------------------------------------------
|
||||
|
||||
=== checkin_check.py (949 lines, 24 funcs) ===
|
||||
sum(1 for ...) x3
|
||||
listcomp with if-filter x9
|
||||
manual +=1 count x5
|
||||
repeated x3: for line in text.splitlines():
|
||||
repeated x3: with open(path, "r", encoding="utf-8") as fh:
|
||||
repeated x3: except ValueError:
|
||||
repeated x4: except OSError as exc:
|
||||
|
||||
=== citation_hash_class.py (454 lines, 10 funcs) ===
|
||||
.count() calls x1
|
||||
manual +=1 count x8
|
||||
repeated x3: bad += 0 if ok else 1
|
||||
|
||||
=== doc_census_check.py (1861 lines, 34 funcs) ===
|
||||
sum(1 for ...) x1
|
||||
manual +=1 count x3
|
||||
repeated x4: "print('widgets: 7')\n")
|
||||
repeated x3: json.dump({"rev": "abc1234",
|
||||
repeated x5: run_gate("--tools", "tools/stub_split.py", "--no-run",
|
||||
repeated x3: out = p.stdout.decode("utf-8", "replace")
|
||||
repeated x3: err = p.stderr.decode("utf-8", "replace")
|
||||
|
||||
=== doc_command_check.py (468 lines, 14 funcs) ===
|
||||
listcomp with if-filter x5
|
||||
repeated x3: capture_output=True, text=True)
|
||||
|
||||
=== gate_blame.py (540 lines, 17 funcs) ===
|
||||
listcomp with if-filter x1
|
||||
.count() calls x1
|
||||
repeated x3: here = os.path.dirname(os.path.abspath(__file__))
|
||||
repeated x4: if proc.returncode != 0:
|
||||
repeated x3: fh.write("# empty ledger\n")
|
||||
|
||||
=== landing_check.py (460 lines, 16 funcs) ===
|
||||
listcomp with if-filter x1
|
||||
manual +=1 count x2
|
||||
repeated x4: rows.append(row)
|
||||
repeated x3: rows = build_rows(repo, rev1, "main", forge, ["tools/a.py"])
|
||||
repeated x3: with open(os.path.join(served, "tools/a.py"), "wb") as fh:
|
||||
|
||||
=== path_claim_check.py (892 lines, 24 funcs) ===
|
||||
listcomp with if-filter x4
|
||||
manual +=1 count x6
|
||||
repeated x6: subprocess.run(["git", "-C", repo, "add", "-A"], capture_output=True)
|
||||
repeated x3: subprocess.run(["git", "-C", repo, "-c", "user.email=t@t", "-c", "user.name=t",
|
||||
repeated x5: "rc=%d %s" % (rc, out[-300:]))
|
||||
repeated x4: with open(q, "w") as fh:
|
||||
|
||||
=== queue_append_audit.py (549 lines, 18 funcs) ===
|
||||
listcomp with if-filter x3
|
||||
repeated x3: s, _ = out_text(rc, out, b"")
|
||||
repeated x3: row.update(verdict="UNGRADABLE",
|
||||
repeated x5: _sh(repo, "git", "add", "-A")
|
||||
repeated x3: with open(queue, "a", encoding="utf-8") as fh:
|
||||
|
||||
=== queue_claim_check.py (1079 lines, 21 funcs) ===
|
||||
sum(1 for ...) x2
|
||||
len([...]) x1
|
||||
listcomp with if-filter x11
|
||||
manual +=1 count x3
|
||||
repeated x12: # --------------------------------------------------------------------------
|
||||
repeated x3: s = strip_code(line)
|
||||
repeated x3: for n, l in enumerate(lines, 1):
|
||||
repeated x3: lines = text.splitlines()
|
||||
repeated x3: for r in records:
|
||||
|
||||
=== queue_index.py (900 lines, 17 funcs) ===
|
||||
manual +=1 count x4
|
||||
repeated x3: for i, ln in enumerate(lines, 1):
|
||||
repeated x12: lines.append("")
|
||||
repeated x9: buf = io.StringIO()
|
||||
repeated x9: with contextlib.redirect_stdout(buf):
|
||||
repeated x10: rc = run(argparse.Namespace(repo=repo, queue="WORK-QUEUE.md", index=False,
|
||||
|
||||
=== reply_check.py (881 lines, 22 funcs) ===
|
||||
sum(1 for ...) x1
|
||||
listcomp with if-filter x3
|
||||
.count() calls x1
|
||||
manual +=1 count x2
|
||||
repeated x3: s = line.strip()
|
||||
repeated x5: for row in rows:
|
||||
repeated x3: file=sys.stderr)
|
||||
repeated x3: with contextlib.redirect_stdout(io.StringIO()):
|
||||
repeated x3: buf = io.StringIO()
|
||||
|
||||
=== selftest_sweep.py (1176 lines, 24 funcs) ===
|
||||
sum(1 for ...) x3
|
||||
listcomp with if-filter x12
|
||||
repeated x4: 2, "cannot read repo"),
|
||||
repeated x4: 2, "not a git work tree"),
|
||||
repeated x6: except OSError as exc:
|
||||
repeated x3: results.append(rec)
|
||||
repeated x4: with tempfile.TemporaryDirectory() as td:
|
||||
|
||||
=== served_drift_check.py (522 lines, 22 funcs) ===
|
||||
sum(1 for ...) x2
|
||||
listcomp with if-filter x5
|
||||
manual +=1 count x4
|
||||
repeated x5: rows = run(repo, "HEAD", base)
|
||||
repeated x4: except GradeError as e:
|
||||
|
||||
=== tool_map.py (642 lines, 16 funcs) ===
|
||||
sum(1 for ...) x2
|
||||
listcomp with if-filter x1
|
||||
repeated x3: except OSError as exc:
|
||||
|
||||
=== verify.py (1019 lines, 16 funcs) ===
|
||||
listcomp with if-filter x3
|
||||
manual +=1 count x3
|
||||
repeated x6: capture_output=True, text=True)
|
||||
repeated x3: c = counts(rows)
|
||||
repeated x9: rc, out = _verify("--repo", repo)
|
||||
repeated x9: rc, out = _verify("--repo", repo, "--fast")
|
||||
repeated x5: seen = open(av).read() if os.path.exists(av) else ""
|
||||
|
||||
=== write_probe.py (734 lines, 12 funcs) ===
|
||||
sum(1 for ...) x1
|
||||
len([...]) x3
|
||||
listcomp with if-filter x2
|
||||
repeated x3: "--json", "{target}"]),
|
||||
repeated x3: rows.append(rec)
|
||||
repeated x4: except OSError as exc:
|
||||
@@ -0,0 +1,126 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Scan tools/*.py for DRY smells: repeated sum(1 for ...)/list-comprehension
|
||||
patterns, duplicated function bodies, and other refactor candidates.
|
||||
|
||||
Refactored (forge lane: Refactor) -- the target *is* a DRY scanner, so the
|
||||
fixes are the smells a detector should flag on its own author:
|
||||
|
||||
* Moved the module-level scan loop under ``if __name__ == "__main__"``: the
|
||||
original ran the full scan (and printed its report) at import time, an
|
||||
import-time side effect that punished any reuse of ``scan_file``.
|
||||
* Collapsed the redundant ``"len([" in l or "len([" in l`` clause: the second
|
||||
operand is identical to the first, so ``or`` adds no match -- removing it
|
||||
changes no results (byte-identical output).
|
||||
* Removed the unused ``sys`` import.
|
||||
* Replaced the resource-leaking ``open(path).read()`` with a ``with`` block.
|
||||
* Extracted the pure helpers ``scan_file`` and ``format_report`` so the
|
||||
scanning logic is unit-testable without running the probe.
|
||||
|
||||
Output is byte-identical to the original (verified against baseline).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import collections
|
||||
import hashlib
|
||||
import os
|
||||
|
||||
REPO = "/root/troupe-standup"
|
||||
TOOLS = os.path.join(REPO, "tools")
|
||||
|
||||
# Minimum repetition count for a line to be flagged as "repeated".
|
||||
REPEATED_MIN = 3
|
||||
# Minimum stripped length for a line to be flagged as repeated.
|
||||
REPEATED_MIN_LEN = 15
|
||||
# Hex-char length for function-body duplicate fingerprints.
|
||||
BODY_HASH_LEN = 12
|
||||
|
||||
|
||||
def scan_file(path: str):
|
||||
"""Scan a single .py file. Return
|
||||
|
||||
``(info, dup_funcs, repeated_lines, nlines, funcs)`` where ``info`` is a
|
||||
Counter of smell hits. Pure apart from the file read at ``path``.
|
||||
"""
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
src = fh.read()
|
||||
lines = src.splitlines()
|
||||
info: collections.Counter = collections.Counter()
|
||||
|
||||
# sum(1 for ... patterns - potential DRY
|
||||
info["sum_gen"] = sum(1 for l in lines if "sum(1 for" in l)
|
||||
# len([...]) vs len([...]) patterns -- the original's
|
||||
# "len([" in l or "len([" in l doubled the same substring; one suffices.
|
||||
info["len_listcomp"] = sum(1 for l in lines if "len([" in l)
|
||||
# list comprehensions that filter
|
||||
info["listcomp_filter"] = sum(1 for l in lines if "if r[" in l)
|
||||
# count() calls on lists
|
||||
info["count_method"] = sum(1 for l in lines if ".count(" in l)
|
||||
# manual counting loops
|
||||
info["manual_count"] = sum(1 for l in lines if "+= 1" in l)
|
||||
|
||||
# Find duplicate function bodies using AST.
|
||||
try:
|
||||
tree = ast.parse(src)
|
||||
funcs = [(node.name, node.body) for node in ast.walk(tree)
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))]
|
||||
# hash each function body
|
||||
body_hashes: dict[str, list[str]] = {}
|
||||
for name, body in funcs:
|
||||
h = hashlib.sha256(ast.unparse(body).encode()).hexdigest()[:BODY_HASH_LEN]
|
||||
body_hashes.setdefault(h, []).append(name)
|
||||
dup_funcs = {h: names for h, names in body_hashes.items() if len(names) > 1}
|
||||
except Exception:
|
||||
dup_funcs = {}
|
||||
funcs = []
|
||||
|
||||
# Find repeated string patterns (lines that appear >= REPEATED_MIN times).
|
||||
line_counts = collections.Counter(l.strip() for l in lines if l.strip())
|
||||
repeated_lines = [(l, c) for l, c in line_counts.items()
|
||||
if c >= REPEATED_MIN and len(l) > REPEATED_MIN_LEN]
|
||||
|
||||
return info, dup_funcs, repeated_lines, len(lines), funcs
|
||||
|
||||
|
||||
def format_report(info, dup_funcs, repeated_lines, nlines, funcs, fname):
|
||||
"""Build the report lines for one file (pure; no I/O).
|
||||
|
||||
Returns a list of strings whose ``print``-joined form is byte-identical to
|
||||
the original probe's per-file output.
|
||||
"""
|
||||
out = ["\n=== {0} ({1} lines, {2} funcs) ===".format(fname, nlines, len(funcs))]
|
||||
if info["sum_gen"]:
|
||||
out.append(" sum(1 for ...) x{0}".format(info["sum_gen"]))
|
||||
if info["len_listcomp"]:
|
||||
out.append(" len([...]) x{0}".format(info["len_listcomp"]))
|
||||
if info["listcomp_filter"]:
|
||||
out.append(" listcomp with if-filter x{0}".format(info["listcomp_filter"]))
|
||||
if info["count_method"]:
|
||||
out.append(" .count() calls x{0}".format(info["count_method"]))
|
||||
if info["manual_count"]:
|
||||
out.append(" manual +=1 count x{0}".format(info["manual_count"]))
|
||||
if dup_funcs:
|
||||
out.append(" DUPLICATE function bodies: {0}".format(dup_funcs))
|
||||
for l, c in repeated_lines[:5]:
|
||||
out.append(" repeated x{0}: {1}".format(c, l[:90]))
|
||||
return out
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Scan every tools/*.py and print the DRY report.
|
||||
|
||||
Entry point guarded so importing this module has no side effects. The
|
||||
output is byte-identical to ``_scan_dry.py``.
|
||||
"""
|
||||
for f in sorted(os.listdir(TOOLS)):
|
||||
if not f.endswith(".py") or f.startswith("_"):
|
||||
continue
|
||||
path = os.path.join(TOOLS, f)
|
||||
info, dup_funcs, repeated_lines, nlines, funcs = scan_file(path)
|
||||
if info["sum_gen"] >= 2 or dup_funcs or repeated_lines:
|
||||
for line in format_report(info, dup_funcs, repeated_lines, nlines, funcs, f):
|
||||
print(line)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,13 @@
|
||||
OK: py_compile passes
|
||||
OK: identical output (exit 0, 4811 bytes)
|
||||
OK: __main__ guard present
|
||||
OK: no 'global' statements in code
|
||||
OK: unused 'sys' import removed
|
||||
OK: redundant doubled 'or' clause removed; single 'len([ in l' present
|
||||
OK: bare open() replaced with 'with' context manager
|
||||
OK: scan_file, format_report, main extracted; main() calls scan_file
|
||||
OK: import is side-effect-free (no probe output at import)
|
||||
OK: scan_file real (/root/troupe-standup/tools/check_lint.py: 574 lines, 11 funcs); dup-body detection works
|
||||
OK: original still exhibits the removed smells (refactor is non-vacuous)
|
||||
|
||||
=== Verification: 11/11 checks passed ===
|
||||
@@ -0,0 +1,292 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Verification harness for the _scan_dry_refactored.py refactor.
|
||||
|
||||
Exit 0 = all checks pass; exit 1 = at least one failure.
|
||||
|
||||
Checks:
|
||||
1. Refactored file compiles (py_compile).
|
||||
2. Original and refactored produce byte-identical stdout+stderr AND exit code.
|
||||
3. ``if __name__ == "__main__"`` guard present (no probe at import time).
|
||||
4. No ``global`` statements (code only).
|
||||
5. Unused ``sys`` import removed (``sys`` absent from import set).
|
||||
6. Redundant doubled ``or`` clause gone: ``"len([" in l or "len([" in l``
|
||||
removed; single ``"len([" in l`` present.
|
||||
7. Resource leak fixed: bare ``= open(path`` absent, ``with open(path`` present.
|
||||
8. Pure helpers ``scan_file`` and ``format_report`` extracted as module-level
|
||||
functions, and ``main`` calls ``scan_file``.
|
||||
9. Import of the refactored module is side-effect-free (no stdout/stderr).
|
||||
10. ``scan_file`` is a real search, not a stub: runs on an actual tools file
|
||||
and detects duplicate function bodies.
|
||||
|
||||
Non-vacuity guard (11): the *original* ``_scan_dry.py`` must still exhibit the
|
||||
smells we removed (no __main__ guard, doubled ``or``, bare open, module-level
|
||||
scan loop). If the original no longer smells, the refactor is vacuous.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import collections
|
||||
import contextlib
|
||||
import difflib
|
||||
import importlib
|
||||
import io
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
REPO_ROOT = "/root/troupe-standup"
|
||||
WORK_DIR = os.path.join(REPO_ROOT, "wizard-work", "forge")
|
||||
ORIGINAL = "_scan_dry.py"
|
||||
REFACTORED = "_scan_dry_refactored.py"
|
||||
MOD_NAME = REFACTORED[:-3] # strip ".py"
|
||||
|
||||
# A real file the scan can run against (non-empty).
|
||||
SAMPLE_TOOL = os.path.join(REPO_ROOT, "tools", "check_lint.py")
|
||||
|
||||
|
||||
def _run(*args: str) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(args, cwd=WORK_DIR, capture_output=True, text=True)
|
||||
|
||||
|
||||
def _read(name: str) -> str:
|
||||
with open(os.path.join(WORK_DIR, name), encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
def _code_only(content: str) -> str:
|
||||
"""Strip the module docstring and comment-only lines.
|
||||
|
||||
Smell-absence checks must inspect code, not documentation: the refactored
|
||||
module's docstring *names* the patterns it removes, which would be false
|
||||
positives as residual smells.
|
||||
"""
|
||||
tree = ast.parse(content)
|
||||
skip: set[int] = set()
|
||||
if (tree.body
|
||||
and isinstance(tree.body[0], ast.Expr)
|
||||
and isinstance(tree.body[0].value, ast.Constant)
|
||||
and isinstance(tree.body[0].value.value, str)):
|
||||
ds = tree.body[0]
|
||||
for ln in range(ds.lineno, (ds.end_lineno or ds.lineno) + 1):
|
||||
skip.add(ln)
|
||||
kept = []
|
||||
for i, line in enumerate(content.splitlines(), 1):
|
||||
if i in skip or line.lstrip().startswith("#"):
|
||||
continue
|
||||
kept.append(line)
|
||||
return "\n".join(kept)
|
||||
|
||||
|
||||
def check_py_compile() -> bool:
|
||||
r = _run("python3", "-m", "py_compile", REFACTORED)
|
||||
if r.returncode != 0:
|
||||
print(f"FAIL: py_compile failed: {r.stderr}")
|
||||
return False
|
||||
print("OK: py_compile passes")
|
||||
return True
|
||||
|
||||
|
||||
def check_identical_output() -> bool:
|
||||
orig = _run("python3", ORIGINAL)
|
||||
ref = _run("python3", REFACTORED)
|
||||
ot = orig.stdout + orig.stderr
|
||||
rt = ref.stdout + ref.stderr
|
||||
if ot != rt:
|
||||
print("FAIL: original and refactored output differ")
|
||||
print("".join(difflib.unified_diff(
|
||||
ot.splitlines(keepends=True), rt.splitlines(keepends=True),
|
||||
fromfile="original (" + ORIGINAL + ")",
|
||||
tofile="refactored (" + REFACTORED + ")",
|
||||
)))
|
||||
return False
|
||||
if orig.returncode != ref.returncode:
|
||||
print(f"FAIL: exit codes differ ({orig.returncode} vs {ref.returncode})")
|
||||
return False
|
||||
print(f"OK: identical output (exit {orig.returncode}, {len(ot)} bytes)")
|
||||
return True
|
||||
|
||||
|
||||
def check_main_guard() -> bool:
|
||||
content = _read(REFACTORED)
|
||||
if '__name__ == "__main__"' not in content:
|
||||
print("FAIL: no __main__ guard")
|
||||
return False
|
||||
print("OK: __main__ guard present")
|
||||
return True
|
||||
|
||||
|
||||
def check_no_globals() -> bool:
|
||||
code = _code_only(_read(REFACTORED))
|
||||
for i, line in enumerate(code.splitlines(), 1):
|
||||
if line.lstrip().startswith("global "):
|
||||
print(f"FAIL: 'global' found at code line {i}: {line.strip()}")
|
||||
return False
|
||||
print("OK: no 'global' statements in code")
|
||||
return True
|
||||
|
||||
|
||||
def check_unused_sys_removed() -> bool:
|
||||
tree = ast.parse(_read(REFACTORED))
|
||||
imported = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
imported.add(a.name)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
if node.module:
|
||||
imported.add(node.module)
|
||||
if "sys" in imported:
|
||||
print("FAIL: 'sys' still imported")
|
||||
return False
|
||||
print("OK: unused 'sys' import removed")
|
||||
return True
|
||||
|
||||
|
||||
def check_redundant_or_removed() -> bool:
|
||||
code = _code_only(_read(REFACTORED))
|
||||
doubled = '"len([" in l or "len([" in l'
|
||||
if doubled in code:
|
||||
print("FAIL: redundant doubled 'or' clause still present")
|
||||
return False
|
||||
if 'if "len([" in l' not in code:
|
||||
print("FAIL: expected single '\"len([\" in l' clause not found")
|
||||
return False
|
||||
print("OK: redundant doubled 'or' clause removed; single 'len([ in l' present")
|
||||
return True
|
||||
|
||||
|
||||
def check_resource_leak_fixed() -> bool:
|
||||
code = _code_only(_read(REFACTORED))
|
||||
if "= open(path" in code:
|
||||
print("FAIL: bare '= open(path' (resource leak) still present")
|
||||
return False
|
||||
if "with open(path" not in code:
|
||||
print("FAIL: expected 'with open(path' context manager not found")
|
||||
return False
|
||||
print("OK: bare open() replaced with 'with' context manager")
|
||||
return True
|
||||
|
||||
|
||||
def check_helpers_extracted() -> bool:
|
||||
tree = ast.parse(_read(REFACTORED))
|
||||
func_names = {
|
||||
n.name for n in tree.body
|
||||
if isinstance(n, ast.FunctionDef)
|
||||
}
|
||||
for required in ("scan_file", "format_report", "main"):
|
||||
if required not in func_names:
|
||||
print(f"FAIL: {required} not defined as module-level function")
|
||||
return False
|
||||
main_func = next(n for n in tree.body
|
||||
if isinstance(n, ast.FunctionDef) and n.name == "main")
|
||||
calls = {
|
||||
n.func.id for n in ast.walk(main_func)
|
||||
if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
|
||||
}
|
||||
if "scan_file" not in calls:
|
||||
print("FAIL: main() does not call scan_file()")
|
||||
return False
|
||||
print("OK: scan_file, format_report, main extracted; main() calls scan_file")
|
||||
return True
|
||||
|
||||
|
||||
def check_import_side_effect_free() -> bool:
|
||||
"""Importing the refactored module must not run the probe or print."""
|
||||
sys.modules.pop(MOD_NAME, None)
|
||||
if WORK_DIR not in sys.path:
|
||||
sys.path.insert(0, WORK_DIR)
|
||||
out, err = io.StringIO(), io.StringIO()
|
||||
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
||||
importlib.import_module(MOD_NAME)
|
||||
if out.getvalue() or err.getvalue():
|
||||
print(f"FAIL: import produced output: stdout={out.getvalue()!r} stderr={err.getvalue()!r}")
|
||||
return False
|
||||
print("OK: import is side-effect-free (no probe output at import)")
|
||||
return True
|
||||
|
||||
|
||||
def check_search_is_real() -> bool:
|
||||
"""scan_file runs on a real tool file and detects duplicate bodies."""
|
||||
if MOD_NAME in sys.modules:
|
||||
mod = sys.modules[MOD_NAME]
|
||||
else:
|
||||
sys.path.insert(0, WORK_DIR)
|
||||
mod = importlib.import_module(MOD_NAME)
|
||||
info, dup_funcs, repeated, nlines, funcs = mod.scan_file(SAMPLE_TOOL)
|
||||
if not isinstance(info, collections.Counter):
|
||||
print("FAIL: scan_file[0] is not a Counter")
|
||||
return False
|
||||
if not isinstance(nlines, int) or nlines <= 0:
|
||||
print(f"FAIL: scan_file[3] nlines not a positive int ({nlines!r})")
|
||||
return False
|
||||
if not isinstance(funcs, list):
|
||||
print("FAIL: scan_file[4] funcs is not a list")
|
||||
return False
|
||||
# Counter-control: a temp file with two identical bodies must be detected.
|
||||
smelly = "def f():\n x = 1\n x = 2\n return x\n\n\n\n\ndef g():\n x = 1\n x = 2\n return x\n"
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as tf:
|
||||
tf.write(smelly)
|
||||
tmp_path = tf.name
|
||||
try:
|
||||
sinfo, sdup, srep, snl, sfn = mod.scan_file(tmp_path)
|
||||
if not sdup:
|
||||
print("FAIL: scan_file failed to detect duplicate function bodies")
|
||||
return False
|
||||
finally:
|
||||
os.unlink(tmp_path)
|
||||
print(f"OK: scan_file real ({SAMPLE_TOOL}: {nlines} lines, {len(funcs)} funcs); dup-body detection works")
|
||||
return True
|
||||
|
||||
|
||||
def check_original_still_smells() -> bool:
|
||||
"""The original must still carry the smells we removed (non-vacuous)."""
|
||||
src = _read(ORIGINAL)
|
||||
code = _code_only(src)
|
||||
problems = []
|
||||
# The original must LACK a __main__ guard (that is the smell we fixed).
|
||||
if '__name__ == "__main__"' in src:
|
||||
problems.append("original already has __main__ guard (refactor not needed)")
|
||||
doubled = '"len([" in l or "len([" in l'
|
||||
if doubled not in code:
|
||||
problems.append("redundant doubled 'or' clause missing from original (expected)")
|
||||
if "= open(path" not in code:
|
||||
problems.append("bare '= open(path' (resource leak) missing from original (expected)")
|
||||
# The original must execute the scan at import time: a module-level for-loop.
|
||||
tree = ast.parse(src)
|
||||
has_module_level_for = any(isinstance(n, ast.For) for n in tree.body)
|
||||
if not has_module_level_for:
|
||||
problems.append("no module-level scan loop in original (expected)")
|
||||
if problems:
|
||||
print("FAIL: original does not exhibit the expected smells:")
|
||||
for p in problems:
|
||||
print(f" - {p}")
|
||||
return False
|
||||
print("OK: original still exhibits the removed smells (refactor is non-vacuous)")
|
||||
return True
|
||||
|
||||
|
||||
def main() -> None:
|
||||
checks = [
|
||||
check_py_compile,
|
||||
check_identical_output,
|
||||
check_main_guard,
|
||||
check_no_globals,
|
||||
check_unused_sys_removed,
|
||||
check_redundant_or_removed,
|
||||
check_resource_leak_fixed,
|
||||
check_helpers_extracted,
|
||||
check_import_side_effect_free,
|
||||
check_search_is_real,
|
||||
check_original_still_smells,
|
||||
]
|
||||
results = [fn() for fn in checks]
|
||||
passed = sum(results)
|
||||
total = len(results)
|
||||
print(f"\n=== Verification: {passed}/{total} checks passed ===")
|
||||
if passed != total:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user