913 lines
41 KiB
Python
913 lines
41 KiB
Python
#!/usr/bin/env python3
|
|
"""import_resolve_check.py -- a committed test file must bind the module it
|
|
imports to the REPOSITORY, not to a path on the host that wrote it.
|
|
|
|
WHY THIS EXISTS
|
|
Q61(b) was "the gate resolves the module from a host path ahead of its own
|
|
directory". Closing it meant fixing `wizard-work/beam/test_toposort.py` to
|
|
resolve from `pathlib.Path(__file__).resolve().parent` and to assert the
|
|
imported module's `__file__` is the repo file -- one file, and the closure
|
|
sentence read as if the shape were gone. It was not.
|
|
|
|
On 2026-10-27 `wizard-work/beam/test_lfu_cache.py:2` landed as
|
|
`sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")`. Measured on that
|
|
head (`5886bdd`): with the repo copy's `get` poisoned to `return None`, the
|
|
suite still read **8 passed**, because all eight cells graded the HOST copy;
|
|
the lane's own `bash wizard-work/beam/CHECK` read **rc 1**,
|
|
`ImportError: 'test_lru_cache' module incorrectly imported from
|
|
'/root/.hermes/wizard-outputs/beam'`; deleting only that line read **rc 0**,
|
|
`Ran 230 tests ... OK`. A suite that grades a file nobody committed is worse
|
|
than no suite: it certifies the wrong bytes. Counting the class rather than
|
|
estimating it found **41** absolute `sys.path.insert(0, "/root...")` lines
|
|
repo-wide, **9** in `test_*.py`, **2** of those prose recording the Q61 fix,
|
|
**7 live**.
|
|
|
|
A per-file fix was tried and did not hold. This is the CLASS guard: one
|
|
instrument that reads every committed test file and names the ones that
|
|
bind outside the tree.
|
|
|
|
WHAT IT MEASURES -- a byte reading, not a text one
|
|
The scan parses each tracked test file with `ast` and grades CALLS and
|
|
ASSIGNMENTS, never prose. That distinction is the whole reason it is not a
|
|
`grep`: two of the nine hits are docstrings that RECORD the Q61 fix
|
|
(`wizard-work/beam/test_toposort.py:12`, `test_robin_hood_hash.py:5`), and a
|
|
grep-based scan reports them as faults forever -- which is how a guard gets
|
|
switched off. `ast` also cannot be fooled by a commented-out insert.
|
|
|
|
WHAT COUNTS AS A TEST FILE -- three names, not one (Q100)
|
|
`test_*.py`, `_test_*.py`, `*_test.py`. The first draft of this guard read
|
|
only `test_*.py`, and Q100 measured what that cost: two tracked forge files,
|
|
`wizard-work/forge/_test_roundtrip.py:13` and
|
|
`_test_roundtrip_refactored.py:27`, each bound
|
|
`/root/.hermes/wizard-outputs/forge` and read **34 passed, 0 failed** while
|
|
the repo copy of `chunker.py` under them was poisoned to `return []`. A test
|
|
suite is a file that runs assertions and imports the module under test; the
|
|
spelling of its name is not evidence about what it does, so the class is
|
|
closed on the two shapes the first pass missed. A name that merely CONTAINS
|
|
`test` (`latest.py`, `attestation.py`, `protest.py`) is not a test file and
|
|
stays an ADVISORY row -- cell 21 pins that.
|
|
|
|
|
|
OUTSIDE_REPO `sys.path.insert/append/extend`, a `sys.path = [...]`
|
|
assignment, or `importlib.util.spec_from_file_location`
|
|
carries a constant absolute path that resolves OUTSIDE the
|
|
repo root. The module under test is whatever sits on that
|
|
host path. FATAL.
|
|
ABS_PATH_IN_REPO the same construct, and the constant is absolute but
|
|
INSIDE the repo (the `ledger/test_scratch_manifest.py:12`
|
|
shape: `/root/troupe-standup/wizard-work/ledger`). It
|
|
grades the right bytes on THIS host and the wrong ones --
|
|
or none -- in any clone, which is the same defect one
|
|
release later. FATAL.
|
|
UNGRADABLE a tracked test file that cannot be decoded as UTF-8 or
|
|
parsed as Python. Not a pass: rc 2, because a file this
|
|
instrument could not read is a file it did not grade.
|
|
ADVISORY an absolute binding in a tracked file that is not a test
|
|
file by the name rule (`test_*.py`, `_test_*.py`,
|
|
`*_test.py`). Counted and printed, NOT graded -- the class
|
|
is 30 probe scripts across other lanes, most of them
|
|
throwaway reproductions that name a host path on purpose,
|
|
and reddening on them would make this guard unrunnable.
|
|
`--all-files` grades every tracked `*.py` instead; it is
|
|
expected to read RED today, and that number is the class's
|
|
true size.
|
|
|
|
Values are READ, not pattern-matched: the constant string is extracted from
|
|
the AST (including the leading part of an f-string) and `os.path.realpath`
|
|
decides containment. `sys.path.insert(0, "lib")` and
|
|
`sys.path.insert(0, str(Path(__file__).resolve().parent))` carry no absolute
|
|
constant and are therefore silent -- both are correct bindings and a scan
|
|
that flagged them would be noise.
|
|
|
|
--import-probe -- THE SECOND READING, FOR WHAT STATIC SCAN CANNOT SEE
|
|
A path built at runtime (`os.path.join(os.environ["HOME"], "wizard-outputs")`)
|
|
holds no constant, so the static scan cannot grade it. `--import-probe`
|
|
loads each tracked test file in a CLEAN SUBPROCESS (`python3 -I`, so cwd and
|
|
PYTHONPATH are not inherited), then reports what the file actually bound:
|
|
every `sys.path` entry it added that resolves outside the repo, and every
|
|
module it left in `sys.modules` whose `__file__` is outside the repo and
|
|
outside the interpreter's own stdlib/site-packages. A path entry or module
|
|
from outside the tree is the SAME fault, found the other way. An exception
|
|
during the load is printed as `probe: UNGRADED <file> (<exc>)` and does not
|
|
change rc: a test file that needs a dependency this host lacks is not a
|
|
path-binding fault, and folding the two would make the axis unusable.
|
|
`--import-probe` executes test code, so it is opt-in and never part of the
|
|
default reading.
|
|
|
|
THE EXIT CONTRACT -- the house three codes
|
|
0 every tracked test file was read and none binds outside the repo
|
|
1 at least one file was graded and binds outside (or, with --all-files, any
|
|
tracked file does)
|
|
2 the inventory or a file could not be read -- not a git work tree, git
|
|
unreadable, a tracked test file that will not decode or parse, or a
|
|
--json target this host cannot write
|
|
|
|
rc 2 outranks rc 1 for the reason verify.py states: a run that could not
|
|
read part of its inventory has not earned a claim about the whole tree.
|
|
|
|
NOT CLAIMED
|
|
That every import in a test file is graded -- only the four constructs above,
|
|
with constant paths. That a clean reading means the test asserts its own
|
|
binding (`test_toposort.py` does; `--import-probe` is what checks the fact
|
|
rather than the intent). That non-test files are safe: 30 carry the same
|
|
line today and `--all-files` names them. That this fixes the seven rows --
|
|
it names them; the edits are separate and are recorded with their gates.
|
|
|
|
USAGE
|
|
python3 import_resolve_check.py
|
|
python3 import_resolve_check.py --repo /path/to/worktree --rev HEAD
|
|
python3 import_resolve_check.py --all-files
|
|
python3 import_resolve_check.py --import-probe
|
|
python3 import_resolve_check.py --json PATH
|
|
python3 import_resolve_check.py --selftest # 38 cells
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import warnings
|
|
|
|
OUTSIDE_REPO = "OUTSIDE_REPO"
|
|
ABS_PATH_IN_REPO = "ABS_PATH_IN_REPO"
|
|
UNGRADABLE = "UNGRADABLE"
|
|
|
|
FATAL_CLASSES = (OUTSIDE_REPO, ABS_PATH_IN_REPO)
|
|
|
|
# The repo this file lives in. Same rule as every other tool here: a bare run
|
|
# in a clone grades the clone, not a fixed path on the host that wrote it.
|
|
DEFAULT_REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
TEST_NAME = re.compile(r"(^|/)(_?test_[^/]*|[^/]*_test)\.py$")
|
|
TRACEBACK = "Traceback (most recent call last)"
|
|
|
|
# sys.path mutators whose last argument is the path being added.
|
|
PATH_CALLS = {"insert", "append", "extend"}
|
|
|
|
|
|
class CannotGrade(Exception):
|
|
"""The inventory (or one of its files) could not be read."""
|
|
|
|
|
|
# --------------------------------------------------------------------- inventory
|
|
|
|
def _git(repo, *args):
|
|
try:
|
|
proc = subprocess.run(["git", "-C", repo] + list(args),
|
|
capture_output=True, text=True)
|
|
except OSError as exc:
|
|
raise CannotGrade("git is not runnable: %s" % exc)
|
|
if proc.returncode != 0:
|
|
why = [l for l in (proc.stderr or "").strip().splitlines() if l.strip()]
|
|
raise CannotGrade("git %s failed: %s"
|
|
% (" ".join(args[:2]),
|
|
why[-1] if why else "rc %d" % proc.returncode))
|
|
return proc.stdout
|
|
|
|
|
|
def work_tree(repo):
|
|
"""`--repo` must be a work tree ROOT; ask git, as verify.py does."""
|
|
repo = os.path.abspath(repo)
|
|
if not os.path.isdir(repo):
|
|
raise CannotGrade("cannot read repo: %s is not a directory" % repo)
|
|
top = _git(repo, "rev-parse", "--show-toplevel").strip()
|
|
if not top:
|
|
raise CannotGrade("cannot read repo: %s is not a git work tree" % repo)
|
|
if os.path.realpath(top) != os.path.realpath(repo):
|
|
raise CannotGrade("cannot read repo: %s is inside the work tree rooted "
|
|
"at %s; --repo takes the root" % (repo, top))
|
|
return repo
|
|
|
|
|
|
def _control_chars(path):
|
|
"""True when a tracked path carries a byte a path cannot meaningfully hold."""
|
|
return any(ord(ch) < 0x20 or ord(ch) == 0x7f for ch in path)
|
|
|
|
|
|
def inventory(repo, rev):
|
|
"""[(path, text)] for every tracked *.py at `rev` (WORKTREE = the index).
|
|
|
|
TRACKED ONLY, for doc_census_check's reason: an untracked scratch file a
|
|
lane is mid-way through writing is not a committed test suite, and grading
|
|
it would red the gate for work that has not landed. `WORKTREE` is
|
|
`git ls-files`, so a file that is added-but-not-committed is graded -- the
|
|
tick's own edit is exactly what a lane needs graded before it commits.
|
|
"""
|
|
if rev == "WORKTREE":
|
|
out = _git(repo, "ls-files", "-z")
|
|
else:
|
|
out = _git(repo, "ls-tree", "-r", "-z", "--name-only", rev)
|
|
paths = [p for p in out.split("\0") if p]
|
|
rows = []
|
|
for path in paths:
|
|
if not path.endswith(".py"):
|
|
continue
|
|
if rev == "WORKTREE":
|
|
full = os.path.join(repo, path)
|
|
try:
|
|
with open(full, "rb") as fh:
|
|
raw = fh.read()
|
|
except OSError as exc:
|
|
rows.append((path, None, "cannot read %s: %s" % (full, exc)))
|
|
continue
|
|
else:
|
|
try:
|
|
raw = _git(repo, "show", "%s:%s" % (rev, path)).encode("utf-8")
|
|
except CannotGrade as exc:
|
|
rows.append((path, None, "cannot read %s:%s: %s"
|
|
% (rev, path, exc)))
|
|
continue
|
|
try:
|
|
text = raw.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
rows.append((path, None, "not UTF-8: %s" % exc))
|
|
continue
|
|
rows.append((path, text, None))
|
|
return rows
|
|
|
|
|
|
# ------------------------------------------------------------------- the scan
|
|
|
|
def _const_text(node):
|
|
"""The literal a path argument is written as, or None if it is computed.
|
|
|
|
Returns (text, is_prefix). An f-string contributes only its LEADING
|
|
constant part, flagged as a prefix: `f"/root/{name}"` cannot be resolved
|
|
exactly, but a leading `/root/` is enough to say the binding is absolute
|
|
and to name it. A value that starts with a placeholder (`f"{base}/x"`) is
|
|
computed and returns None -- the static scan does not guess.
|
|
"""
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return node.value, False
|
|
if isinstance(node, ast.JoinedStr):
|
|
if node.values and isinstance(node.values[0], ast.Constant) \
|
|
and isinstance(node.values[0].value, str):
|
|
return node.values[0].value, True
|
|
return None
|
|
|
|
|
|
def _sys_path_mutator(call):
|
|
"""'insert' / 'append' / 'extend' for a `sys.path.<x>(...)` call, else None."""
|
|
func = call.func
|
|
if not isinstance(func, ast.Attribute) or func.attr not in PATH_CALLS:
|
|
return None
|
|
inner = func.value
|
|
if not isinstance(inner, ast.Attribute) or inner.attr != "path":
|
|
return None
|
|
if not isinstance(inner.value, ast.Name) or inner.value.id != "sys":
|
|
return None
|
|
return func.attr
|
|
|
|
|
|
def _is_spec_load(call):
|
|
func = call.func
|
|
return (isinstance(func, ast.Attribute)
|
|
and func.attr == "spec_from_file_location")
|
|
|
|
|
|
def _path_args(call, attr):
|
|
"""The path-valued arguments of a graded call (index 0 of a list = all)."""
|
|
if attr == "insert":
|
|
return [call.args[1]] if len(call.args) > 1 else []
|
|
if attr in ("append", "extend") or attr == "spec":
|
|
out = []
|
|
for arg in call.args[1:] if attr == "spec" else call.args[:1]:
|
|
if isinstance(arg, (ast.List, ast.Tuple)):
|
|
out.extend(arg.elts)
|
|
else:
|
|
out.append(arg)
|
|
return out
|
|
return []
|
|
|
|
|
|
def classify(raw, repo_real):
|
|
"""OUTSIDE_REPO / ABS_PATH_IN_REPO for an absolute literal, else None."""
|
|
text = raw.strip()
|
|
if not text:
|
|
return None
|
|
expanded = os.path.expanduser(text)
|
|
if not os.path.isabs(expanded):
|
|
return None # relative: not a host path
|
|
real = os.path.realpath(expanded)
|
|
if real == repo_real or real.startswith(repo_real + os.sep):
|
|
return ABS_PATH_IN_REPO
|
|
return OUTSIDE_REPO
|
|
|
|
|
|
def scan(text, path, repo_real):
|
|
"""Findings for one file's source: [(line, cls, how, raw, is_prefix)]."""
|
|
try:
|
|
with warnings.catch_warnings():
|
|
# A tracked file whose source trips a SyntaxWarning (an invalid
|
|
# "\u" escape in a docstring) must not print to stderr from a scan
|
|
# that only builds a tree: the warning is not this instrument's
|
|
# finding, and it landed in the middle of the report on the first
|
|
# live run.
|
|
warnings.simplefilter("ignore")
|
|
tree = ast.parse(text, filename=path)
|
|
except SyntaxError as exc:
|
|
raise CannotGrade("%s does not parse: line %s: %s"
|
|
% (path, exc.lineno, exc.msg))
|
|
found = []
|
|
|
|
def note(node, how, args):
|
|
for arg in args:
|
|
lit = _const_text(arg)
|
|
if not lit:
|
|
continue
|
|
raw, is_prefix = lit
|
|
cls = classify(raw, repo_real)
|
|
if cls:
|
|
found.append((getattr(node, "lineno", 0), cls, how, raw,
|
|
is_prefix))
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Call):
|
|
attr = _sys_path_mutator(node)
|
|
if attr:
|
|
note(node, "sys.path." + attr, _path_args(node, attr))
|
|
elif _is_spec_load(node):
|
|
note(node, "spec_from_file_location", _path_args(node, "spec"))
|
|
elif isinstance(node, ast.Assign):
|
|
tgt = node.targets[0] if len(node.targets) == 1 else None
|
|
if (isinstance(tgt, ast.Attribute) and tgt.attr == "path"
|
|
and isinstance(tgt.value, ast.Name)
|
|
and tgt.value.id == "sys"):
|
|
vals = (node.value.elts if isinstance(node.value, (ast.List,
|
|
ast.Tuple))
|
|
else [node.value])
|
|
note(node, "sys.path = [...]", vals)
|
|
return sorted(found)
|
|
|
|
|
|
# ------------------------------------------------------------------ import probe
|
|
|
|
_PROBE = r"""
|
|
import importlib.util, json, os, sys
|
|
path, repo = sys.argv[1], os.path.realpath(sys.argv[2])
|
|
thedir = os.path.dirname(os.path.abspath(path))
|
|
before = list(sys.path)
|
|
before_mods = set(sys.modules)
|
|
sys.path.insert(0, thedir)
|
|
row = {"path": path, "added": [], "modules": [], "error": None}
|
|
try:
|
|
spec = importlib.util.spec_from_file_location("_probe_" + os.path.basename(path)[:-3], path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
except BaseException as exc: # noqa: BLE001 -- the probe grades binding, not the suite
|
|
row["error"] = "%s: %s" % (type(exc).__name__, exc)
|
|
seen = set(before)
|
|
row["added"] = [p for p in sys.path if p not in seen]
|
|
lib = os.path.realpath(sys.base_prefix)
|
|
here = os.path.realpath(sys.prefix)
|
|
for name, m in list(sys.modules.items()):
|
|
if name in before_mods:
|
|
continue
|
|
f = getattr(m, "__file__", None)
|
|
if not f:
|
|
continue
|
|
real = os.path.realpath(f)
|
|
if (real.startswith(repo + os.sep) or real.startswith(lib + os.sep)
|
|
or real.startswith(here + os.sep)):
|
|
continue
|
|
row["modules"].append([name, real])
|
|
print(json.dumps(row))
|
|
"""
|
|
|
|
|
|
def import_probe(repo, path, timeout=20):
|
|
"""Load one test file in a clean subprocess; return its binding row."""
|
|
try:
|
|
proc = subprocess.run([sys.executable, "-I", "-c", _PROBE,
|
|
os.path.join(repo, path), repo],
|
|
capture_output=True, text=True, timeout=timeout,
|
|
cwd=os.path.dirname(os.path.join(repo, path)))
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
return {"path": path, "added": [], "modules": [],
|
|
"error": "TIMEOUT/OSError: %s" % exc}
|
|
for line in reversed((proc.stdout or "").strip().splitlines()):
|
|
try:
|
|
row = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if isinstance(row, dict) and row.get("path"):
|
|
return row
|
|
return {"path": path, "added": [], "modules": [],
|
|
"error": "probe produced no row (rc %d): %s"
|
|
% (proc.returncode, (proc.stderr or "").strip()
|
|
.splitlines()[-1:] or "")}
|
|
|
|
|
|
# ---------------------------------------------------------------------- report
|
|
|
|
def run(repo, rev, all_files, probe, quiet=False):
|
|
repo = work_tree(repo)
|
|
repo_real = os.path.realpath(repo)
|
|
rows = inventory(repo, rev)
|
|
if not quiet:
|
|
print("import_resolve_check -- repo %s at %s" % (repo, rev))
|
|
|
|
faults = [] # (path, line, cls, how, raw, is_prefix)
|
|
ungraded = [] # (path, why)
|
|
bad_paths = [] # tracked paths that are not a single-line path name
|
|
advisory = 0
|
|
graded = 0
|
|
for path, text, why in rows:
|
|
if _control_chars(path):
|
|
# A tracked path carrying a newline is an INDEX defect, not a
|
|
# binding: `git ls-tree -r HEAD` at e179df3 carries exactly one --
|
|
# a beam CHECK file whose name is the whole shell body it was meant
|
|
# to hold (`wizard-work/beam/CHECK\ncontent">\n#!/bin/bash...`).
|
|
# It is named here and skipped, because a name that cannot be a
|
|
# path cannot be a test file, and inheriting another lane's index
|
|
# damage as a permanent rc 2 would be the "gate that gets switched
|
|
# off" this repo already has a rule about. The owner fixes it.
|
|
bad_paths.append(path)
|
|
continue
|
|
is_test = bool(TEST_NAME.search(path))
|
|
if why:
|
|
if is_test:
|
|
ungraded.append((path, why))
|
|
continue
|
|
if not is_test and not all_files:
|
|
if scan_silent(text, repo_real):
|
|
advisory += 1
|
|
continue
|
|
graded += 1
|
|
try:
|
|
hits = scan(text, path, repo_real)
|
|
except CannotGrade as exc:
|
|
ungraded.append((path, str(exc)))
|
|
continue
|
|
for line, cls, how, raw, is_prefix in hits:
|
|
faults.append((path, line, cls, how, raw, is_prefix))
|
|
|
|
probe_rows = []
|
|
if probe:
|
|
for path, text, why in rows:
|
|
if why or not TEST_NAME.search(path):
|
|
continue
|
|
row = import_probe(repo, path)
|
|
# Only OUTSIDE_REPO is a probe fault: the probe itself puts the
|
|
# test's own directory on sys.path, so an in-repo absolute entry is
|
|
# the harness's line, not the file's binding.
|
|
outside = [p for p in row["added"]
|
|
if classify(p, repo_real) == OUTSIDE_REPO]
|
|
mods = [m for m in row["modules"]
|
|
if not m[1].startswith(repo_real + os.sep)]
|
|
row["outside_added"] = outside
|
|
row["outside_modules"] = mods
|
|
probe_rows.append(row)
|
|
if outside or mods:
|
|
faults.append((path, 0, OUTSIDE_REPO, "--import-probe",
|
|
outside[0] if outside else mods[0][1], False))
|
|
|
|
if not quiet:
|
|
for path, line, cls, how, raw, is_prefix in faults:
|
|
print("FAULT %-16s %s:%s %s -> %s%s"
|
|
% (cls, path, line, how, raw,
|
|
" (f-string prefix)" if is_prefix else ""))
|
|
for path, why in ungraded:
|
|
print("UNGRADED %s %s" % (path, why))
|
|
for path in bad_paths:
|
|
shown = path if len(path) <= 60 else path[:57] + "..."
|
|
print("SKIP BAD_PATH %r (a tracked path with control characters: "
|
|
"not a path, not graded -- an index defect for its owner)"
|
|
% shown)
|
|
if probe:
|
|
for row in probe_rows:
|
|
if row["error"]:
|
|
print("probe: UNGRADED %s (%s)" % (row["path"],
|
|
row["error"]))
|
|
elif not row["outside_added"] and not row["outside_modules"]:
|
|
print("probe: in-tree %s" % row["path"])
|
|
if advisory:
|
|
print("advisory: %d tracked non-test file(s) carry an absolute "
|
|
"binding (not graded; --all-files grades them)" % advisory)
|
|
|
|
classes = {}
|
|
for f in faults:
|
|
classes[f[2]] = classes.get(f[2], 0) + 1
|
|
detail = ", ".join("%s %d" % (k, classes[k]) for k in sorted(classes))
|
|
row = {"repo": repo, "rev": rev, "graded": graded, "faults": faults,
|
|
"ungraded": ungraded, "advisory": advisory, "bad_paths": bad_paths,
|
|
"probe": probe_rows}
|
|
print("import_resolve_check: %d file(s) graded, %d fault(s)%s%s"
|
|
% (graded, len(faults), " (%s)" % detail if detail else "",
|
|
", %d bad path(s) skipped" % len(bad_paths) if bad_paths else ""))
|
|
if ungraded:
|
|
print("cannot grade: %d file(s) unreadable" % len(ungraded))
|
|
return 2, row
|
|
if faults:
|
|
print("FAIL: %d finding(s) in %d test file(s) bind a module to a path "
|
|
"outside the repo" % (len(faults), len({f[0] for f in faults})))
|
|
return 1, row
|
|
return 0, row
|
|
|
|
|
|
def scan_silent(text, repo_real):
|
|
"""Cheap question for the advisory count: does this file bind anything?"""
|
|
try:
|
|
return bool(scan(text, "<advisory>", repo_real))
|
|
except CannotGrade:
|
|
return False
|
|
|
|
|
|
# -------------------------------------------------------------------- selftest
|
|
|
|
def _write(path, text):
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
with open(path, "w") as fh:
|
|
fh.write(text)
|
|
return path
|
|
|
|
|
|
def _git_init(repo):
|
|
subprocess.run(["git", "init", "-q", repo], check=True,
|
|
capture_output=True, text=True)
|
|
|
|
|
|
def selftest():
|
|
cells = 0
|
|
passed = 0
|
|
failures = []
|
|
|
|
def check(label, got, want):
|
|
nonlocal cells, passed
|
|
cells += 1
|
|
if got == want:
|
|
passed += 1
|
|
else:
|
|
failures.append("%s\n got: %r\n want: %r" % (label, got, want))
|
|
|
|
def call(*args, **kw):
|
|
"""run() with its report swallowed and a refusal returned as rc 2.
|
|
|
|
The corpus grades the CLI contract, so a CannotGrade has to arrive as
|
|
the code the CLI would exit with -- not as a traceback out of the
|
|
fixture, which is what the first draft of cell 16 did.
|
|
"""
|
|
buf = io.StringIO()
|
|
old, sys.stdout = sys.stdout, buf
|
|
try:
|
|
return run(*args, **kw)
|
|
except CannotGrade:
|
|
return 2, {}
|
|
finally:
|
|
sys.stdout = old
|
|
|
|
def call_main(argv):
|
|
"""main() through the CLI path, with its report swallowed."""
|
|
out, err = io.StringIO(), io.StringIO()
|
|
old = sys.stdout, sys.stderr
|
|
sys.stdout, sys.stderr = out, err
|
|
try:
|
|
return main(argv)
|
|
finally:
|
|
sys.stdout, sys.stderr = old
|
|
|
|
clean = ('import pathlib, sys\n'
|
|
'HERE = pathlib.Path(__file__).resolve().parent\n'
|
|
'sys.path.insert(0, str(HERE))\n'
|
|
'from target import thing\n')
|
|
|
|
with tempfile.TemporaryDirectory(prefix="irc-selftest-") as td:
|
|
# 1. a repo whose one test file binds from __file__ is clean
|
|
repo = os.path.join(td, "c1")
|
|
_write(os.path.join(repo, "test_ok.py"), clean)
|
|
_write(os.path.join(repo, "target.py"), "thing = 1\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("1 a __file__ binding is clean -> rc 0", rc, 0)
|
|
check("1b and exactly one file was graded", obj["graded"], 1)
|
|
|
|
# 2. THE DEFECT: a constant host path outside the repo
|
|
repo = os.path.join(td, "c2")
|
|
_write(os.path.join(repo, "test_bad.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/beam")\n'
|
|
'from target import thing\n')
|
|
_write(os.path.join(repo, "target.py"), "thing = 1\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("2 a host path outside the repo -> rc 1", rc, 1)
|
|
check("2b the class, the file and the line are named",
|
|
(obj["faults"][0][0], obj["faults"][0][1], obj["faults"][0][2]),
|
|
("test_bad.py", 2, OUTSIDE_REPO))
|
|
|
|
# 3. the ledger shape: absolute but INSIDE the repo -- right bytes here,
|
|
# none in a clone, which is the same defect one release later
|
|
repo = os.path.join(td, "c3")
|
|
inside = os.path.join(repo, "sub")
|
|
_write(os.path.join(repo, "test_abs_in.py"),
|
|
'import sys\nsys.path.insert(0, "%s")\n' % inside)
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("3 an absolute path inside the repo -> rc 1", rc, 1)
|
|
check("3b classed ABS_PATH_IN_REPO, not OUTSIDE_REPO",
|
|
obj["faults"][0][2], ABS_PATH_IN_REPO)
|
|
|
|
# 4. AST, NOT GREP: a docstring that records the old line is not a fault.
|
|
# Two of the nine hits in the live tree are exactly this shape, and a
|
|
# grep-based guard reports them forever.
|
|
repo = os.path.join(td, "c4")
|
|
_write(os.path.join(repo, "test_prose.py"),
|
|
'"""The previous line read\n'
|
|
' sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")\n'
|
|
'and it graded a host copy."""\n'
|
|
'import pathlib, sys\n'
|
|
'sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))\n'
|
|
'# sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("4 a commented/docstring mention is not a fault -> rc 0", rc, 0)
|
|
|
|
# 5. a non-test file is an ADVISORY, not a fault: the class is 30 probe
|
|
# scripts in other lanes and a guard that reds on them never gets run.
|
|
repo = os.path.join(td, "c5")
|
|
_write(os.path.join(repo, "test_ok.py"), clean)
|
|
_write(os.path.join(repo, "target.py"), "thing = 1\n")
|
|
_write(os.path.join(repo, "probe_thing.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/forge")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("5 a non-test binding is advisory, not a fault -> rc 0", rc, 0)
|
|
check("5b and it is counted, not silent", obj["advisory"], 1)
|
|
|
|
# 6. --all-files promotes the advisory to a fault: the class's size is
|
|
# available on demand rather than hidden
|
|
rc, obj = call(repo, "WORKTREE", True, False, quiet=True)
|
|
check("6 --all-files grades the non-test file -> rc 1", rc, 1)
|
|
check("6b and names it", obj["faults"][0][0], "probe_thing.py")
|
|
|
|
# 7. append, not just insert
|
|
repo = os.path.join(td, "c7")
|
|
_write(os.path.join(repo, "test_app.py"),
|
|
'import sys\nsys.path.append("/opt/lib")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("7 sys.path.append is graded too -> rc 1", rc, 1)
|
|
check("7b named as an append", obj["faults"][0][3], "sys.path.append")
|
|
|
|
# 8. a whole-list assignment
|
|
repo = os.path.join(td, "c8")
|
|
_write(os.path.join(repo, "test_assign.py"),
|
|
'import sys\nsys.path = ["/root/wizard-outputs/spark"]\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("8 a sys.path list assignment is graded -> rc 1", rc, 1)
|
|
|
|
# 9. opening a module by absolute file path
|
|
repo = os.path.join(td, "c9")
|
|
_write(os.path.join(repo, "test_spec.py"),
|
|
'import importlib.util\n'
|
|
'spec = importlib.util.spec_from_file_location("m", "/root/x/m.py")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("9 spec_from_file_location by host path -> rc 1", rc, 1)
|
|
check("9b named", obj["faults"][0][3], "spec_from_file_location")
|
|
|
|
# 10. an f-string with an absolute LEADING part is named by its prefix
|
|
repo = os.path.join(td, "c10")
|
|
_write(os.path.join(repo, "test_f.py"),
|
|
'import os, sys\n'
|
|
'sys.path.insert(0, f"/root/.hermes/wizard-outputs/{os.sep}spark")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("10 an f-string host prefix -> rc 1", rc, 1)
|
|
check("10b flagged as a prefix rather than resolved",
|
|
obj["faults"][0][5], True)
|
|
|
|
# 11. computed and relative arguments are NOT graded: the static scan
|
|
# does not guess, and `sys.path.insert(0, "lib")` is not the defect
|
|
repo = os.path.join(td, "c11")
|
|
_write(os.path.join(repo, "test_computed.py"),
|
|
'import os, sys\n'
|
|
'sys.path.insert(0, os.path.join(os.environ["HOME"], "wiz"))\n'
|
|
'sys.path.append("lib")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("11 computed/relative bindings are not statically graded -> rc 0",
|
|
rc, 0)
|
|
|
|
# 12. THE PROBE EARNS ITS KEEP: a computed host path is invisible to
|
|
# the static scan and named by --import-probe, as a real resolution.
|
|
host = os.path.join(td, "c12-host") # outside the repo
|
|
_write(os.path.join(host, "hosttarget.py"), "thing = 7\n")
|
|
repo = os.path.join(td, "c12")
|
|
_write(os.path.join(repo, "test_probe.py"),
|
|
'import os, sys\n'
|
|
'HOST = %r\n'
|
|
'sys.path.insert(0, HOST)\n'
|
|
'from hosttarget import thing\n' % host)
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc_static, obj_static = call(repo, "WORKTREE", False, False,
|
|
quiet=True)
|
|
check("12a a computed host path is invisible to the static scan",
|
|
rc_static, 0)
|
|
rc_probe, obj_probe = call(repo, "WORKTREE", False, True, quiet=True)
|
|
check("12b and --import-probe names it -> rc 1", rc_probe, 1)
|
|
check("12c the probe row names the host module it bound",
|
|
any("hosttarget" in m[0] or host in m[1]
|
|
for m in obj_probe["probe"][0]["outside_modules"]), True)
|
|
|
|
# 13. and the probe is silent on a correct binding
|
|
repo = os.path.join(td, "c13")
|
|
_write(os.path.join(repo, "test_ok.py"), clean)
|
|
_write(os.path.join(repo, "target.py"), "thing = 1\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, True, quiet=True)
|
|
check("13 the probe reads a correct binding clean -> rc 0", rc, 0)
|
|
|
|
# 14. --rev HEAD grades the COMMITTED blob, not the worktree
|
|
repo = os.path.join(td, "c14")
|
|
_write(os.path.join(repo, "test_ok.py"), clean)
|
|
_write(os.path.join(repo, "target.py"), "thing = 1\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
subprocess.run(["git", "-C", repo, "-c", "user.name=t",
|
|
"-c", "user.email=t@invalid", "-c",
|
|
"commit.gpgsign=false", "commit", "-q", "-m", "c"],
|
|
check=True, capture_output=True)
|
|
_write(os.path.join(repo, "test_bad.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/x")\n')
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc_wt, obj_wt = call(repo, "WORKTREE", False, False, quiet=True)
|
|
rc_head, obj_head = call(repo, "HEAD", False, False, quiet=True)
|
|
check("14a WORKTREE grades the tick's own edit -> rc 1", rc_wt, 1)
|
|
check("14b --rev HEAD grades the committed blob -> rc 0", rc_head, 0)
|
|
|
|
# 15. an untracked test file is not graded (tracked-only inventory)
|
|
repo = os.path.join(td, "c15")
|
|
_write(os.path.join(repo, "test_ok.py"), clean)
|
|
_write(os.path.join(repo, "target.py"), "thing = 1\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
_write(os.path.join(repo, "test_scratch.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/z")\n')
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("15 an UNTRACKED test file is not graded -> rc 0", rc, 0)
|
|
|
|
# 16. a directory that is not a work tree is a refusal at rc 2
|
|
notrepo = os.path.join(td, "c16")
|
|
_write(os.path.join(notrepo, "README.txt"), "not a repo\n")
|
|
rc, _ = call(notrepo, "WORKTREE", False, False, quiet=True)
|
|
check("16 a directory that is not a work tree -> rc 2", rc, 2)
|
|
|
|
# 17. a tracked test file that does not parse is UNGRADABLE, not clean
|
|
repo = os.path.join(td, "c17")
|
|
_write(os.path.join(repo, "test_broken.py"), "def oops(:\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("17 an unparsable test file -> rc 2", rc, 2)
|
|
check("17b and it is named as ungradable", obj["ungraded"][0][0],
|
|
"test_broken.py")
|
|
|
|
# 18. --json is a real record and agrees with the rc it returned
|
|
repo = os.path.join(td, "c18")
|
|
_write(os.path.join(repo, "test_bad.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/q")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
jpath = os.path.join(td, "c18.json")
|
|
rc = call_main(["--repo", repo, "--json", jpath, "--quiet"])
|
|
obj = json.load(open(jpath)) if os.path.exists(jpath) else {}
|
|
check("18 --json writes the faults", len(obj.get("faults", [])), 1)
|
|
check("18b and the record's graded count is the run's", obj.get("graded"), 1)
|
|
rc2 = call_main(["--repo", repo, "--json", os.path.join(td, "no", "x.json"),
|
|
"--quiet"])
|
|
check("18c an unwritable --json target -> rc 2", rc2, 2)
|
|
|
|
# 19. Q100: an UNDERSCORE-prefixed suite is a test file. This is the
|
|
# measured shape: both forge `_test_roundtrip*.py` files bound
|
|
# /root/.hermes/wizard-outputs/forge and read 34/34 while the repo copy
|
|
# under them was poisoned. Under the first selector they were invisible.
|
|
repo = os.path.join(td, "c19")
|
|
_write(os.path.join(repo, "_test_roundtrip.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/forge")\n'
|
|
'import chunker\n')
|
|
_write(os.path.join(repo, "chunker.py"), "def chunk_text(t, **k):\n return []\n")
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("19 an _test_*.py host binding -> rc 1", rc, 1)
|
|
check("19b and it is graded, not advised",
|
|
(obj["graded"], obj["faults"][0][0], obj["faults"][0][1]),
|
|
(1, "_test_roundtrip.py", 2))
|
|
|
|
# 20. and the suffix spelling binds the same way
|
|
repo = os.path.join(td, "c20")
|
|
_write(os.path.join(repo, "roundtrip_test.py"),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/forge")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("20 a *_test.py host binding -> rc 1", rc, 1)
|
|
check("20b named as a fault row", obj["faults"][0][0],
|
|
"roundtrip_test.py")
|
|
|
|
# 21. NEGATIVE CONTROL: a name that merely CONTAINS `test` is not a
|
|
# test file. Widening the class must not sweep in latest.py/protest.py
|
|
# and turn the advisory count into a gate nobody can run.
|
|
repo = os.path.join(td, "c21")
|
|
for name in ("latest.py", "attestation.py", "protest.py"):
|
|
_write(os.path.join(repo, name),
|
|
'import sys\nsys.path.insert(0, "/root/.hermes/wizard-outputs/x")\n')
|
|
_git_init(repo)
|
|
subprocess.run(["git", "-C", repo, "add", "-A"], check=True,
|
|
capture_output=True)
|
|
rc, obj = call(repo, "WORKTREE", False, False, quiet=True)
|
|
check("21 a name containing `test` is not a test file -> rc 0", rc, 0)
|
|
check("21b it stays an advisory row, ungraded",
|
|
(obj["graded"], obj["advisory"]), (0, 3))
|
|
|
|
print("selftest: %d cells, %d passed, %d failed"
|
|
% (cells, passed, len(failures)))
|
|
for f in failures:
|
|
print(" FAIL %s" % f)
|
|
return 0 if passed == cells else 1
|
|
|
|
|
|
# ------------------------------------------------------------------------ main
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(
|
|
description="Grade every committed test file (test_*.py, _test_*.py, "
|
|
"*_test.py) on where it binds the module it imports: the "
|
|
"repo, or a path on the host.")
|
|
ap.add_argument("--repo", default=DEFAULT_REPO,
|
|
help="git work tree root (default: the checkout holding "
|
|
"this tool)")
|
|
ap.add_argument("--rev", default="WORKTREE", metavar="REV",
|
|
help="WORKTREE (the index, default) or a committed "
|
|
"revision such as HEAD")
|
|
ap.add_argument("--all-files", action="store_true",
|
|
help="grade every tracked *.py, not only test files -- the "
|
|
"non-test class is advisory by default")
|
|
ap.add_argument("--import-probe", action="store_true",
|
|
help="also load each test file in a clean subprocess and "
|
|
"report what it actually bound (catches computed paths)")
|
|
ap.add_argument("--json", metavar="PATH", help="write the rows here")
|
|
ap.add_argument("--quiet", action="store_true")
|
|
ap.add_argument("--selftest", action="store_true",
|
|
help="run the fixture corpus (the run prints its cell count)")
|
|
a = ap.parse_args(argv)
|
|
|
|
if a.selftest:
|
|
return selftest()
|
|
try:
|
|
rc, obj = run(a.repo, a.rev, a.all_files, a.import_probe, a.quiet)
|
|
except CannotGrade as exc:
|
|
print("cannot grade: %s" % exc, file=sys.stderr)
|
|
return 2
|
|
if a.json:
|
|
try:
|
|
with open(a.json, "w") as fh:
|
|
json.dump(obj, fh, indent=1, default=str)
|
|
except OSError as exc:
|
|
print("cannot write --json %s: %s" % (a.json, exc), file=sys.stderr)
|
|
return 2
|
|
return rc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|