volt
1423c4d77a
49 sections, 45 ids, 4 continuation sections, 13 ids with no heading at all (Q41 Q42 Q46 Q47 Q50 Q51 Q52 Q53 Q54 Q55 Q56 Q58 Q62, opened as records nested inside Q37/Q44/Q49/Q45). docs/queue-index.md is generated and carries the parse pinned in a machine block; --check re-derives and goes RED on any drift. selftest 32 cells; verify_index.py (independent parser) 26 assertions.
584 lines
25 KiB
Python
584 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""queue_index.py -- the work queue's table of contents, and the gate that it is current.
|
|
|
|
WHY THIS EXISTS
|
|
`WORK-QUEUE.md` is 1.2 MB and 9,901 lines long. It has no table of contents.
|
|
Every lane cites it by line number ("Q37 `:5278`"), so a reader who wants one
|
|
item has to grep for its heading and then find where that section ends -- and a
|
|
reader who wants the shape of the whole queue has nothing to read at all.
|
|
|
|
The shape is not obvious even after a grep. Measured on 2026-10-17 at HEAD
|
|
(`de0b1fd`): 49 `## ` sections carry 45 distinct ids, because four ids (Q37,
|
|
Q38, Q39, Q43) own a second section each -- a `closure` or `claimed` block
|
|
appended under a fresh heading. And 13 further ids (Q41, Q42, Q46, Q47, Q50,
|
|
Q51, Q52, Q53, Q54, Q55, Q56, Q58, Q62) have NO heading of their own: they are
|
|
opened as bold or prose records nested inside somebody else's section. So
|
|
"read the queue" is two different operations depending on which item you want,
|
|
and nothing said so.
|
|
|
|
WHAT IT MEASURES -- structural facts only, one row per `## ` section
|
|
id, line, end_line, role, title, lines, bytes, first `Status:` marker.
|
|
role=opening the first section carrying that id
|
|
role=continuation a later section for an id already seen (a closure or
|
|
claim block); its `role_detail` names the first word
|
|
lines/bytes the section's own extent, header line to last line
|
|
status the first `Status:` line in the section, VERBATIM and
|
|
truncated. NOT adjudicated: whether an item is open is
|
|
`tools/queue_claim_check.py`'s verdict, not this tool's.
|
|
Two instruments disagreeing about closure is a bug this
|
|
tool declines to introduce.
|
|
|
|
Plus the two legs that make the structure visible:
|
|
nested ids first seen inside a section that has no heading of their own,
|
|
with the section that contains them and the shape of that line
|
|
outside ids referenced by the prose corpus (checkins/ replies/ recoveries/
|
|
audits/ access/ decisions/ docs/ and the top-level documents) but
|
|
absent from the queue entirely -- `tools/` is excluded on purpose:
|
|
its fixtures carry synthetic ids by construction, and this tool's
|
|
own selftest is one of them.
|
|
|
|
Headings inside a fenced block are never sections; a heading whose id is
|
|
followed by `closure --` or `§9 follow-on --` is still that id's section.
|
|
|
|
WHAT IT DOES NOT DO
|
|
It does not grade the work, does not adjudicate closure, and does not read
|
|
git history. It answers "what is in this file and where", and it keeps the
|
|
generated copy of that answer honest: `--write` renders `docs/queue-index.md`
|
|
with the parse pinned inside a machine block, and `--check` re-derives the
|
|
parse and goes RED the moment the document and the queue disagree.
|
|
|
|
`--check` exit codes: 0 current, 1 stale (with the delta printed), 2 the
|
|
document carries no parseable pinned block.
|
|
|
|
READ-ONLY except `--json PATH` and `--write PATH`.
|
|
|
|
USAGE
|
|
python3 tools/queue_index.py --index # 49 sections, one screen each side
|
|
python3 tools/queue_index.py --json out.json
|
|
python3 tools/queue_index.py --write docs/queue-index.md
|
|
python3 tools/queue_index.py --check docs/queue-index.md
|
|
python3 tools/queue_index.py --selftest # fixture corpus, 32 cells
|
|
|
|
Re-run `--write` after any append to the queue. The document carries the
|
|
queue's sha256, so a stale copy is a RED check, never a silent lie.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
QUEUE_DEFAULT = "WORK-QUEUE.md"
|
|
BEGIN = "<!-- queue-index:begin -->"
|
|
END = "<!-- queue-index:end -->"
|
|
|
|
HEADING = re.compile(r"^##\s+(Q\d{1,2})\b(.*)$")
|
|
ID_TOKEN = re.compile(r"\bQ\d{1,2}\b")
|
|
STATUS = re.compile(r"^\s*\*{0,2}Status:\s*(.*)$")
|
|
FENCE = re.compile(r"^(```|~~~)")
|
|
|
|
ROLE_WORDS = ("closure", "claimed", "§", "status")
|
|
|
|
|
|
def split_lines(text):
|
|
"""Lines without their terminators, plus the byte length each line occupies."""
|
|
lines = text.split("\n")
|
|
trailing_newline = text.endswith("\n")
|
|
if trailing_newline:
|
|
lines = lines[:-1]
|
|
return lines, trailing_newline
|
|
|
|
|
|
def line_bytes(line, is_last, trailing_newline):
|
|
if is_last and not trailing_newline:
|
|
return len(line.encode("utf-8"))
|
|
return len(line.encode("utf-8")) + 1
|
|
|
|
|
|
def parse(text):
|
|
"""Return the structural parse: sections, nested ids, id map, byte accounting."""
|
|
lines, trailing_newline = split_lines(text)
|
|
n = len(lines)
|
|
sizes = [line_bytes(ln, i == n - 1, trailing_newline) for i, ln in enumerate(lines)]
|
|
|
|
headings = [] # (line_no, id, rest)
|
|
in_fence = False
|
|
for i, ln in enumerate(lines, 1):
|
|
if FENCE.match(ln.strip()):
|
|
in_fence = not in_fence
|
|
continue
|
|
if in_fence:
|
|
continue
|
|
m = HEADING.match(ln)
|
|
if m:
|
|
headings.append((i, m.group(1), m.group(2).strip()))
|
|
|
|
sections = []
|
|
seen = {}
|
|
for idx, (line_no, qid, rest) in enumerate(headings):
|
|
end = headings[idx + 1][0] - 1 if idx + 1 < len(headings) else n
|
|
role = "opening" if qid not in seen else "continuation"
|
|
detail = ""
|
|
if role == "continuation":
|
|
words = rest.lstrip("—- ").split()
|
|
detail = words[0].strip("*(") if words else ""
|
|
seen.setdefault(qid, line_no)
|
|
body = lines[line_no:end]
|
|
status = None
|
|
status_line = None
|
|
for off, ln in enumerate(body, start=line_no + 1):
|
|
m = STATUS.match(ln)
|
|
if m:
|
|
raw = m.group(1).strip()
|
|
if "**" in raw: # a bold status marker closes mid-line
|
|
raw = raw.split("**", 1)[0]
|
|
status = raw.strip()
|
|
status_line = off
|
|
break
|
|
title = rest.lstrip("—- ").strip()
|
|
sections.append({
|
|
"id": qid,
|
|
"line": line_no,
|
|
"end_line": end,
|
|
"role": role,
|
|
"role_detail": detail,
|
|
"title": clip(title, 200),
|
|
"lines": end - line_no + 1,
|
|
"bytes": sum(sizes[line_no - 1:end]),
|
|
"status": clip(status, 200) if status else None,
|
|
"status_line": status_line,
|
|
})
|
|
|
|
ids: dict = {}
|
|
for s in sections:
|
|
ids.setdefault(s["id"], []).append(s["line"])
|
|
|
|
# every id token in the queue, with its first occurrence and shape.
|
|
# Fenced lines are skipped: an id in a code block is quoted text, not a record.
|
|
mentioned = {}
|
|
in_fence = False
|
|
for i, ln in enumerate(lines, 1):
|
|
if FENCE.match(ln.strip()):
|
|
in_fence = not in_fence
|
|
continue
|
|
if in_fence:
|
|
continue
|
|
for m in ID_TOKEN.finditer(ln):
|
|
qid = m.group(0)
|
|
hit = mentioned.get(qid)
|
|
if hit is None:
|
|
mentioned[qid] = [i, ln, 1]
|
|
else:
|
|
hit[2] += 1
|
|
nested = []
|
|
for qid in sorted(mentioned, key=lambda s: int(s[1:])):
|
|
if qid in ids:
|
|
continue
|
|
line_no, raw, refs = mentioned[qid]
|
|
owner = None
|
|
for s in sections:
|
|
if s["line"] <= line_no <= s["end_line"]:
|
|
owner = s["id"]
|
|
break
|
|
stripped = raw.lstrip()
|
|
if stripped.startswith("#"):
|
|
shape = "HEADING"
|
|
elif stripped.startswith(("-", "*", "|")):
|
|
shape = "BULLET"
|
|
elif stripped.startswith("**"):
|
|
shape = "BOLD"
|
|
else:
|
|
shape = "PLAIN"
|
|
if shape != "HEADING" and "**" in raw:
|
|
shape += "+BOLD"
|
|
nested.append({"id": qid, "line": line_no, "section": owner, "shape": shape,
|
|
"refs": refs})
|
|
|
|
first = headings[0][0] if headings else n + 1
|
|
preamble_end = first - 1
|
|
payload = {
|
|
"queue_lines": n,
|
|
"queue_bytes": sum(sizes),
|
|
"queue_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
|
|
"preamble": {"line": 1, "end_line": preamble_end,
|
|
"bytes": sum(sizes[:preamble_end])},
|
|
"sections": sections,
|
|
"ids_with_section": sorted(ids, key=lambda s: int(s[1:])),
|
|
"ids_multi_section": {k: v for k, v in sorted(
|
|
ids.items(), key=lambda kv: int(kv[0][1:])) if len(v) > 1},
|
|
"nested": nested,
|
|
}
|
|
return payload
|
|
|
|
|
|
def outside_ids(repo, queue_path, exclude_tools=True):
|
|
"""Ids referenced by the prose corpus that the queue never mentions."""
|
|
try:
|
|
out = subprocess.check_output(["git", "-C", repo, "ls-files"], text=True)
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
return None
|
|
files = [f for f in out.split("\n") if f.endswith(".md")]
|
|
if exclude_tools:
|
|
files = [f for f in files if not f.startswith("tools/")]
|
|
queue_abs = os.path.normpath(os.path.join(repo, queue_path))
|
|
queue_text = open(queue_abs, encoding="utf-8", errors="replace").read()
|
|
in_queue = set(ID_TOKEN.findall(queue_text))
|
|
hits = {}
|
|
for f in files:
|
|
path = os.path.join(repo, f)
|
|
try:
|
|
txt = open(path, encoding="utf-8", errors="replace").read()
|
|
except OSError:
|
|
continue
|
|
for m in ID_TOKEN.finditer(txt):
|
|
hits.setdefault(m.group(0), set()).add(f)
|
|
return {k: sorted(v) for k, v in sorted(
|
|
hits.items(), key=lambda kv: int(kv[0][1:])) if k not in in_queue}, len(files)
|
|
|
|
|
|
def summary(payload):
|
|
secs = payload["sections"]
|
|
cont = [s for s in secs if s["role"] == "continuation"]
|
|
return ("index: %d sections, %d ids, %d continuation sections, %d nested ids, "
|
|
"%d lines / %d bytes" % (len(secs), len(payload["ids_with_section"]),
|
|
len(cont), len(payload["nested"]),
|
|
payload["queue_lines"], payload["queue_bytes"]))
|
|
|
|
|
|
def clip(text, limit):
|
|
"""Cut at a word boundary and say so; a silent mid-word cut reads as data loss."""
|
|
if len(text) <= limit:
|
|
return text
|
|
cut = text[:limit]
|
|
if " " in cut[max(0, limit - 20):]:
|
|
cut = cut.rsplit(" ", 1)[0]
|
|
return cut.rstrip(" ,;:—-") + " …"
|
|
|
|
|
|
def build_doc(payload, outside, corpus_files):
|
|
lines = []
|
|
lines.append("# The work queue's table of contents")
|
|
lines.append("")
|
|
lines.append("*Generated by `tools/queue_index.py --write`; do not hand-edit. `python3 "
|
|
"tools/queue_index.py --check docs/queue-index.md` re-derives the parse and "
|
|
"goes RED when this document and `WORK-QUEUE.md` disagree, which is what an "
|
|
"append to the queue produces. Regenerate, commit, move on.*")
|
|
lines.append("")
|
|
lines.append("**What this is.** One row per `## ` section of `WORK-QUEUE.md`: its line "
|
|
"span, its size, and the first `Status:` marker in its body verbatim. "
|
|
"Whether an item is *open* is not decided here -- that verdict is "
|
|
"`tools/queue_claim_check.py`'s, and this document does not overrule it.")
|
|
lines.append("")
|
|
lines.append("**What it exposes.** The queue is two-tier. %d sections carry %d ids because "
|
|
"%d ids own a second, continuation section. A further **%d ids have no heading "
|
|
"at all** and are opened as records nested inside another item's section -- so "
|
|
"grepping `^## %s` finds nothing while %s is a live item the file names %d times. "
|
|
"Those are listed in their own table below."
|
|
% (len(payload["sections"]), len(payload["ids_with_section"]),
|
|
len(payload["ids_multi_section"]), len(payload["nested"]),
|
|
payload["nested"][0]["id"], payload["nested"][0]["id"],
|
|
payload["nested"][0]["refs"]))
|
|
lines.append("")
|
|
lines.append("| # | id | lines | bytes | role | title | first `Status:` |")
|
|
lines.append("|---|---|---|---|---|---|---|")
|
|
for i, s in enumerate(payload["sections"], 1):
|
|
status = (s["status"] or "--").replace("|", "\\|")
|
|
title = s["title"].replace("|", "\\|")
|
|
role = s["role"] if s["role"] == "opening" else "cont: %s" % (s["role_detail"] or "?")
|
|
lines.append("| %d | %s | %d-%d | %d | %s | %s | %s |"
|
|
% (i, s["id"], s["line"], s["end_line"], s["bytes"], role, title, status))
|
|
lines.append("")
|
|
lines.append("## Ids with no `## ` heading")
|
|
lines.append("")
|
|
lines.append("Opened as records inside another section. `section` is the heading whose "
|
|
"body holds the first mention; `shape` is the form of that line.")
|
|
lines.append("")
|
|
lines.append("| id | first line | refs | inside section | shape |")
|
|
lines.append("|---|---|---|---|---|")
|
|
for r in payload["nested"]:
|
|
lines.append("| %s | %d | %d | %s | %s |"
|
|
% (r["id"], r["line"], r["refs"], r["section"] or "--", r["shape"]))
|
|
lines.append("")
|
|
if outside is not None:
|
|
lines.append("## Ids referenced outside the queue entirely")
|
|
lines.append("")
|
|
lines.append("Id tokens in the %d tracked `*.md` files under the prose corpus "
|
|
"(`tools/` excluded: its fixtures carry synthetic ids by construction) "
|
|
"that `WORK-QUEUE.md` never mentions." % corpus_files)
|
|
lines.append("")
|
|
if outside:
|
|
lines.append("| id | files |")
|
|
lines.append("|---|---|")
|
|
for qid, files in outside.items():
|
|
lines.append("| %s | %s |" % (qid, ", ".join("`%s`" % f for f in files[:4])
|
|
+ (" +%d" % (len(files) - 4) if len(files) > 4 else "")))
|
|
else:
|
|
lines.append("None: every id the prose corpus cites is mentioned in the queue.")
|
|
lines.append("")
|
|
lines.append("## The pinned parse")
|
|
lines.append("")
|
|
lines.append("The block below is the parse this document renders, byte for byte. "
|
|
"`--check` compares a fresh parse against it.")
|
|
lines.append("")
|
|
lines.append(BEGIN)
|
|
lines.append("```json")
|
|
lines.append(json.dumps(payload, indent=1, sort_keys=True))
|
|
lines.append("```")
|
|
lines.append(END)
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def extract_pinned(text):
|
|
if BEGIN not in text or END not in text:
|
|
return None
|
|
body = text.split(BEGIN, 1)[1].split(END, 1)[0]
|
|
m = re.search(r"```json\n(.*?)\n```", body, re.S)
|
|
if not m:
|
|
return None
|
|
try:
|
|
return json.loads(m.group(1))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def delta(pinned, fresh):
|
|
"""A short, specific account of how a pinned parse and a fresh one differ."""
|
|
out = []
|
|
p_sec = {(s["id"], s["line"]): s for s in pinned.get("sections", [])}
|
|
f_sec = {(s["id"], s["line"]): s for s in fresh.get("sections", [])}
|
|
added = [k for k in f_sec if k not in p_sec]
|
|
removed = [k for k in p_sec if k not in f_sec]
|
|
moved = [k for k in f_sec if k in p_sec and
|
|
(f_sec[k]["end_line"], f_sec[k]["bytes"], f_sec[k]["status"])
|
|
!= (p_sec[k]["end_line"], p_sec[k]["bytes"], p_sec[k]["status"])]
|
|
if added:
|
|
out.append("%d new section(s): %s" % (len(added), ", ".join("%s@%d" % k for k in added)))
|
|
if removed:
|
|
out.append("%d gone section(s): %s" % (len(removed), ", ".join("%s@%d" % k for k in removed)))
|
|
if moved:
|
|
out.append("%d section(s) changed extent or status, e.g. %s@%d %d->%d lines"
|
|
% (len(moved), moved[0][0], moved[0][1],
|
|
p_sec[moved[0]]["end_line"] - p_sec[moved[0]]["line"] + 1,
|
|
f_sec[moved[0]]["end_line"] - f_sec[moved[0]]["line"] + 1))
|
|
p_nest = [r["id"] for r in pinned.get("nested", [])]
|
|
f_nest = [r["id"] for r in fresh.get("nested", [])]
|
|
if p_nest != f_nest:
|
|
out.append("nested id set changed: +%s -%s"
|
|
% (",".join(sorted(set(f_nest) - set(p_nest), key=lambda s: int(s[1:]))),
|
|
",".join(sorted(set(p_nest) - set(f_nest), key=lambda s: int(s[1:])))))
|
|
if pinned.get("queue_sha256") != fresh.get("queue_sha256"):
|
|
out.append("queue sha256 %s -> %s" % (str(pinned.get("queue_sha256"))[:12],
|
|
str(fresh.get("queue_sha256"))[:12]))
|
|
return "; ".join(out) if out else "no structural difference (byte-level drift only)"
|
|
|
|
|
|
def run(args):
|
|
repo = args.repo
|
|
queue_path = os.path.join(repo, args.queue)
|
|
with open(queue_path, encoding="utf-8") as fh:
|
|
text = fh.read()
|
|
payload = parse(text)
|
|
|
|
if args.check:
|
|
doc = open(args.check, encoding="utf-8").read()
|
|
pinned = extract_pinned(doc)
|
|
if pinned is None:
|
|
print("UNVERIFIABLE: no parseable pinned block in %s" % args.check)
|
|
return 2
|
|
if json.dumps(pinned, sort_keys=True) == json.dumps(payload, sort_keys=True):
|
|
print("index current: %s" % summary(payload))
|
|
return 0
|
|
print("STALE: %s" % delta(pinned, payload))
|
|
print("regenerate with: python3 tools/queue_index.py --repo %s --write %s"
|
|
% (repo, args.check))
|
|
return 1
|
|
|
|
outside, corpus = outside_ids(repo, args.queue)
|
|
if args.write:
|
|
with open(args.write, "w", encoding="utf-8") as fh:
|
|
fh.write(build_doc(payload, outside, corpus or 0))
|
|
print("wrote %s (%d bytes)" % (args.write, os.path.getsize(args.write)))
|
|
return 0
|
|
|
|
if args.json:
|
|
with open(args.json, "w", encoding="utf-8") as fh:
|
|
json.dump(payload, fh, indent=2, sort_keys=True)
|
|
fh.write("\n")
|
|
|
|
if args.index:
|
|
print("%-4s %-5s %-11s %-7s %-13s %-22s %s"
|
|
% ("#", "id", "lines", "bytes", "role", "status", "title"))
|
|
for i, s in enumerate(payload["sections"], 1):
|
|
print("%-4d %-5s %-11s %-7d %-13s %-22s %s"
|
|
% (i, s["id"], "%d-%d" % (s["line"], s["end_line"]), s["bytes"],
|
|
s["role"][:13], clip((s["status"] or "--"), 22), clip(s["title"], 72)))
|
|
print()
|
|
for r in payload["nested"]:
|
|
print("nested %-4s line %-5d refs %-3d inside %-4s %s"
|
|
% (r["id"], r["line"], r["refs"], r["section"] or "--", r["shape"]))
|
|
if outside is not None and outside:
|
|
print("outside the queue: %s" % ", ".join(outside))
|
|
print()
|
|
|
|
print(summary(payload))
|
|
return 0
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# selftest: a fixture corpus, one cell per stated rule
|
|
# --------------------------------------------------------------------------
|
|
|
|
FIXTURE = """# Work queue
|
|
|
|
Preface text.
|
|
|
|
## Q1 — First item *(lane)*
|
|
|
|
**Status: DONE at `abc1234`.** Body.
|
|
|
|
```
|
|
## Q9 — not a section, this is fenced
|
|
```
|
|
|
|
## Q2 — Second item
|
|
|
|
Body with no status marker.
|
|
|
|
## Q1 — closure — verified
|
|
|
|
**Status: OPEN.** A continuation section for Q1.
|
|
|
|
## Q3 §9 follow-on — odd shape
|
|
|
|
- **Q41 — OPEN (pipe, 2026-10-13, unclaimed).** A nested record.
|
|
- prose mentioning Q42 in passing.
|
|
"""
|
|
|
|
|
|
def selftest():
|
|
cells = [0]
|
|
bad = [0]
|
|
|
|
def check(name, got, want):
|
|
cells[0] += 1
|
|
if got != want:
|
|
bad[0] += 1
|
|
print("FAIL %s\n got %r\n want %r" % (name, got, want))
|
|
|
|
p = parse(FIXTURE)
|
|
secs = p["sections"]
|
|
check("four sections parsed", len(secs), 4)
|
|
check("ids in order", [s["id"] for s in secs], ["Q1", "Q2", "Q1", "Q3"])
|
|
check("fenced heading is not a section", "Q9" in p["ids_with_section"], False)
|
|
check("second Q1 is a continuation", secs[2]["role"], "continuation")
|
|
check("continuation detail is the role word", secs[2]["role_detail"], "closure")
|
|
check("odd heading shape still parses", secs[3]["id"], "Q3")
|
|
check("multi-section map names Q1", p["ids_multi_section"], {"Q1": [5, 17]})
|
|
check("status marker extracted verbatim", secs[0]["status"], "DONE at `abc1234`.")
|
|
check("status line number", secs[0]["status_line"], 7)
|
|
check("section with no marker is None", secs[1]["status"], None)
|
|
check("continuation status is its own", secs[2]["status"], "OPEN.")
|
|
check("nested ids found", [r["id"] for r in p["nested"]], ["Q41", "Q42"])
|
|
check("nested owner is the containing section", p["nested"][0]["section"], "Q3")
|
|
check("nested shape is bullet carrying a bold lead", p["nested"][0]["shape"], "BULLET+BOLD")
|
|
check("nested shape is a plain bullet", p["nested"][1]["shape"], "BULLET")
|
|
check("sections tile the file with no gap",
|
|
all(secs[i]["end_line"] + 1 == secs[i + 1]["line"] for i in range(len(secs) - 1)), True)
|
|
check("last section ends at the last line", secs[-1]["end_line"], p["queue_lines"])
|
|
check("preamble ends before the first heading", p["preamble"]["end_line"], 4)
|
|
check("byte accounting is exact",
|
|
p["preamble"]["bytes"] + sum(s["bytes"] for s in secs), p["queue_bytes"])
|
|
check("bytes are the file's own length", p["queue_bytes"], len(FIXTURE.encode()))
|
|
check("sha256 is the file's own", p["queue_sha256"],
|
|
hashlib.sha256(FIXTURE.encode()).hexdigest())
|
|
|
|
# --- the gate, round trip and both directions of falsification ----------
|
|
root = tempfile.mkdtemp(prefix="queue-index-selftest-")
|
|
repo = os.path.join(root, "repo")
|
|
os.makedirs(repo)
|
|
queue = os.path.join(repo, "WORK-QUEUE.md")
|
|
doc_path = os.path.join(repo, "index.md")
|
|
subprocess.check_call(["git", "-C", repo, "init", "-q", "-b", "main"])
|
|
subprocess.check_call(["git", "-C", repo, "config", "user.email", "s@invalid"])
|
|
subprocess.check_call(["git", "-C", repo, "config", "user.name", "s"])
|
|
with open(queue, "w", encoding="utf-8") as fh:
|
|
fh.write(FIXTURE)
|
|
subprocess.check_call(["git", "-C", repo, "add", "-A"])
|
|
subprocess.check_call(["git", "-C", repo, "commit", "-q", "-m", "fixture"])
|
|
|
|
import contextlib
|
|
import io
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = run(argparse.Namespace(repo=repo, queue="WORK-QUEUE.md", index=False,
|
|
json=None, write=doc_path, check=None))
|
|
check("--write exits 0", rc, 0)
|
|
check("document written", os.path.exists(doc_path), True)
|
|
doc = open(doc_path, encoding="utf-8").read()
|
|
check("document carries the pinned block", extract_pinned(doc) is not None, True)
|
|
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = run(argparse.Namespace(repo=repo, queue="WORK-QUEUE.md", index=False,
|
|
json=None, write=None, check=doc_path))
|
|
check("--check on its own output exits 0", rc, 0)
|
|
check("--check says current", "index current:" in buf.getvalue(), True)
|
|
|
|
# falsify 1: an append moves the last section's extent
|
|
with open(queue, "a", encoding="utf-8") as fh:
|
|
fh.write("- an appended line.\n")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = run(argparse.Namespace(repo=repo, queue="WORK-QUEUE.md", index=False,
|
|
json=None, write=None, check=doc_path))
|
|
check("--check goes RED on an append", rc, 1)
|
|
check("RED names the moved section", "changed extent or status" in buf.getvalue(), True)
|
|
|
|
# falsify 2: a new heading is a new section
|
|
with open(queue, "a", encoding="utf-8") as fh:
|
|
fh.write("\n## Q7 — a brand new item\n\n**Status: OPEN.**\n")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = run(argparse.Namespace(repo=repo, queue="WORK-QUEUE.md", index=False,
|
|
json=None, write=None, check=doc_path))
|
|
check("--check goes RED on a new heading", rc, 1)
|
|
check("RED names the new section", "new section(s): Q7" in buf.getvalue(), True)
|
|
|
|
# falsify 3: a document with no block cannot be verified
|
|
open(doc_path, "w", encoding="utf-8").write("# empty\n")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = run(argparse.Namespace(repo=repo, queue="WORK-QUEUE.md", index=False,
|
|
json=None, write=None, check=doc_path))
|
|
check("no pin -> exit 2", rc, 2)
|
|
check("no pin -> says so", "UNVERIFIABLE" in buf.getvalue(), True)
|
|
|
|
print("selftest: %d cells, %d passed, %d failed" % (cells[0], cells[0] - bad[0], bad[0]))
|
|
return 1 if bad[0] else 0
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(
|
|
description="queue_index.py -- the work queue's table of contents, and the gate "
|
|
"that it is current.")
|
|
ap.add_argument("--repo", default=".")
|
|
ap.add_argument("--queue", default=QUEUE_DEFAULT)
|
|
ap.add_argument("--index", action="store_true", help="print the section table")
|
|
ap.add_argument("--json", metavar="PATH", help="write the parse here")
|
|
ap.add_argument("--write", metavar="PATH", help="render the index document here")
|
|
ap.add_argument("--check", metavar="PATH", help="re-derive and compare a rendered index")
|
|
ap.add_argument("--selftest", action="store_true", help="fixture corpus, 32 cells")
|
|
args = ap.parse_args(argv)
|
|
if args.selftest:
|
|
return selftest()
|
|
return run(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|