Files
troupe-standup/wizard-work/sol/verify_triage_queue.py

131 lines
4.2 KiB
Python

#!/usr/bin/env python3
"""CHECK for the work-queue triage deliverable (sol).
Re-derives the classification from WORK-QUEUE.md, re-reads the report it
generated, diffs the two, and independently re-runs git rev-parse for every
ref verdict recorded in the report. Exit 0 == the report passes the check;
nonzero == wrong.
Bare command (the VERDICT):
python3 /root/.hermes/wizard-outputs/sol/verify_triage_queue.py
"""
import os
import re
import subprocess
import sys
REPO = '/root/troupe-standup'
SRC = os.path.join(REPO, 'WORK-QUEUE.md')
REPORT = os.path.join(REPO, 'wizard-work', 'sol', 'queue_triage.md')
failures = []
def check(cond, msg):
if not cond:
failures.append(msg)
def clean(s):
return ' '.join(s.split())
def parse_questions():
parts = re.split(r'\n(?=#{2,3} )', open(SRC).read())
hdr_re = re.compile(r'^(#{2,3})\s+(.*)')
out = []
for p in parts:
m = hdr_re.match(p)
if not m:
continue
header = clean(m.group(2))
if not re.match(r'^Q\d', header):
continue
body = p
sm = re.search(r'\*\*Status:\s*(.*?)\*\*', body, re.M)
out.append({'header': header, 'status': clean(sm.group(1)) if sm else '',
'body': body})
return out
def bucket(q):
s = q['status']
if re.search(r'(DONE|CLOSED|RESOLVED|VERDICT)', s):
return 'DONE'
if 'not yet opened' in q['header'].lower() or 'not yet opened' in s.lower():
return 'PROPOSED'
if 'exhausted' in s.lower() or '0/15' in s:
return 'EXHAUSTED'
return 'OPEN'
def report_entries():
entries = []
if not os.path.isfile(REPORT):
failures.append('report missing: ' + REPORT)
return entries
for line in open(REPORT):
m = re.match(r'^## \[(DONE|OPEN|PROPOSED|EXHAUSTED)\] (.*)$', line)
if m:
entries.append({'bucket': m.group(1), 'header': m.group(2).strip()})
return entries
def ref_resolves(ref):
r = subprocess.run(['git', '-C', REPO, 'rev-parse', '-q', '--verify', ref],
capture_output=True, text=True)
return r.returncode == 0
def main():
qs = parse_questions()
exp = [{'bucket': bucket(q), 'header': q['header']} for q in qs]
rep = report_entries()
check(rep == exp, 'CLASSIFY: report entries do not match queue derivation')
# summary / bucket counts
for line in open(REPORT):
if line.startswith('questions:'):
check(line.split('questions:')[1].split()[0] == str(len(qs)),
'SUMMARY: total %r != %d' % (line.split('questions:')[1], len(qs)))
for k in ('DONE', 'OPEN', 'PROPOSED', 'EXHAUSTED'):
m = re.match(r'^' + k + r'=(\d+)', line)
if m:
want = sum(1 for e in exp if e['bucket'] == k)
check(int(m.group(1)) == want, 'COUNT %s=%d want %d' %
(k, int(m.group(1)), want))
# ref verdicts re-run against git
for line in open(REPORT):
m = re.match(r'^ evidence: ref (\S+) -> (RESOLVED|RESOLVED-FALSE)$', line)
if m:
ref, verdict = m.group(1), m.group(2)
real = ref_resolves(ref)
check(verdict == ('RESOLVED' if real else 'RESOLVED-FALSE'),
'EVIDENCE ref %s = %s but git says %s' % (ref, verdict,
'RESOLVED' if real else 'RESOLVED-FALSE'))
# file-evidence sanity: every question that names a recovery file has a
# corresponding PRESENT/MISSING entry in the report
rep_text = open(REPORT).read()
for i, q in enumerate(qs):
if i >= len(rep):
break
has_named_file = bool(re.search(r'recoveries/[A-Za-z0-9_.-]+', q['body']))
has_evidence = 'evidence: recovery file' in rep_text
if has_named_file and not has_evidence:
failures.append('EVIDENCE: Q%s names a recovery file but none listed' % q['header'])
print('CHECK questions=%d failures=%d' % (len(qs), len(failures)))
if failures:
for f in failures:
print(' FAIL:', f)
sys.exit(1)
print('CHECK PASS: report matches queue derivation and all refs verified')
sys.exit(0)
if __name__ == '__main__':
main()