147 lines
5.5 KiB
Python
147 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verification for the workqueue-census-v2 artifact.
|
|
|
|
RE-APPROACH: this checker does NOT import the extractor. It re-derives the
|
|
expected rows from WORK-QUEUE.md with plain string operations (split on
|
|
whitespace, startswith); the committed TSV is the artifact of record.
|
|
The verdict is the script's exit code: 0 = correct, nonzero = wrong.
|
|
|
|
Checks:
|
|
1. every '## Q<n>' heading in the source has exactly one committed row, in
|
|
source order, with a verbatim title;
|
|
2. row counts, distinct-q count and reopened-q list in extract_manifest.json
|
|
agree with an independent recomputation;
|
|
3. extracted_at_utc is a measured timestamp (not the stale constant) within
|
|
120s of now and parseable as an UTC date;
|
|
4. no committed row has an empty q or title, and no status/done value
|
|
contains a literal tab (delimiter corruption).
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
|
SRC = os.path.join(REPO_ROOT, "WORK-QUEUE.md")
|
|
TSV = os.path.join(HERE, "workqueue-census.tsv")
|
|
MD = os.path.join(HERE, "workqueue-census.md")
|
|
MANIFEST = os.path.join(HERE, "extract_manifest.json")
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
if cond:
|
|
print("PASS " + name)
|
|
else:
|
|
print("FAIL " + name + " " + repr(detail))
|
|
fails.append(name)
|
|
|
|
|
|
# --- 1. Independent re-derivation from the source (string ops, no regex) ------
|
|
golden = [] # (qid, title) in source order
|
|
qid = None
|
|
status = ""
|
|
done_implies = ""
|
|
with open(SRC, encoding="utf-8") as fh:
|
|
src_lines = fh.read().splitlines()
|
|
|
|
for i, line in enumerate(src_lines):
|
|
if line.startswith("## "):
|
|
rest = line[3:] # strip the "## " marker
|
|
parts = rest.split(None, 1)
|
|
if parts and len(parts[0]) > 1 and parts[0][0] == "Q" and parts[0][1:].isdigit():
|
|
# flush previous block
|
|
if qid is not None:
|
|
golden.append((qid, title))
|
|
qid = parts[0]
|
|
title = parts[1].strip() if len(parts) > 1 else ""
|
|
status = ""
|
|
done_implies = ""
|
|
else:
|
|
# not a new Q heading: nothing to do (source has no other '## ' lines)
|
|
continue
|
|
else:
|
|
if qid is None:
|
|
continue
|
|
if status == "" and line.startswith("**Status:"):
|
|
status = line[len("**Status:"):].strip()
|
|
# collapse whitespace, truncate at first sentence end
|
|
status = re.sub(r"\s+", " ", status)
|
|
m = re.search(r"\.\s", status)
|
|
if m:
|
|
status = status[: m.start()] + "."
|
|
if status.endswith("."):
|
|
status = status[: -1]
|
|
status += "."
|
|
if done_implies == "" and line.startswith("**Done means:"):
|
|
done_implies = line[len("**Done means:"):].strip()
|
|
|
|
if qid is not None:
|
|
golden.append((qid, title))
|
|
|
|
# --- 2. Committed TSV --------------------------------------------------------
|
|
with open(TSV, encoding="utf-8", newline="") as fh:
|
|
rows = list(csv.reader(fh, delimiter="\t"))
|
|
header, committed = rows[0], rows[1:]
|
|
|
|
check("tsv header", header == ["q", "title", "status", "done_implies"], repr(header))
|
|
check("row count",
|
|
len(committed) == len(golden),
|
|
"committed=%d golden=%d" % (len(committed), len(golden)))
|
|
|
|
mism = [k for k, (c, g) in enumerate(zip(committed, golden))
|
|
if c[0] != g[0] or c[1] != g[1]]
|
|
check("qid+title identical to independent re-derive", not mism,
|
|
"first mismatch row %r" % (mism[0] if mism else None))
|
|
|
|
# --- 3. Manifest self-consistency (independent recomputation) ------------------
|
|
with open(MANIFEST, encoding="utf-8") as fh:
|
|
man = json.load(fh)
|
|
|
|
counts = {}
|
|
for q, _ in golden:
|
|
counts[q] = counts.get(q, 0) + 1
|
|
distinct = sorted(counts)
|
|
reopened = sorted(q for q, c in counts.items() if c > 1)
|
|
|
|
check("manifest.rows", man.get("rows") == len(golden),
|
|
"manifest=%r actual=%d" % (man.get("rows"), len(golden)))
|
|
check("manifest.distinct_q_labels", man.get("distinct_q_labels") == len(distinct),
|
|
"manifest=%r actual=%d" % (man.get("distinct_q_labels"), len(distinct)))
|
|
check("manifest.reopened_q_labels", man.get("reopened_q_labels") == reopened,
|
|
"manifest=%r actual=%r" % (man.get("reopened_q_labels"), reopened))
|
|
|
|
# --- 4. extracted_at_utc is measured, not a hardcoded constant -----------------
|
|
auto = man.get("extracted_at_utc", "")
|
|
check("extracted_at_utc key", isinstance(auto, str) and auto,
|
|
repr(auto))
|
|
try:
|
|
dt = datetime.strptime(auto, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
|
|
now = datetime.now(timezone.utc)
|
|
check("extracted_at_utc recency",
|
|
abs((now - dt).total_seconds()) < 120,
|
|
"auto=%s now=%s delta=%ss" % (auto, now.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
abs((now - dt).total_seconds())))
|
|
except ValueError:
|
|
check("extracted_at_utc format", False, repr(auto))
|
|
|
|
# --- 5. Corruption checks ------------------------------------------------------
|
|
empty = [r for r in committed if not r[0] or not r[1]]
|
|
check("no empty q/title", not empty, repr(empty[:1]))
|
|
bad_tab = [r for r in committed if any("\t" in c for c in r[2:])]
|
|
check("status/done no tabs", not bad_tab, repr(bad_tab[:1]))
|
|
|
|
if fails:
|
|
print("")
|
|
print("%d check(s) FAILED" % len(fails))
|
|
sys.exit(1)
|
|
|
|
print("")
|
|
print("ALL CHECKS PASSED (%d rows, %d distinct labels, %d reopened)"
|
|
% (len(golden), len(distinct), len(reopened)))
|