volt
2ccf8a6b6b
The quill draft-compliance census graded a snapshot against a moving tree
and shipped two implementations of the same gate; the lane copy scanned
/root/troupe-standup by absolute path and appended a row from a host path.
Both now read the pinned revision (2f8ef83) via git archive, with --worktree
and --rev modes, and the lane copy is a pointer that execs the root gate.
Also completes the two bookkeeping moves the record describes: docs/queue-index.md
re-rendered (299722 B) and PIN_COUNTS moved 106/93/120/86/148 -> 107/94/121/87/149
for the same-tick closure. verify.py --fast: 12 legs, 11 pass, 0 fault, 1 skipped.
580 lines
22 KiB
Python
580 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier for draft-compliance-report-2026-10-31.{tsv,md}.
|
|
|
|
WHAT IT GRADES, AND AGAINST WHAT
|
|
The committed TSV is a claim about the .md drafts beside it. This verifier
|
|
re-derives every row from the files with its OWN code -- its own line counter,
|
|
its own `VERDICT:`/`CHECK:` detection and its own implementation of the bare-
|
|
command rule -- and compares in both directions: every TSV row against the
|
|
file it names, and every file against the row it should have. It also grades
|
|
the report's Evidence table against the counts it just measured, so a stale
|
|
numeral in the prose is a fault rather than a reader's problem.
|
|
|
|
WHICH TREE -- three modes, because a dated census describes a revision
|
|
(default) the lane as it stands at the pin in
|
|
`draft-compliance-report-2026-10-31.pin`, extracted with
|
|
`git archive`; the committed TSV and report are graded against
|
|
it. Drift since the pin is printed as a note, never a fault.
|
|
--worktree the live worktree; this is the "is the census current?" reading
|
|
and it faults on any difference, additions included.
|
|
--rev REV the lane at REV, for probing whether a given tree's census is
|
|
true and for grading an older landing.
|
|
|
|
Why: the 2026-10-31 table was byte-true at 2f8ef83 and read red at HEAD for
|
|
eight drafts that landed after it -- a snapshot cannot be graded against a
|
|
tree it never described. And the fix must not be "grade nothing": the pin is
|
|
what makes the claim falsifiable (Q154).
|
|
|
|
Paths come from THIS FILE's location. The 2026-10-31 version hardcoded an
|
|
absolute home-directory path in both the scanner and the verifier, so a
|
|
clone's CHECK graded the host's tree and a mutated copy of the committed TSV
|
|
still read PASS; `--selftest` cell `no-host-path` greps the sources for that
|
|
class, because this file's own fix must not reintroduce it.
|
|
|
|
EXIT CODES -- the archive's convention
|
|
0 the TSV and the report re-derive from the tree they describe
|
|
1 a fault: a row, a file-set entry or a reported count disagrees, or the
|
|
committed TSV is not the bytes the pin names
|
|
2 could not grade: the report, the TSV or the pin is missing/unreadable, the
|
|
pin names no rev, the rev cannot be extracted, or the repository root
|
|
cannot be found. Never a pass.
|
|
|
|
--selftest runs the cells in a temp tree: one green control, eight mutants and
|
|
four pin-contract cells, so a green reading is never the only thing this
|
|
checker can print.
|
|
|
|
NOTE ON THE SECOND IMPLEMENTATION OF `is_bare`
|
|
It reads operators on the raw string and only checks quoting for parity, so an
|
|
operator inside a quoted argument reads as not bare where the shlex-based rule
|
|
would read it as bare. That is deliberate and conservative (it over-reports
|
|
"not bare", never over-reports "bare"), and `rule-agreement` fails loudly if
|
|
the two implementations ever disagree on a command this corpus actually holds.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
|
|
import draft_compliance_rule as rule # noqa: E402
|
|
|
|
# The path inside the repository that holds this lane. Pinned grading archives
|
|
# exactly this directory at the pinned rev; nothing outside it is read.
|
|
PATHSPEC = "wizard-work/quill"
|
|
|
|
# The report's Evidence table, label -> the metric it claims. The labels are
|
|
# matched verbatim; a missing label is a fault (the report must say what it
|
|
# measured).
|
|
REPORT_LABELS = (
|
|
("Total `.md` drafts scanned", "total"),
|
|
("Drafts with `VERDICT:`", "verdict"),
|
|
("Drafts with `CHECK:`", "check"),
|
|
("Drafts with a bare `CHECK:` command", "bare"),
|
|
("Drafts failing at least one column", "failing"),
|
|
("`VERDICT:` and `CHECK:` but a non-bare command", "nonbare_both"),
|
|
)
|
|
|
|
NULL_RE = re.compile(r"(?:^|[\s;&|(])(true|false)(?=$|[\s;&|)])")
|
|
|
|
|
|
def own_is_bare(cmd):
|
|
"""The same contract as `rule.is_bare`, implemented differently.
|
|
|
|
Quoted regions are removed before the operator scan, so an operator inside a
|
|
quoted argument is an argument, not an operator; parity of quotes is checked
|
|
first, because a command that does not close its quotes does not tokenize
|
|
and is therefore not bare under the contract.
|
|
"""
|
|
if not cmd:
|
|
return False
|
|
if cmd.count("'") % 2 or cmd.count('"') % 2:
|
|
return False
|
|
stripped = re.sub(r"'[^']*'|\"[^\"]*\"", "''", cmd)
|
|
if ";" in stripped or "||" in stripped:
|
|
return False
|
|
rest = stripped.replace("&&", "")
|
|
if "&" in rest or "|" in rest:
|
|
return False
|
|
return NULL_RE.search(stripped) is None
|
|
|
|
|
|
def own_line_count(text):
|
|
"""Line count without splitlines(): newlines plus a final unterminated line."""
|
|
return text.count("\n") + (1 if text and not text.endswith("\n") else 0)
|
|
|
|
|
|
def own_rows(lane, exclude=rule.ARTIFACT_MD):
|
|
"""Re-derive the table from the files, without calling the generator."""
|
|
rows = []
|
|
names = sorted(n for n in os.listdir(lane)
|
|
if n.endswith(".md") and n != exclude)
|
|
for name in names:
|
|
path = os.path.join(lane, name)
|
|
if not os.path.isfile(path):
|
|
continue
|
|
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
|
text = f.read()
|
|
has_v = has_c = False
|
|
cmd = ""
|
|
for ln in text.split("\n"):
|
|
if ln.startswith("VERDICT:"):
|
|
has_v = True
|
|
elif ln.startswith("CHECK:"):
|
|
has_c = True
|
|
rest = ln[len("CHECK:"):].strip()
|
|
if rest and not cmd:
|
|
cmd = rest
|
|
rows.append((name, str(own_line_count(text)), str(has_v), str(has_c),
|
|
str(own_is_bare(cmd)), cmd))
|
|
return rows
|
|
|
|
|
|
def read_tsv(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
raw = f.read().split("\n")
|
|
comment = ""
|
|
if raw and raw[0].startswith("#"):
|
|
comment = raw[0]
|
|
raw = raw[1:]
|
|
body = [ln for ln in raw if ln.strip()]
|
|
header = body[0].split("\t") if body else []
|
|
rows = [tuple(ln.split("\t")) for ln in body[1:]]
|
|
return comment, header, rows
|
|
|
|
|
|
def read_report_counts(path):
|
|
counts = {}
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for ln in f:
|
|
m = re.match(r"^\|\s*(.+?)\s*\|\s*(\d+)\s*\|\s*$", ln.rstrip("\n"))
|
|
if m:
|
|
counts[m.group(1)] = int(m.group(2))
|
|
return counts
|
|
|
|
|
|
def metrics(rows):
|
|
def ok(r):
|
|
return r[2] == "True" and r[3] == "True" and r[4] == "True"
|
|
|
|
return {
|
|
"total": len(rows),
|
|
"verdict": sum(1 for r in rows if r[2] == "True"),
|
|
"check": sum(1 for r in rows if r[3] == "True"),
|
|
"bare": sum(1 for r in rows if r[4] == "True"),
|
|
"failing": sum(1 for r in rows if not ok(r)),
|
|
"nonbare_both": sum(1 for r in rows
|
|
if r[2] == "True" and r[3] == "True" and r[4] != "True"),
|
|
}
|
|
|
|
|
|
def check(lane=None, report=None, tsv=None):
|
|
"""Return (rc, faults, notes). Never raises for bad input."""
|
|
lane = lane or HERE
|
|
report = report or os.path.join(lane, rule.ARTIFACT_MD)
|
|
tsv = tsv or os.path.join(lane, rule.ARTIFACT_TSV)
|
|
faults, notes = [], []
|
|
|
|
if not os.path.isdir(lane):
|
|
return 2, ["lane directory missing: %s" % lane], notes
|
|
for path, what in ((report, "report"), (tsv, "TSV")):
|
|
if not os.path.isfile(path):
|
|
return 2, ["%s missing: %s" % (what, path)], notes
|
|
|
|
try:
|
|
comment, header, claimed = read_tsv(tsv)
|
|
reported = read_report_counts(report)
|
|
except (OSError, UnicodeDecodeError) as exc:
|
|
return 2, ["cannot read the artifact: %s" % exc], notes
|
|
|
|
if tuple(header) != rule.COLUMNS:
|
|
faults.append("TSV header is %r, contract is %r"
|
|
% (tuple(header), rule.COLUMNS))
|
|
|
|
derived = own_rows(lane)
|
|
der_map = {r[0]: r for r in derived}
|
|
claim_map = {r[0]: r for r in claimed}
|
|
|
|
for name in sorted(set(claim_map) - set(der_map)):
|
|
faults.append("TSV row names a file the lane does not hold: %s" % name)
|
|
for name in sorted(set(der_map) - set(claim_map)):
|
|
faults.append("lane file has no TSV row: %s" % name)
|
|
for name in sorted(set(der_map) & set(claim_map)):
|
|
if der_map[name] != claim_map[name]:
|
|
faults.append("row mismatch for %s: claimed=%r derived=%r"
|
|
% (name, claim_map[name], der_map[name]))
|
|
|
|
expected = "# measured %d markdown files" % len(derived)
|
|
if comment != expected:
|
|
faults.append("TSV comment is %r, measured %r" % (comment, expected))
|
|
|
|
m = metrics(derived)
|
|
for label, key in REPORT_LABELS:
|
|
if label not in reported:
|
|
faults.append("report table row missing: %r" % label)
|
|
elif reported[label] != m[key]:
|
|
faults.append("report says %r = %d, measured %d"
|
|
% (label, reported[label], m[key]))
|
|
|
|
# The contract has two implementations; they must agree on every command the
|
|
# corpus carries, or one of them is grading a different rule.
|
|
for r in derived:
|
|
if rule.is_bare(r[5]) != own_is_bare(r[5]):
|
|
faults.append("rule divergence on %s: rule=%s verifier=%s cmd=%r"
|
|
% (r[0], rule.is_bare(r[5]), own_is_bare(r[5]), r[5]))
|
|
|
|
notes.append("population %d, verdict %d, check %d, bare %d, failing %d"
|
|
% (m["total"], m["verdict"], m["check"], m["bare"], m["failing"]))
|
|
return (1 if faults else 0), faults, notes
|
|
|
|
|
|
# --- the pin: which revision this census describes ---------------------------
|
|
|
|
def repo_root(start):
|
|
"""Nearest ancestor that holds a `.git` entry, or None."""
|
|
path = os.path.abspath(start)
|
|
while True:
|
|
if os.path.exists(os.path.join(path, ".git")):
|
|
return path
|
|
parent = os.path.dirname(path)
|
|
if parent == path:
|
|
return None
|
|
path = parent
|
|
|
|
|
|
def materialize(rev, root, dest):
|
|
"""Extract PATHSPEC at `rev` under `dest`; return (lane_dir, error)."""
|
|
try:
|
|
src = subprocess.run(["git", "-C", root, "archive", rev, "--", PATHSPEC],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
except OSError as exc:
|
|
return None, "cannot run git: %s" % exc
|
|
if src.returncode != 0:
|
|
return None, "cannot archive %s: %s" % (
|
|
rev, src.stderr.decode("utf-8", "replace").strip()[:200])
|
|
lane = os.path.join(dest, "lane")
|
|
os.makedirs(lane)
|
|
try:
|
|
untar = subprocess.run(["tar", "-x", "-C", lane], input=src.stdout,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
except OSError as exc:
|
|
return None, "cannot run tar: %s" % exc
|
|
if untar.returncode != 0:
|
|
return None, "cannot extract %s: %s" % (
|
|
rev, untar.stderr.decode("utf-8", "replace").strip()[:200])
|
|
return os.path.join(lane, PATHSPEC), ""
|
|
|
|
|
|
def read_pin(path):
|
|
"""The pin as a dict; raises OSError when it cannot be read."""
|
|
values = {}
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
parts = line.split(None, 1)
|
|
if len(parts) == 2:
|
|
values[parts[0]] = parts[1].strip()
|
|
return values
|
|
|
|
|
|
def pin_verdict(pin, tsv):
|
|
"""(rc, faults) for the pin contract: 2 ungradable, 1 not the bytes, 0 ok.
|
|
|
|
The digest is what makes "the committed table" a fixed claim: without it a
|
|
silent edit to the TSV would be graded as the new truth.
|
|
"""
|
|
try:
|
|
values = read_pin(pin)
|
|
except OSError as exc:
|
|
return 2, ["pin missing or unreadable: %s" % exc]
|
|
if not values.get("rev"):
|
|
return 2, ["pin names no rev: %s" % pin]
|
|
try:
|
|
with open(tsv, "rb") as f:
|
|
digest = hashlib.sha256(f.read()).hexdigest()
|
|
except OSError as exc:
|
|
return 2, ["TSV unreadable: %s" % exc]
|
|
want = values.get("tsv")
|
|
if want and want != digest:
|
|
return 1, ["committed TSV is not the pinned bytes: sha256 %s, pin %s"
|
|
% (digest, want)]
|
|
return 0, []
|
|
|
|
|
|
def grade_at_rev(rev, claim_dir, lane=None):
|
|
"""Grade the artifact in `claim_dir` against the drafts at `rev`."""
|
|
root = repo_root(claim_dir)
|
|
if root is None:
|
|
return 2, ["cannot find the repository root above %s" % claim_dir], []
|
|
tmp = tempfile.mkdtemp(prefix="q154-rev-")
|
|
try:
|
|
tree, error = materialize(rev, root, tmp)
|
|
if tree is None:
|
|
return 2, [error], []
|
|
return check(tree, os.path.join(claim_dir, rule.ARTIFACT_MD),
|
|
os.path.join(claim_dir, rule.ARTIFACT_TSV))
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def drift_note(pinned_rows, live_rows):
|
|
"""What moved between the pinned census and the live tree. Advisory text."""
|
|
pin_map = {r[0]: r for r in pinned_rows}
|
|
live_map = {r[0]: r for r in live_rows}
|
|
added = sorted(set(live_map) - set(pin_map))
|
|
gone = sorted(set(pin_map) - set(live_map))
|
|
changed = sorted(n for n in set(pin_map) & set(live_map)
|
|
if pin_map[n] != live_map[n])
|
|
text = "drift since the pin: %d added, %d changed, %d removed" % (
|
|
len(added), len(changed), len(gone))
|
|
if added:
|
|
text += "; added: " + ", ".join(added[:6]) + (" ..." if len(added) > 6 else "")
|
|
return text
|
|
|
|
|
|
FIXTURE_FILES = {
|
|
"good.md": "# t\n\nVERDICT: ok\nCHECK: python3 x.py\n",
|
|
"plain.md": "# nothing\n",
|
|
"piped.md": "# t\n\nVERDICT: ok\nCHECK: python3 x.py | grep -q ok\n",
|
|
}
|
|
|
|
|
|
def _fixture_report(claimed):
|
|
lines = ["# fixture", "", "| Metric | Value |", "|---|---|"]
|
|
for label, _key in REPORT_LABELS:
|
|
lines.append("| %s | %d |" % (label, claimed[label]))
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _fixture_tree(root, mutate=None):
|
|
lane = os.path.join(root, "lane")
|
|
os.makedirs(lane)
|
|
for name, text in FIXTURE_FILES.items():
|
|
with open(os.path.join(lane, name), "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
rows = own_rows(lane)
|
|
m = metrics(rows)
|
|
claimed = {label: m[key] for label, key in REPORT_LABELS}
|
|
report = os.path.join(lane, rule.ARTIFACT_MD)
|
|
with open(report, "w", encoding="utf-8") as f:
|
|
f.write(_fixture_report(claimed))
|
|
tsv = os.path.join(lane, rule.ARTIFACT_TSV)
|
|
with open(tsv, "w", encoding="utf-8") as f:
|
|
f.write("# measured %d markdown files\n" % len(rows))
|
|
f.write("\t".join(rule.COLUMNS) + "\n")
|
|
for r in rows:
|
|
f.write("\t".join(r) + "\n")
|
|
if mutate:
|
|
mutate(lane, report, tsv)
|
|
return lane, report, tsv
|
|
|
|
|
|
def selftest(verbose=True):
|
|
cells, failed = [], []
|
|
|
|
def cell(name, want, mutate=None):
|
|
tmp = tempfile.mkdtemp(prefix="q142-")
|
|
try:
|
|
lane, report, tsv = _fixture_tree(tmp, mutate)
|
|
got, faults, _notes = check(lane, report, tsv)
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
cells.append(name)
|
|
if got != want:
|
|
failed.append((name, want, got, faults[:3]))
|
|
if verbose:
|
|
print("%-22s rc=%d want=%d %s"
|
|
% (name, got, want, "ok" if got == want else "FAIL"))
|
|
|
|
def bump_line(lane, report, tsv):
|
|
with open(os.path.join(lane, "good.md"), "a", encoding="utf-8") as f:
|
|
f.write("one more line\n")
|
|
|
|
def drop_verdict(lane, report, tsv):
|
|
path = os.path.join(lane, "good.md")
|
|
with open(path, encoding="utf-8") as f:
|
|
text = f.read()
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(text.replace("VERDICT: ok\n", ""))
|
|
|
|
def drop_row(lane, report, tsv):
|
|
with open(tsv, encoding="utf-8") as f:
|
|
lines = f.read().split("\n")
|
|
with open(tsv, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(ln for ln in lines if not ln.startswith("piped.md")))
|
|
|
|
def extra_file(lane, report, tsv):
|
|
with open(os.path.join(lane, "late.md"), "w", encoding="utf-8") as f:
|
|
f.write("# late\n")
|
|
|
|
def no_report(lane, report, tsv):
|
|
os.remove(report)
|
|
|
|
def no_tsv(lane, report, tsv):
|
|
os.remove(tsv)
|
|
|
|
def stale_count(lane, report, tsv):
|
|
with open(report, encoding="utf-8") as f:
|
|
text = f.read()
|
|
text = re.sub(r"(\| Total `\.md` drafts scanned \| )3( \|)",
|
|
r"\g<1>2\g<2>", text)
|
|
with open(report, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
|
|
def check_line_edited(lane, report, tsv):
|
|
with open(os.path.join(lane, "good.md"), "w", encoding="utf-8") as f:
|
|
f.write("# t\n\nVERDICT: ok\nCHECK: python3 y.py\n")
|
|
|
|
cell("green-control", 0)
|
|
cell("line-count-moved", 1, bump_line)
|
|
cell("verdict-removed", 1, drop_verdict)
|
|
cell("row-deleted", 1, drop_row)
|
|
cell("file-without-row", 1, extra_file)
|
|
cell("report-missing", 2, no_report)
|
|
cell("tsv-missing", 2, no_tsv)
|
|
cell("report-count-stale", 1, stale_count)
|
|
cell("check-line-edited", 1, check_line_edited)
|
|
|
|
# The pin contract: a snapshot is graded against the bytes and the rev it
|
|
# names, and an unreadable pin is ungradable rather than a pass.
|
|
good_tsv = "x\n"
|
|
|
|
def pin_cell(name, want, pin_text):
|
|
tmp = tempfile.mkdtemp(prefix="q154-pin-")
|
|
try:
|
|
tsv = os.path.join(tmp, "t.tsv")
|
|
with open(tsv, "w", encoding="utf-8") as f:
|
|
f.write(good_tsv)
|
|
pin = os.path.join(tmp, "p.pin")
|
|
if pin_text is not None:
|
|
with open(pin, "w", encoding="utf-8") as f:
|
|
f.write(pin_text)
|
|
got, faults = pin_verdict(pin, tsv)
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
cells.append(name)
|
|
if got != want:
|
|
failed.append((name, want, got, faults[:2]))
|
|
if verbose:
|
|
print("%-22s rc=%d want=%d %s"
|
|
% (name, got, want, "ok" if got == want else "FAIL"))
|
|
|
|
digest = hashlib.sha256(good_tsv.encode("utf-8")).hexdigest()
|
|
pin_cell("pin-ok", 0, "rev 0123456789abcdef\ntsv %s\n" % digest)
|
|
pin_cell("pin-tsv-digest-moved", 1, "rev 0123456789abcdef\ntsv %s\n" % ("0" * 64))
|
|
pin_cell("pin-no-rev", 2, "tsv %s\n" % digest)
|
|
pin_cell("pin-absent", 2, None)
|
|
|
|
# The two implementations of the rule, over the fixture cases and this
|
|
# directory's own corpus. A divergence here is what the 2026-10-31 copies had.
|
|
divergence = [cmd for cmd, want, _why in rule.SELFTEST_CASES
|
|
if own_is_bare(cmd) != want]
|
|
for r in own_rows(HERE):
|
|
if rule.is_bare(r[5]) != own_is_bare(r[5]):
|
|
divergence.append(r[0])
|
|
cells.append("rule-agreement")
|
|
if divergence:
|
|
failed.append(("rule-agreement", 0, 1, divergence[:3]))
|
|
if verbose:
|
|
print("%-22s %s" % ("rule-agreement",
|
|
"ok" if not divergence else "FAIL %r" % divergence[:3]))
|
|
|
|
# No host path in the sources: the 2026-10-31 defect, graded at the source.
|
|
offenders = []
|
|
needle = os.sep + "root" + os.sep
|
|
for name in (os.path.basename(__file__), "draft_compliance_report.py",
|
|
"draft_compliance_rule.py"):
|
|
with open(os.path.join(HERE, name), encoding="utf-8") as f:
|
|
if needle in f.read():
|
|
offenders.append(name)
|
|
cells.append("no-host-path")
|
|
if offenders:
|
|
failed.append(("no-host-path", 0, 1, offenders))
|
|
if verbose:
|
|
print("%-22s %s" % ("no-host-path",
|
|
"ok" if not offenders else "FAIL %r" % offenders))
|
|
|
|
print("\nselftest: %d cells, %d passed, %d failed"
|
|
% (len(cells), len(cells) - len(failed), len(failed)))
|
|
for name, want, got, detail in failed:
|
|
print(" FAIL %s: rc %d wanted %d -- %r" % (name, got, want, detail))
|
|
return 1 if failed else 0
|
|
|
|
|
|
def _emit(rc, faults, notes, mode):
|
|
for note in notes:
|
|
print(note)
|
|
if rc == 0:
|
|
print("VERDICT: PASS -- TSV and report re-derive from %s (%s)"
|
|
% (mode, notes[0] if notes else "no counts"))
|
|
elif rc == 2:
|
|
print("VERDICT: UNGRADABLE")
|
|
for f in faults:
|
|
print(" -", f)
|
|
else:
|
|
print("VERDICT: FAIL")
|
|
for f in faults:
|
|
print(" -", f)
|
|
return rc
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(
|
|
description="Verify the quill draft-compliance report.")
|
|
ap.add_argument("--lane", default=None,
|
|
help="grade this directory directly (fixtures; default: "
|
|
"the pinned revision)")
|
|
ap.add_argument("--report", default=None, help="report .md (default: in --lane)")
|
|
ap.add_argument("--tsv", default=None, help="report .tsv (default: in --lane)")
|
|
ap.add_argument("--pin", default=None,
|
|
help="pin file (default: %s beside the artifact)"
|
|
% rule.ARTIFACT_PIN)
|
|
ap.add_argument("--rev", default=None,
|
|
help="grade the committed artifact against the lane at REV")
|
|
ap.add_argument("--worktree", action="store_true",
|
|
help="grade against the live worktree instead of the pin")
|
|
ap.add_argument("--selftest", action="store_true", help="run the cells")
|
|
a = ap.parse_args(argv)
|
|
|
|
if a.selftest:
|
|
return selftest()
|
|
|
|
claim = a.lane or HERE
|
|
report = a.report or os.path.join(claim, rule.ARTIFACT_MD)
|
|
tsv = a.tsv or os.path.join(claim, rule.ARTIFACT_TSV)
|
|
pin = a.pin or os.path.join(claim, rule.ARTIFACT_PIN)
|
|
|
|
if a.worktree or a.lane:
|
|
rc, faults, notes = check(claim, report, tsv)
|
|
mode = "the live worktree" if a.worktree else "the tree at %s" % claim
|
|
elif a.rev:
|
|
rc, faults, notes = grade_at_rev(a.rev, claim)
|
|
mode = "the lane at %s" % a.rev
|
|
else:
|
|
rc_pin, faults_pin = pin_verdict(pin, tsv)
|
|
if rc_pin:
|
|
return _emit(rc_pin, faults_pin, [], "the pin")
|
|
rev = read_pin(pin)["rev"]
|
|
rc, faults, notes = grade_at_rev(rev, claim)
|
|
mode = "the pinned revision %s" % rev[:12]
|
|
|
|
# A snapshot is graded against the revision it names; what has landed since
|
|
# is drift to report, not a fault of the table. `--worktree` grades it.
|
|
if mode != "the live worktree":
|
|
try:
|
|
notes = list(notes) + [drift_note(read_tsv(tsv)[2], own_rows(HERE))]
|
|
except OSError:
|
|
pass
|
|
return _emit(rc, faults, notes, mode)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|