107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier for sol-l030-classify-messages.json.
|
|
|
|
Exits 0 iff the committed artifact re-derives EVERY inbox message by the
|
|
triage rule — re-read from the raw inputs, without importing the generator —
|
|
and the aggregate counts match.
|
|
|
|
Run: python3 verify_sol_l030_classify.py
|
|
"""
|
|
import pathlib
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
INBOX = REPO / "wizard-work" / "sol" / "inbox"
|
|
ART = REPO / "wizard-work" / "sol" / "sol-l030-classify-messages.json"
|
|
|
|
# The exact rule the artifact is supposed to implement (copied for
|
|
# independent re-derivation, not imported).
|
|
LANE_RULES = [
|
|
("ops", ["cron", "deploy", "disk", "backup", "server", "fail", "uptime"]),
|
|
("finance", ["invoice", "payment", "expense", "salary", "bitcoin", "fee"]),
|
|
("person", ["alexander", "sol", "wizard", "troupe", "meeting", "alex"]),
|
|
]
|
|
LANES = ["ops", "finance", "person", "other"]
|
|
|
|
|
|
def rederive():
|
|
"""Re-derive the classification of every inbox message straight from the
|
|
raw files, using only the rule above."""
|
|
items = []
|
|
for path in sorted(INBOX.glob("*.txt")):
|
|
subject, body = "", ""
|
|
for raw in path.read_text(encoding="utf-8").splitlines():
|
|
if raw.startswith("subject:"):
|
|
subject = raw.split(":", 1)[1].strip()
|
|
elif raw.startswith("body:"):
|
|
body = raw.split(":", 1)[1].strip()
|
|
text = (subject + " " + body).lower()
|
|
lane = "other"
|
|
for name, kws in LANE_RULES:
|
|
if any(k in text for k in kws):
|
|
lane = name
|
|
break
|
|
items.append({"file": path.name, "lane": lane})
|
|
return items
|
|
|
|
|
|
def main():
|
|
problems = []
|
|
|
|
if not INBOX.is_dir():
|
|
problems.append("inbox dir missing: " + str(INBOX))
|
|
if not ART.is_file():
|
|
problems.append("artifact missing: " + str(ART))
|
|
if problems:
|
|
_fail(problems)
|
|
return 1
|
|
|
|
expected = rederive()
|
|
expected_names = {e["file"] for e in expected}
|
|
art = json.loads(ART.read_text(encoding="utf-8"))
|
|
|
|
# 1. every input re-derives
|
|
for e in expected:
|
|
a = next((i for i in art.get("items", []) if i.get("file") == e["file"]),
|
|
None)
|
|
if a is None:
|
|
problems.append("no report item for " + e["file"])
|
|
elif a.get("lane") != e["lane"]:
|
|
problems.append("lane mismatch for %r: got %r want %r" % (
|
|
e["file"], a.get("lane"), e["lane"]))
|
|
|
|
# 2. every report item came from an input
|
|
for a in art.get("items", []):
|
|
if a.get("file") not in expected_names:
|
|
problems.append("report item not in inbox: " + a.get("file", "?"))
|
|
|
|
# 3. counts
|
|
exp_counts = {l: 0 for l in LANES}
|
|
for e in expected:
|
|
exp_counts[e["lane"]] += 1
|
|
if art.get("counts") != exp_counts:
|
|
problems.append("counts %r != expected %r" % (art.get("counts"), exp_counts))
|
|
if art.get("total") != len(expected):
|
|
problems.append("total %r != expected %d" % (art.get("total"), len(expected)))
|
|
|
|
if problems:
|
|
_fail(problems)
|
|
return 1
|
|
|
|
print("PASS: sol-l030-classify-messages.json re-derives %d/%d inbox messages "
|
|
"by the triage rule; counts=%s" % (len(expected), len(expected), art.get("counts")))
|
|
return 0
|
|
|
|
|
|
def _fail(problems):
|
|
print("FAIL")
|
|
for p in problems:
|
|
print(" -", p)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|