138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier for the work-queue triage artifact.
|
|
|
|
Recomputes the classification of every work-queue item straight from
|
|
WORK-QUEUE.md (the source of truth), then asserts that the committed
|
|
workqueue-triage.{json,md} match: id, title, status_text and the
|
|
rule-based OPEN/CLOSED flag for every record, plus that the markdown
|
|
report carries a row for every record.
|
|
|
|
Rule (identical to workqueue_triage.py):
|
|
OPEN = recorded status is empty OR the lowercased status text
|
|
contains one of: open, unclaimed, draft, pending
|
|
CLOSED = anything recorded as DONE / COMPLETE / CLOSED / RESOLVED etc.
|
|
|
|
Exit 0 on pass, 1 on any mismatch.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
VERIFIER_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
REPO_DIR = os.path.dirname(os.path.dirname(VERIFIER_DIR)) # /root/troupe-standup
|
|
SRC = os.path.join(REPO_DIR, "WORK-QUEUE.md")
|
|
JSON_PATH = os.path.join(VERIFIER_DIR, "workqueue-triage.json")
|
|
MD_PATH = os.path.join(VERIFIER_DIR, "workqueue-triage.md")
|
|
|
|
HEADER = re.compile(r"^## Q(\d+)\s+—\s+(.*)$")
|
|
STATUS_LINE = re.compile(r"^\s*\*\*?Status:\s*(.*)$")
|
|
OPEN_SIGNALS = ("open", "unclaimed", "draft", "pending")
|
|
|
|
|
|
def classify(status_text):
|
|
t = status_text.lower()
|
|
if not status_text.strip():
|
|
return True
|
|
return any(sig in t for sig in OPEN_SIGNALS)
|
|
|
|
|
|
def extract_status(lines, start):
|
|
for i in range(start, len(lines)):
|
|
m = STATUS_LINE.match(lines[i])
|
|
if m:
|
|
body = m.group(1).strip()
|
|
while len(body) >= 2 and ((body.startswith("**") and body.endswith("**"))
|
|
or (body.startswith("*") and body.endswith("*"))):
|
|
body = body[1:-1].strip()
|
|
return body
|
|
return ""
|
|
|
|
|
|
def recompute():
|
|
with open(SRC, encoding="utf-8") as fh:
|
|
lines = fh.read().split("\n")
|
|
records = []
|
|
for idx, line in enumerate(lines):
|
|
m = HEADER.match(line)
|
|
if not m:
|
|
continue
|
|
qid = m.group(1)
|
|
title = m.group(2).strip()
|
|
status = extract_status(lines, idx + 1)
|
|
records.append({"id": qid, "title": title,
|
|
"status_text": status, "open": classify(status)})
|
|
# collapse duplicate ids (Q37, Q38) to the last occurrence
|
|
dedup = {}
|
|
for rec in reversed(records):
|
|
dedup[rec["id"]] = rec
|
|
ordered = sorted(dedup.values(), key=lambda r: int(r["id"]))
|
|
return ordered
|
|
|
|
|
|
def main():
|
|
errors = []
|
|
try:
|
|
source_lines = open(SRC, encoding="utf-8").read().split("\n")
|
|
except OSError as exc:
|
|
print("VERIFY FAIL: cannot read source %s: %s" % (SRC, exc))
|
|
sys.exit(1)
|
|
|
|
expected = recompute()
|
|
|
|
if not os.path.exists(JSON_PATH):
|
|
print("VERIFY FAIL: missing %s" % JSON_PATH)
|
|
sys.exit(1)
|
|
with open(JSON_PATH, encoding="utf-8") as fh:
|
|
stored = json.load(fh)
|
|
stored_records = stored.get("records", [])
|
|
if len(stored_records) != len(expected):
|
|
errors.append("record count artifact=%d measured=%d"
|
|
% (len(stored_records), len(expected)))
|
|
|
|
for sr, er in zip(stored_records, expected):
|
|
for key in ("id", "title", "status_text", "open"):
|
|
if sr.get(key) != er[key]:
|
|
errors.append("%s.%s artifact=%r expected=%r"
|
|
% (sr.get("id"), key, sr.get(key), er[key]))
|
|
|
|
summary = stored.get("summary", {})
|
|
exp_summary = {
|
|
"total": len(expected),
|
|
"open": sum(1 for r in expected if r["open"]),
|
|
"closed": sum(1 for r in expected if not r["open"]),
|
|
"top_open_item": next((r["id"] for r in expected if r["open"]), None),
|
|
}
|
|
for key, val in exp_summary.items():
|
|
if summary.get(key) != val:
|
|
errors.append("summary.%s artifact=%r expected=%r"
|
|
% (key, summary.get(key), val))
|
|
|
|
if not os.path.exists(MD_PATH):
|
|
print("VERIFY FAIL: missing %s" % MD_PATH)
|
|
sys.exit(1)
|
|
md = open(MD_PATH, encoding="utf-8").read()
|
|
for rec in expected:
|
|
if "| Q%s |" % rec["id"] not in md:
|
|
errors.append("markdown missing row for %s" % rec["id"])
|
|
|
|
if errors:
|
|
print("VERIFY FAIL")
|
|
for e in errors:
|
|
print(" -", e)
|
|
sys.exit(1)
|
|
|
|
counts = {}
|
|
for r in expected:
|
|
counts["OPEN" if r["open"] else "CLOSED"] = \
|
|
counts.get("OPEN" if r["open"] else "CLOSED", 0) + 1
|
|
print("VERIFY PASS")
|
|
print("generated source=%s records=%d OPEN=%d CLOSED=%d"
|
|
% (SRC, len(expected), counts["OPEN"], counts["CLOSED"]))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|