tegwick
e85f511c4f
Compact-sitting v8 can admit a new memo from carried class, submitter, and approval attributes, but shipped sets are empty and exact-record decisions stay on their pins. The sitting image pin is unchanged. Assistant: grok Assistant-Session: 01a10329-31cb-78d3-b7a3-788c2f91d6d4
119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate attribute-admission sets from informed-decision's submitter registry.
|
|
|
|
The compact-sitting attribute branch does not author who may pose. It copies
|
|
submitter ids and the union of their memo classes from
|
|
``deploy/intake-submitters.json``. An empty registry generates empty sets, so
|
|
the branch admits nobody. A malformed registry refuses to write.
|
|
|
|
Usage:
|
|
python3 tools/gen_informed_decision_admission.py [--source PATH]
|
|
python3 tools/gen_informed_decision_admission.py [--source PATH] --check
|
|
|
|
``--check`` reports drift and writes nothing. ``--policy`` is for tests.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
POLICY = ROOT / "examples" / "informed-decision-sitting" / "policy.md"
|
|
DEFAULT_SOURCE = Path(os.path.expanduser("~/informed-decision/deploy/intake-submitters.json"))
|
|
|
|
BEGIN = "# BEGIN GENERATED admitted_sets (tools/gen_informed_decision_admission.py; do not edit by hand)"
|
|
END = "# END GENERATED admitted_sets"
|
|
SA = re.compile(r"^system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+$")
|
|
SLUG = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}$")
|
|
ROW_KEYS = {"submitter_id", "memo_classes", "review_groups", "daily_limit"}
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise SystemExit(message)
|
|
|
|
|
|
def load_sets(path: Path) -> tuple[list[str], list[str]]:
|
|
try:
|
|
document = json.loads(path.read_text())
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
fail(f"cannot read submitter registry {path}: {exc}")
|
|
if not isinstance(document, dict):
|
|
fail("intake submitters must be an object; refusing to generate")
|
|
submitters: list[str] = []
|
|
classes: set[str] = set()
|
|
for username, row in document.items():
|
|
if not isinstance(username, str) or not SA.fullmatch(username) or not isinstance(row, dict):
|
|
fail(f"refusing submitter row {username!r}")
|
|
if set(row) != ROW_KEYS:
|
|
fail(f"refusing submitter row {username!r}: keys must be {sorted(ROW_KEYS)}")
|
|
submitter_id = row["submitter_id"]
|
|
if not isinstance(submitter_id, str) or not SLUG.fullmatch(submitter_id):
|
|
fail(f"refusing submitter id {submitter_id!r}")
|
|
limit = row["daily_limit"]
|
|
if type(limit) is not int or limit < 1:
|
|
fail(f"refusing daily_limit for {username}")
|
|
for field in ("memo_classes", "review_groups"):
|
|
values = row[field]
|
|
if not isinstance(values, list) or any(not isinstance(item, str) or not SLUG.fullmatch(item) for item in values):
|
|
fail(f"refusing {field} for {username}")
|
|
submitters.append(submitter_id)
|
|
classes.update(row["memo_classes"])
|
|
if len(set(submitters)) != len(submitters):
|
|
fail("duplicate submitter_id; refusing to generate")
|
|
return sorted(set(submitters)), sorted(classes)
|
|
|
|
|
|
def render_set(name: str, values: list[str]) -> str:
|
|
if not values:
|
|
return f"{name} := set()"
|
|
body = "\n".join(f' "{value}",' for value in values)
|
|
return f"{name} := {{\n{body}\n}}"
|
|
|
|
|
|
def render_block(submitters: list[str], classes: list[str]) -> str:
|
|
return (
|
|
f"{BEGIN}\n"
|
|
f"{render_set('admitted_submitters', submitters)}\n"
|
|
"\n"
|
|
f"{render_set('admitted_classes', classes)}\n"
|
|
f"{END}"
|
|
)
|
|
|
|
|
|
def replace_block(policy: str, block: str, path: Path) -> str:
|
|
pattern = re.compile(re.escape(BEGIN) + r".*?" + re.escape(END), re.S)
|
|
if not pattern.search(policy):
|
|
fail(f"{path}: generated markers not found")
|
|
return pattern.sub(lambda _: block, policy, count=1)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
|
|
parser.add_argument("--policy", type=Path, default=POLICY)
|
|
parser.add_argument("--check", action="store_true", help="report drift, write nothing")
|
|
args = parser.parse_args()
|
|
|
|
submitters, classes = load_sets(args.source)
|
|
current = args.policy.read_text()
|
|
updated = replace_block(current, render_block(submitters, classes), args.policy)
|
|
if args.check:
|
|
if updated != current:
|
|
print(f"drift: {args.policy}", file=sys.stderr)
|
|
print(f"{len(submitters)} submitters, {len(classes)} classes; drift")
|
|
return 1
|
|
print(f"{len(submitters)} submitters, {len(classes)} classes; in sync")
|
|
return 0
|
|
if updated != current:
|
|
args.policy.write_text(updated)
|
|
print(f"wrote {args.policy}")
|
|
print(f"{len(submitters)} submitters, {len(classes)} classes")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|