Files
troupe-standup/tools/tool_map.py
T

521 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""tool_map.py -- the instruments get a map, and the map is generated from the tree.
WHY THIS EXISTS
`tools/` holds sixteen scripts and there is no document that says what they
are. A reader meets them one at a time -- in a `verify.py` leg, in a queue
block, in a reply -- and nothing states the whole set: which tool grades what,
which flags each one actually accepts, which of them the two probe axes cover,
and which leg of the operator's gate a failure shows up in. The closest thing
is `verify.py`'s `LEGS` table, which names five of the sixteen.
A hand-written map would be a fourth notation of the defect this suite keeps
grading: prose about the tree, one step from the thing it describes, with no
reader. So this document is not written, it is DERIVED -- every cell comes
from the artefact it describes, at a stated revision:
what it grades the module docstring's first line, after `<name> -- `
flags the tool's own `--help`, run from a temporary copy of the
blob at `--rev` (the only statement of an interface that
cannot drift from the interface)
probes `selftest_sweep.py`'s READ_PROBES / NOT_UTF8_PROBES /
MALFORMED_PROBES tables, read with `ast.literal_eval`
gate leg `verify.py`'s `LEGS` table, likewise
Nothing is inferred and nothing is typed by hand, so the document cannot say
something the tree does not: `--check` re-derives it and compares bytes.
WHAT IT MEASURES -- one verdict per invocation
CURRENT the document on disk is byte-identical to the render
DRIFT the document differs; the first differing line is printed. A
FAULT: the map is a claim about the tree and it is no longer true.
UNREADABLE the document does not exist, or the revision will not
materialise, or a tool's `--help` outlived its timeout. rc 2,
never rc 1 -- an ungradable map is not a stale one.
`--selftest` ends with the currency cell: it re-renders `docs/TOOLS.md` at
HEAD and compares. That cell is the reason the map stays true -- the tool
sweep runs every tool's `--selftest`, so a commit that changes a tool's
docstring, flags or probe rows without re-rendering the map turns the sweep
RED instead of leaving a stale document that reads as current.
READ-ONLY except for the one path it is told to write. The only process it
starts is `<tool> --help`, which does no work by construction; it never runs a
documented command, so it cannot write, post, or deploy anything.
EXIT CODES (the tools/ contract)
0 the map is current (or was written)
1 the map has drifted from the tree
2 could not grade: not a git work tree, a revision that will not
materialise, or a document that cannot be read
USAGE
python3 tool_map.py --repo . --write docs/TOOLS.md # render
python3 tool_map.py --repo . --check docs/TOOLS.md # grade drift, rc 1 on drift
python3 tool_map.py --repo . --json out.json
python3 tool_map.py --selftest # 16 cells
"""
from __future__ import annotations
import argparse
import ast
import json
import os
import re
import subprocess
import sys
import tempfile
DEFAULT_REPO = "/root/troupe-standup"
DEFAULT_DOC = "docs/TOOLS.md"
HELP_TIMEOUT = 30
# A long option, or a short one, as argparse prints them in the usage block.
# `-h`/`--help` are dropped: every tool has them, and a column of sixteen
# identical pairs is noise, not information.
OPT_RE = re.compile(r"(?<![\w-])(--[a-z0-9][a-z0-9-]*|-[A-Za-z])(?![\w-])")
SKIP_OPTS = {"-h", "--help"}
def git(repo, *args, inp=None):
r = subprocess.run(["git", "-C", repo, *args], input=inp,
capture_output=True, text=True)
return r.returncode, r.stdout, r.stderr
def read_blob(repo, rev, path):
rc, out, err = git(repo, "show", f"{rev}:{path}")
return out if rc == 0 else None
def tracked_tools(repo, rev):
"""Sorted `tools/*.py` basenames tracked at `rev`."""
rc, out, _ = git(repo, "ls-tree", "-r", "--name-only", rev, "--", "tools/")
if rc != 0:
return None
names = [p.split("/")[-1] for p in out.split("\n")
if p.startswith("tools/") and p.endswith(".py")]
return sorted(names)
def literal_assignments(repo, rev, path, names):
"""{name: value} for the module-level assignments whose value is a literal."""
src = read_blob(repo, rev, path)
if src is None:
return {}
try:
tree = ast.parse(src)
except SyntaxError:
return {}
found = {}
for node in tree.body:
if not isinstance(node, ast.Assign) or len(node.targets) != 1:
continue
tgt = node.targets[0]
if not isinstance(tgt, ast.Name) or tgt.id not in names:
continue
try:
found[tgt.id] = ast.literal_eval(node.value)
except (ValueError, SyntaxError):
continue
return found
def usage_block(text):
"""Only the `usage:` paragraph of a `--help` page.
argparse prints the module docstring as the description ABOVE the usage
line, and those docstrings discuss flags in prose -- `citation_hash_class.py`
mentions `--batch-all-objects` while explaining why it does not use it. A
flag that prose mentions is not a flag the tool accepts, so the oracle is
cut to the usage paragraph and stop: it ends at the first blank line after
the line that opens with `usage:`. No usage line means no oracle.
"""
lines = text.splitlines()
start = None
for i, line in enumerate(lines):
if line.lower().startswith("usage:"):
start = i
break
if start is None:
return ""
block = []
for line in lines[start:]:
if not line.strip():
break
block.append(line)
return "\n".join(block)
def purpose_of(src):
"""The module docstring's first sentence as a one-clause summary.
The house shape is `<name> -- <what it grades>.`; documents written without
the separator fall back to the docstring's opening sentence. The whole first
PARAGRAPH is flattened before the sentence cut, because the sentence that
states a tool's purpose routinely wraps across two or three source lines --
a line-local cut is what this suite grades as a defect everywhere else.
Returns "" when there is no module docstring, which the row prints as `-`.
"""
if not src:
return ""
try:
doc = ast.get_docstring(ast.parse(src)) or ""
except SyntaxError:
return ""
first_para = re.split(r"\n\s*\n", doc.strip())[0] if doc.strip() else ""
flat = " ".join(line.strip() for line in first_para.splitlines()).strip()
if not flat:
return ""
flat = re.sub(r"^[a-z0-9_]+\.[a-z]+ *(?:--|—|–|-) *", "", flat).strip()
m = re.search(r"^(.+?[.?!])(?:\s|$)", flat)
out = (m.group(1) if m else flat).rstrip()
if len(out) > 180:
out = out[:177].rstrip() + "..."
return out
def flags_of(repo, rev, name):
"""The tool's accepted options, from its own `--help`.
The blob is written to a temporary directory and run there so a tool is
adjudicated at `--rev` against the parser it will actually use, never
against a worktree copy. Returns (flags, note) where note is "" when the
oracle answered and a short reason when it did not.
"""
src = read_blob(repo, rev, f"tools/{name}")
if src is None:
return [], "no such file at rev"
with tempfile.TemporaryDirectory() as td:
path = os.path.join(td, name)
with open(path, "w", encoding="utf-8") as fh:
fh.write(src)
try:
proc = subprocess.run([sys.executable, path, "--help"],
capture_output=True, text=True,
timeout=HELP_TIMEOUT, cwd=td)
except subprocess.TimeoutExpired:
return [], "help timed out"
text = (proc.stdout or "") + (proc.stderr or "")
usage = usage_block(text)
seen, out = set(), []
for m in OPT_RE.finditer(usage):
opt = m.group(1)
if opt in SKIP_OPTS or opt in seen:
continue
seen.add(opt)
out.append(opt)
if not out:
return [], "no --help oracle"
return out, ""
def _count_by_tool(entries):
"""Count probes per tool name (first element of each entry tuple).
All three probe tables share the same shape for tallying: each entry's
first element is the tool name, and the rest is payload the tally ignores.
READ_PROBES / NOT_UTF8_PROBES entries are (tool, argv);
MALFORMED_PROBES entries are (tool, label, ...). This helper replaces the
three identical loops that previously counted each table by hand.
"""
counts = {}
for entry in entries:
name = entry[0]
counts[name] = counts.get(name, 0) + 1
return counts
def collect(repo, rev):
"""The whole map as data: one dict per tool plus the gate's legs."""
names = tracked_tools(repo, rev)
if names is None:
raise SystemExit(f"tool_map: cannot read repo: {repo} at {rev}")
sweep = literal_assignments(repo, rev, "tools/selftest_sweep.py",
{"READ_PROBES", "NOT_UTF8_PROBES",
"MALFORMED_PROBES"})
read_n = _count_by_tool(sweep.get("READ_PROBES", ()))
utf8_n = _count_by_tool(sweep.get("NOT_UTF8_PROBES", ()))
mal_n = _count_by_tool(sweep.get("MALFORMED_PROBES", ()))
legs = literal_assignments(repo, rev, "tools/verify.py", {"LEGS"}).get("LEGS", ())
leg_of = {}
for leg in legs:
if isinstance(leg, dict) and "tool" in leg and "key" in leg:
leg_of.setdefault(leg["tool"], []).append(leg["key"])
rows = []
for name in names:
src = read_blob(repo, rev, f"tools/{name}")
flags, note = flags_of(repo, rev, name)
rows.append({
"tool": name,
"purpose": purpose_of(src),
"flags": flags,
"no_oracle": note,
"read": read_n.get(name, 0),
"not_utf8": utf8_n.get(name, 0),
"malformed": mal_n.get(name, 0),
"legs": leg_of.get(name, []),
})
return {"rev": rev, "rows": rows,
"legs": [{"key": l.get("key"), "tool": l.get("tool"),
"slow": bool(l.get("slow")), "what": l.get("what", "")}
for l in legs if isinstance(l, dict)]}
def render(data):
"""The document, byte-deterministic for a given tree."""
rows, legs = data["rows"], data["legs"]
with_oracle = sum(1 for r in rows if not r["no_oracle"])
lines = [
"# The instruments",
"",
"Generated by `tools/tool_map.py` from the tree it is asked to read. Do not",
"edit by hand: `python3 tools/tool_map.py --repo . --write docs/TOOLS.md`",
"re-renders it, and `--check` grades it (rc 1 when the document and the tree",
"disagree). Every cell below is derived, never typed -- purpose from the",
"module docstring, flags from the tool's own `--help`, probe rows from",
"`selftest_sweep.py`'s tables, the gate leg from `verify.py`'s `LEGS`.",
"",
f"**{len(rows)} tools, {with_oracle} with a `--help` oracle.** "
f"Probes are read as `read-refusal / not-utf8 / malformed-input` rows; a "
f"`-` means that axis does not name the tool.",
"",
"| tool | what it grades | flags | probes | gate leg |",
"|---|---|---|---|---|",
]
for r in rows:
purpose = r["purpose"] or "-"
if r["no_oracle"]:
flags = f"- *({r['no_oracle']})*"
else:
flags = " ".join(f"`{f}`" for f in r["flags"]) or "-"
probes = (f"{r['read']} / {r['not_utf8']} / {r['malformed']}"
if (r["read"] or r["not_utf8"] or r["malformed"]) else "-")
gate = ", ".join(f"`{k}`" for k in r["legs"]) or "-"
lines.append(f"| `{r['tool']}` | {purpose} | {flags} | {probes} | {gate} |")
lines += [
"",
"## The gate",
"",
"`python3 tools/verify.py --repo .` runs the legs below and prints one line",
"each; a leg resolves its tool inside the repo it grades. The `slow` legs are",
"dropped by `--fast`.",
"",
"| leg | tool | slow | what it grades |",
"|---|---|---|---|",
]
for leg in legs:
lines.append(f"| `{leg['key']}` | `{leg['tool']}` | "
f"{'yes' if leg['slow'] else 'no'} | {leg['what']} |")
lines += [
"",
"A tool that appears in no leg is still graded by `selftest-sweep`, which runs",
"every tracked tool's `--selftest` and both read-refusal axes -- so the",
"`probes` column above is also the sweep's cover of that tool.",
"",
]
return "\n".join(lines)
def selftest(repo):
cells = []
def cell(label, ok, detail=""):
cells.append({"label": label, "ok": bool(ok), "detail": detail})
# -- the derivation, on fixtures -------------------------------------
cell("purpose: house `<name> -- ...` shape is stripped",
purpose_of('"""fake_tool.py -- grades the fake contract, at one line.\n\n'
'More prose here."""\n') ==
"grades the fake contract, at one line.")
cell("purpose: no separator falls back to the first sentence",
purpose_of('"""A map of the tree. Second sentence follows."""\n') ==
"A map of the tree.")
cell("purpose: absent docstring is empty, not invented",
purpose_of("import os\n") == "")
cell("flags: `-h`/`--help` are dropped, real options kept",
[o for o in OPT_RE.findall("usage: t.py [-h] [--repo R] [--rev V]")
if o not in SKIP_OPTS] == ["--repo", "--rev"])
cell("flags: a positional-looking token is not an option",
OPT_RE.findall("usage: t.py [-h] [--repo REPO] filename") ==
["-h", "--repo"])
cell("flags: a flag the description only MENTIONS is not an oracle",
usage_block("desc mentions --nope\n\noptions:\n") == "")
with tempfile.TemporaryDirectory() as td:
fixture = os.path.join(td, "fx")
os.makedirs(os.path.join(fixture, "tools"))
os.makedirs(os.path.join(fixture, "docs"))
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", fixture], check=True, env=env)
open(os.path.join(fixture, "tools", "fake_tool.py"), "w").write(
'#!/usr/bin/env python3\n'
'"""fake_tool.py -- grades the fake contract, which nothing measured.\n\n'
'Prose."""\n'
'import argparse\n'
'ap = argparse.ArgumentParser()\n'
'ap.add_argument("--repo")\n'
'ap.add_argument("--fail", action="store_true")\n'
'if "--help" in __import__("sys").argv:\n'
' ap.print_help()\n')
open(os.path.join(fixture, "tools", "quiet_tool.py"), "w").write(
'#!/usr/bin/env python3\n'
'"""quiet_tool.py -- parses argv by hand, so it prints no usage."""\n'
'import sys\n'
'print("ran")\n')
open(os.path.join(fixture, "tools", "selftest_sweep.py"), "w").write(
'READ_PROBES = (("fake_tool.py", ["--repo", "{nope}"]),)\n'
'NOT_UTF8_PROBES = ()\n'
'MALFORMED_PROBES = (("fake_tool.py", "not-a-repo",\n'
' ["--repo", "{notrepo}"], 2, "cannot read"),)\n')
open(os.path.join(fixture, "tools", "verify.py"), "w").write(
'LEGS = ({"key": "fake", "tool": "fake_tool.py", "slow": False,\n'
' "what": "grade the fake contract"},)\n')
def commit(msg):
subprocess.run(["git", "-C", fixture, "add", "-A"], check=True, env=env)
subprocess.run(["git", "-C", fixture, "commit", "-qm", msg],
check=True, env=env)
commit("fixture")
data = collect(fixture, "HEAD")
by = {r["tool"]: r for r in data["rows"]}
cell("collect: every tracked .py is a row, oracle or not",
len(data["rows"]) == 4 and by["fake_tool.py"]["no_oracle"] == ""
and by["quiet_tool.py"]["no_oracle"] == "no --help oracle",
str([(r["tool"], r["no_oracle"]) for r in data["rows"]]))
cell("collect: probe rows and the gate leg are joined by name",
by["fake_tool.py"]["read"] == 1
and by["fake_tool.py"]["malformed"] == 1
and by["fake_tool.py"]["legs"] == ["fake"],
str({k: (v["read"], v["malformed"], v["legs"]) for k, v in by.items()}))
cell("collect: only printed flags are listed, and --help is not",
by["fake_tool.py"]["flags"] == ["--repo", "--fail"],
str(by["fake_tool.py"]["flags"]))
doc = os.path.join(fixture, "docs", "TOOLS.md")
text = render(data)
open(doc, "w").write(text)
cell("render: deterministic (two renders of one tree are identical)",
text == render(collect(fixture, "HEAD")))
cell("render: the tool without an oracle is marked, not accused",
"quiet_tool.py" in text and "no --help oracle" in text)
rc_ok = compare(doc, text)
cell("--check: rc 0 on the document the same tree renders", rc_ok == 0,
f"rc {rc_ok}")
open(doc, "a").write("stale line\n")
rc_drift = compare(doc, render(collect(fixture, "HEAD")), quiet=True)
cell("--check: rc 1 once a byte drifts",
rc_drift == 1, f"rc {rc_drift}")
open(doc, "w").write(text)
rc_missing = compare(os.path.join(fixture, "docs", "nope.md"), text,
quiet=True)
cell("--check: rc 2, not rc 1, when the document does not exist",
rc_missing == 2, f"rc {rc_missing}")
rc_norepo = graderc(td)
cell("--repo a directory that is not a git work tree: rc 2",
rc_norepo == 2, f"rc {rc_norepo}")
# -- the currency cell: the reason this document stays true ----------
here = os.path.join(repo, "docs", "TOOLS.md")
try:
want = render(collect(repo, "HEAD"))
rc = compare(here, want, quiet=True)
except SystemExit as exc:
rc = 2
cell("HEAD: the shipped map renders", False, str(exc))
cell("HEAD: docs/TOOLS.md is the render of this tree (the drift guard)",
rc == 0, f"rc {rc}")
passed = sum(1 for c in cells if c["ok"])
for c in cells:
print(f" {'ok ' if c['ok'] else 'FAIL'} {c['label']}"
+ (f" [{c['detail']}]" if c["detail"] and not c["ok"] else ""))
print(f"selftest: {len(cells)} cells, {passed} passed, {len(cells) - passed} failed")
return 0 if passed == len(cells) else 1
def graderc(td):
"""rc of a full `--check` run against a directory that is not a repo."""
proc = subprocess.run([sys.executable, os.path.abspath(__file__),
"--repo", td, "--check", os.path.join(td, "x.md")],
capture_output=True, text=True)
return proc.returncode
def compare(path, want, quiet=False):
"""0 current, 1 drift (first differing line printed), 2 unreadable."""
try:
with open(path, encoding="utf-8") as fh:
have = fh.read()
except OSError as exc:
if not quiet:
print(f"tool_map: cannot read document: {exc}")
return 2
if have == want:
return 0
if not quiet:
a, b = have.splitlines(), want.splitlines()
for i in range(max(len(a), len(b))):
got = a[i] if i < len(a) else "<end of document>"
exp = b[i] if i < len(b) else "<end of document>"
if got != exp:
print(f"DRIFT at line {i + 1}:")
print(f" document: {got}")
print(f" tree: {exp}")
break
return 1
def main():
ap = argparse.ArgumentParser(
description="Render the map of tools/ from the tree, and grade its drift.")
ap.add_argument("--repo", default=DEFAULT_REPO)
ap.add_argument("--rev", default="HEAD")
ap.add_argument("--write", metavar="PATH",
help=f"render the map to PATH (default target {DEFAULT_DOC})")
ap.add_argument("--check", metavar="PATH",
help="grade PATH against a fresh render; rc 1 on drift")
ap.add_argument("--json", metavar="PATH", help="write the collected data as JSON")
ap.add_argument("--selftest", action="store_true")
args = ap.parse_args()
if args.selftest:
return selftest(args.repo)
rc, _, _ = git(args.repo, "rev-parse", "--show-toplevel")
if rc != 0:
print(f"tool_map: cannot read repo: {args.repo} is not a git work tree")
return 2
rc, _, err = git(args.repo, "rev-parse", "--verify", f"{args.rev}^{{commit}}")
if rc != 0:
print(f"tool_map: cannot read rev {args.rev}: {err.strip().splitlines()[0] if err.strip() else 'no such object'}")
return 2
data = collect(args.repo, args.rev)
text = render(data)
if args.json:
with open(args.json, "w", encoding="utf-8") as fh:
json.dump(data, fh, indent=2, sort_keys=True)
fh.write("\n")
print(f"tool_map: wrote {args.json}")
if args.write:
with open(args.write, "w", encoding="utf-8") as fh:
fh.write(text)
print(f"tool_map: wrote {args.write} ({len(text)} B, "
f"{len(data['rows'])} tools, {len(data['legs'])} gate legs)")
return 0
if args.check:
return compare(args.check, text)
print(text, end="")
return 0
if __name__ == "__main__":
sys.exit(main())