API.md: every admin route tested over HTTP and approved

123 checks across all 48 routes plus /api/docs, driven on a throwaway
copy of the live database with anonymous, member, leader, admin and
owner access; 0 failures. Calendar checks used 2036 probes against the
real store and left it at 32 objects. API.md records, per route, the
capability and what each request actually returned. tests/api_drive.py
is the harness and tests/api_doc.py regenerates the document from its
results, so the next approval run is a rerun, not a rewrite.
This commit is contained in:
2026-09-04 20:27:19 -04:00
parent d1dfa6df90
commit 752fb08d73
3 changed files with 797 additions and 0 deletions
+80
View File
@@ -0,0 +1,80 @@
"""Turn results.json from api_drive.py into API.md: one section per route group, one table per
method and path, every case tried and the HTTP status it returned. Capability and first docstring
line come from the live route registry."""
import json, collections, re, subprocess, sys, os
D = os.path.dirname(os.path.abspath(__file__))
R = json.load(open(os.path.join(D, "results.json")))
OUT = sys.argv[1] if len(sys.argv) > 1 else os.path.join(D, "API.md")
UUID = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
def norm(p):
p = p.split("?")[0]
p = re.sub("/" + UUID + r"@site73\.greenlanescouts73\.org", "/{uid}", p)
p = re.sub("/" + UUID, "/{id}", p)
p = re.sub(r"/pack-73/zz-(api|old)", "/{id}", p)
p = p.replace("/x@band.us", "/{uid}")
p = re.sub(r"/units/(pack73|troop73|crew99)$", "/units/{slug}", p)
p = re.sub(r"/settings/(api_keys_from|nope)$", "/settings/{key}", p)
p = re.sub(r"/checks/(dues|dob)$", "/checks/{item}", p)
p = re.sub(r"/people/nope/", "/people/{id}/", p)
p = re.sub(r"/(leads|announcements|nearby|keys)/nope$", r"/\1/{id}", p)
return p
groups = collections.OrderedDict()
for m, p, case, exp, got, ok, note in R:
groups.setdefault(norm(p), collections.OrderedDict()).setdefault(m, []).append((case, got))
caps = {}
line = subprocess.check_output(["docker", "exec", "scout-website", "python3", "-c",
"import admin_api,json; print(json.dumps([(d['methods'],d['path'],d['capability'],d['doc']) for d in admin_api.describe_routes()]))"]).decode().strip().splitlines()[-1]
for methods, path, cap, doc in json.loads(line):
for m in methods: caps[(m, path)] = (cap, (doc or "").strip())
def capfor(m, p):
for (mm, pp), v in caps.items():
if mm != m: continue
pat = "^" + re.sub(r"\\\{[^}]+\\\}", r"\\{[a-z_]+\\}", re.escape(pp.replace("{pid:path}", "{id}"))) + "$"
if re.match(pat, p): return v
return (None, "")
SECTIONS = [
("Identity", ["/api/admin/whoami", "/api/docs"], "Who is calling, and the live route registry."),
("Summary and leads", ["/api/admin/summary", "/api/admin/leads", "/api/admin/leads/{id}", "/api/admin/leads/{id}/claim", "/api/admin/leads/{id}/release", "/api/admin/mirrors/failed", "/api/admin/mirrors/retry"],
"Join-form leads are read-only. A claim is who has a lead and when; it is the only thing that changes."),
("Announcements", ["/api/admin/announcements", "/api/admin/announcements/{id}"], "Site banners. `ends_at` is required. Revoke keeps the row."),
("Nearby units", ["/api/admin/nearby", "/api/admin/nearby/{id}"], "The find-a-unit directory. Deactivate, never delete. Saving bumps `verified_at`."),
("Our units", ["/api/admin/units", "/api/admin/units/{slug}"], "Meeting day, time and place. A unit leader edits their own unit only."),
("Settings", ["/api/admin/settings", "/api/admin/settings/{key}"], "Typed keys only; `value: null` clears to the code default."),
("API keys", ["/api/admin/keys", "/api/admin/keys/{id}"], "Bearer keys, honoured on `/api/admin` only. Scopes are a subset of the owner's capabilities; a key can never mint keys. LAN-only unless `api_keys_from` is `anywhere`."),
("History", ["/api/admin/history"], "The action log, newest first. Admin and above."),
("People", ["/api/admin/people", "/api/admin/people/invite", "/api/admin/people/invite/{id}", "/api/admin/people/{id}/roles", "/api/admin/people/{id}/disable", "/api/admin/people/{id}/enable", "/api/admin/people/{id}/reset"],
"Invites and reset links are returned once and handed over by the admin; there is no outbound mail. Grants are limited to what the caller holds. Owner-only: roles, disable, enable."),
("Roster and family", ["/api/admin/roster", "/api/admin/roster/households", "/api/admin/roster/households/{id}", "/api/admin/roster/households/{id}/scouts", "/api/admin/roster/households/{id}/people", "/api/admin/roster/scouts/{id}", "/api/admin/roster/scouts/{id}/checks/{item}", "/api/admin/family"],
"Families, scouts, and a per-year checklist recording that a thing was collected, never the thing. Nothing medical is stored. `/family` is the signed-in parent's own households."),
("Calendar", ["/api/admin/calendar", "/api/admin/calendar/{uid}"], "Writes go to Radicale. The site owns its two UID namespaces and refuses any other before a network call. DELETE really deletes."),
("Facebook posts", ["/api/admin/fbposts", "/api/admin/fbposts/ingest", "/api/admin/fbposts/{id}/image", "/api/admin/fbposts/{id}/cancel", "/api/admin/fbposts/{id}/reschedule"],
"The publisher reports; leaders read, cancel and edit. The image hash is verified on arrival and the image route is gated like the list. Cancel and reschedule go through the publisher's signed per-post link."),
]
out = ["# Green Lane Scouts 73: Admin API", "",
"_Tested and approved 2026-09-05. Every route below was driven over HTTP on a throwaway copy of the",
"live database with four levels of access (anonymous, member, unit leader, admin, plus the site owner",
"where a route is owner-only): %d checks, 0 failures. The calendar checks ran against the real store" % len(R),
"with 2036 probes and left it as found. The live registry is `GET /api/docs`; this file records what",
"each route actually did when asked. Regenerate with `tests/api_drive.py` then `tests/api_doc.py`._", "",
"**Base:** `https://greenlanescouts73.org`. **Auth:** a browser session (cookie `s73_session`) or a",
"bearer API key (`Authorization: Bearer gls73_...`), honoured on `/api/admin` only. `401` means no",
"credential; `403` means a credential without the capability. Error bodies are `{\"detail\": ...}` or,",
"from a store guardrail, `{\"detail\": {\"error\": ...}}`. Every write lands in the action log with who",
"did it and what changed. Times are ISO 8601 with an offset.", ""]
seen = set()
for title, paths, blurb in SECTIONS:
out += ["## " + title, "", blurb, ""]
for p in paths:
if p not in groups: raise SystemExit("no results for " + p)
seen.add(p)
for m, cases in groups[p].items():
cap, doc = capfor(m, p)
out += ["### `%s %s`" % (m, p), ""]
if cap: out.append("Capability `%s`. %s" % (cap, doc.splitlines()[0] if doc else ""))
elif p == "/api/docs": out.append("Capability `api:docs`, leader and above. Generated from the router on every request.")
else: out.append("Any credential; answers who you are, how you authenticated, and your capabilities.")
out += ["", "| Tried | HTTP |", "|---|---|"] + ["| %s | %s |" % (c, g) for c, g in cases] + [""]
missing = [k for k in groups if k not in seen]
assert not missing, missing
open(OUT, "w").write("\n".join(out) + "\n")
print("wrote", OUT, ":", len(out), "lines;", len(groups), "route paths;", len(R), "checks;", sum(1 for l in out if l.startswith("### ")), "sections")
+189
View File
@@ -0,0 +1,189 @@
"""Drive every admin API route over HTTP against a throwaway site. Not a unit test: needs a running
site at the URL in base and the sessions in setup.json (see API.md for how the 2026-09-05 run was set
up). Writes results.json; api_doc.py turns that into API.md."""
import json, urllib.request, urllib.error, urllib.parse, hashlib, struct, zlib, base64, os, subprocess
D = os.path.dirname(os.path.abspath(__file__))
B = open(os.path.join(D, "base")).read().strip()
S = json.load(open(os.path.join(D, "setup.json")))
R = []
def call(method, path, who=None, body=None, headers=None, token=None):
h = {"X-Forwarded-For": "10.0.0.5"}
if who: h["Cookie"] = "s73_session=" + S[who]
if token: h["Authorization"] = "Bearer " + token
if headers: h.update(headers)
data = None
if body is not None:
data = json.dumps(body).encode(); h["Content-Type"] = "application/json"
req = urllib.request.Request(B + path, data=data, method=method, headers=h)
try:
with urllib.request.urlopen(req, timeout=60) as r:
ct = r.headers.get("Content-Type", ""); b = r.read()
return r.status, (json.loads(b) if "json" in ct else b)
except urllib.error.HTTPError as e:
ct = e.headers.get("Content-Type", ""); b = e.read()
try: return e.code, json.loads(b)
except Exception: return e.code, b
def t(method, path, case, expect, who=None, body=None, token=None, headers=None, check=None):
st, d = call(method, path, who, body, headers=headers, token=token)
ok = (st == expect) if isinstance(expect, int) else (st in expect)
note = ""
if ok and check:
try:
ok = bool(check(d)); note = "" if ok else "check failed"
except Exception as e:
ok = False; note = "check raised %s" % e
R.append((method, path, case, expect, st, ok, note))
return d
def png(w, h):
def chunk(t, d): return struct.pack(">I", len(d)) + t + d + struct.pack(">I", zlib.crc32(t + d) & 0xffffffff)
return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)) + chunk(b"IEND", b"")
t("GET", "/api/admin/whoami", "anon", 401)
t("GET", "/api/admin/whoami", "claude session", 200, "claude", check=lambda d: d["via"] == "session" and "leads:read" in d["capabilities"])
t("GET", "/api/admin/whoami", "member session", 200, "member", check=lambda d: "leads:read" not in d["capabilities"])
t("GET", "/api/admin/whoami", "admin token from LAN", 200, headers={"X-Admin-Token": "testtoken"})
t("GET", "/api/admin/whoami", "admin token from WAN", 403, headers={"X-Admin-Token": "testtoken", "X-Forwarded-For": "108.36.248.87"})
t("GET", "/api/docs", "anon", 401)
t("GET", "/api/docs", "member (no api:docs)", 403, "member")
t("GET", "/api/docs", "leader", 200, "leader", check=lambda d: b"/api/admin/leads" in d)
t("GET", "/api/admin/summary", "anon", 401)
t("GET", "/api/admin/summary", "member", 403, "member")
t("GET", "/api/admin/summary", "leader", 200, "leader", check=lambda d: "total" in d and "unclaimed" in d)
leads = t("GET", "/api/admin/leads?limit=5", "leader", 200, "leader", check=lambda d: isinstance(d["leads"], list))
t("GET", "/api/admin/leads?q=zzqqxx", "search miss", 200, "leader", check=lambda d: d["leads"] == [])
lid = leads["leads"][0]["id"] if leads["leads"] else None
if lid:
t("GET", "/api/admin/leads/" + lid, "one lead", 200, "leader", check=lambda d: d["id"] == lid and "mirrors" in d and "claim_history" in d)
t("POST", "/api/admin/leads/%s/claim" % lid, "claim", 200, "leader", check=lambda d: d["claim"]["email"] == "zz.leader@example.test")
t("POST", "/api/admin/leads/%s/claim" % lid, "claim again (self)", 409, "leader")
t("POST", "/api/admin/leads/%s/claim" % lid, "claim by another", 409, "claude")
t("POST", "/api/admin/leads/%s/release" % lid, "release by another admin (no people:manage)", 403, "claude")
t("POST", "/api/admin/leads/%s/release" % lid, "release by owner", 200, "owner", check=lambda d: d["claim"] is None)
t("POST", "/api/admin/leads/%s/release" % lid, "release when nobody has it", 409, "leader")
t("GET", "/api/admin/leads/nope", "unknown lead", 404, "leader")
t("GET", "/api/admin/mirrors/failed?target=google_sheet", "failed mirrors", 200, "leader", check=lambda d: "leads" in d)
t("POST", "/api/admin/mirrors/retry?target=google_sheet", "retry sheet (nothing owed)", 200, "leader", check=lambda d: "retried_ok" in d)
t("POST", "/api/admin/mirrors/retry?target=ntfy", "retry ntfy (not implemented)", 422, "leader")
t("GET", "/api/admin/announcements?include_expired=true", "list", 200, "leader", check=lambda d: isinstance(d["announcements"], list))
t("POST", "/api/admin/announcements", "no ends_at", 422, "leader", body={"message": "x"})
an = t("POST", "/api/admin/announcements", "create (2036)", 201, "leader",
body={"message": "ZZ api test", "starts_at": "2036-05-01T12:00:00+00:00", "ends_at": "2036-05-02T12:00:00+00:00"},
check=lambda d: d["created_by"] == "zz.leader@example.test")
t("DELETE", "/api/admin/announcements/" + an["id"], "revoke", 200, "leader", check=lambda d: d["revoked_at"])
t("DELETE", "/api/admin/announcements/" + an["id"], "revoke again", 409, "leader")
t("DELETE", "/api/admin/announcements/nope", "revoke unknown", 404, "leader")
t("POST", "/api/admin/announcements", "member cannot", 403, "member", body={"message": "x", "ends_at": "2036-05-02T12:00:00+00:00"})
t("GET", "/api/admin/nearby?include_inactive=true", "list", 200, "leader", check=lambda d: len(d["nearby_units"]) >= 19)
t("POST", "/api/admin/nearby", "bad unit_type", 422, "leader", body={"unit_type": "trop", "unit_number": "1"})
row = t("POST", "/api/admin/nearby", "create", 201, "leader", body={"unit_type": "pack", "unit_number": "ZZAPI", "town": "Testville"}, check=lambda d: d["verified_at"])
t("PATCH", "/api/admin/nearby/" + row["id"], "update", 200, "leader", body={"town": "Testburg"}, check=lambda d: d["town"] == "Testburg")
t("PATCH", "/api/admin/nearby/" + row["id"], "unknown field rejected", 422, "leader", body={"unit_typo": "x"})
t("DELETE", "/api/admin/nearby/" + row["id"], "deactivate", 200, "leader", check=lambda d: d["active"] == 0)
t("DELETE", "/api/admin/nearby/" + row["id"], "deactivate again", 409, "leader")
t("PATCH", "/api/admin/nearby/nope", "unknown", 404, "leader", body={"town": "x"})
t("GET", "/api/admin/units", "pack leader sees only the pack", 200, "leader", check=lambda d: [u["slug"] for u in d["units"]] == ["pack73"])
t("GET", "/api/admin/units", "admin sees both", 200, "claude", check=lambda d: len(d["units"]) == 2)
t("PATCH", "/api/admin/units/troop73", "pack leader cannot edit the troop", 403, "leader", body={"meets_time": "19:30"})
t("PATCH", "/api/admin/units/pack73", "bad time", 422, "leader", body={"meets_time": "25:00"})
t("PATCH", "/api/admin/units/pack73", "no-change save", 200, "leader", body={"meets_weekday": 2, "meets_time": "18:00"}, check=lambda d: d["meets_time"] == "18:00")
t("PATCH", "/api/admin/units/crew99", "unknown", 404, "claude", body={"meets_time": "18:00"})
t("GET", "/api/admin/settings", "leader cannot", 403, "leader")
t("GET", "/api/admin/settings", "admin", 200, "claude", check=lambda d: {s["key"] for s in d["settings"]} >= {"api_keys_from", "nearby_source_url"})
t("PUT", "/api/admin/settings/api_keys_from", "bad value", 422, "claude", body={"value": "vpn"})
t("PUT", "/api/admin/settings/api_keys_from", "set", 200, "claude", body={"value": "anywhere"}, check=lambda d: d["value"] == "anywhere")
t("PUT", "/api/admin/settings/api_keys_from", "clear to default", 200, "claude", body={"value": None}, check=lambda d: d["value"] == "lan")
t("PUT", "/api/admin/settings/nope", "unknown key", 422, "claude", body={"value": "x"})
t("GET", "/api/admin/keys", "admin token cannot own keys", 403, headers={"X-Admin-Token": "testtoken"})
t("GET", "/api/admin/keys", "own list", 200, "claude", check=lambda d: "scopable" in d)
t("POST", "/api/admin/keys", "unscopable scope", 422, "claude", body={"label": "x", "scopes": ["apikeys:own"]})
k = t("POST", "/api/admin/keys", "mint", 201, "claude", body={"label": "zz api test", "scopes": ["leads:read"], "expires_days": 1}, check=lambda d: d["key"].startswith("gls73_"))
t("GET", "/api/admin/leads?limit=1", "bearer key works", 200, token=k["key"])
t("GET", "/api/admin/nearby", "bearer key out of scope", 403, token=k["key"])
t("GET", "/api/admin/leads?limit=1", "bearer key from WAN while lan", 403, token=k["key"], headers={"X-Forwarded-For": "108.36.248.87"})
t("POST", "/api/admin/keys", "a key cannot mint keys", 403, token=k["key"], body={"label": "x", "scopes": ["leads:read"]})
t("DELETE", "/api/admin/keys/" + k["id"], "revoke", 200, "claude", check=lambda d: d["state"] == "revoked")
t("GET", "/api/admin/leads?limit=1", "revoked key dead", 401, token=k["key"])
t("DELETE", "/api/admin/keys/" + k["id"], "revoke again", 409, "claude")
t("DELETE", "/api/admin/keys/nope", "foreign/unknown key", 404, "claude")
t("GET", "/api/admin/history", "leader cannot", 403, "leader")
t("GET", "/api/admin/history?kind=nearby.&limit=5", "admin, filtered", 200, "claude", check=lambda d: all(e["kind"].startswith("nearby.") for e in d["events"]) and d["events"])
t("GET", "/api/admin/people", "leader cannot", 403, "leader")
t("GET", "/api/admin/people", "admin", 200, "claude", check=lambda d: d["grantable"]["global"] == ["admin"] and "me" in d)
t("POST", "/api/admin/people/invite", "bad email", 422, "claude", body={"email": "nope", "units": [{"unit_id": S["pack"], "role": "member"}]})
t("POST", "/api/admin/people/invite", "admin cannot grant owner", 403, "claude", body={"email": "zz.new@example.test", "global_role": "owner"})
inv = t("POST", "/api/admin/people/invite", "invite", 201, "claude", body={"email": "zz.new@example.test", "units": [{"unit_id": S["pack"], "role": "member"}]}, check=lambda d: "/invite/" in d["url"])
t("DELETE", "/api/admin/people/invite/" + inv["id"], "revoke invite", 200, "claude")
t("DELETE", "/api/admin/people/invite/" + inv["id"], "revoke again", 404, "claude")
t("PUT", "/api/admin/people/%s/roles" % S["member_id"], "admin cannot change roles (owner only)", 403, "claude", body={"units": []})
t("PUT", "/api/admin/people/%s/roles" % S["member_id"], "owner sets roles", 200, "owner", body={"global_role": None, "units": [{"unit_id": S["troop"], "role": "member"}]}, check=lambda d: [m["slug"] for m in d["memberships"]] == ["troop73"])
t("POST", "/api/admin/people/%s/disable" % S["member_id"], "admin cannot disable", 403, "claude", body={})
t("POST", "/api/admin/people/%s/disable" % S["member_id"], "owner disables", 200, "owner", body={"reason": "api test"}, check=lambda d: d["disabled_at"])
t("GET", "/api/admin/whoami", "disabled person's session is dead", 401, "member")
t("POST", "/api/admin/people/%s/enable" % S["member_id"], "owner enables", 200, "owner")
S["member"] = subprocess.check_output(["docker", "exec", "sw-test", "python3", "-c", "import identity; print(identity.start_session('%s'))" % S["member_id"]]).decode().strip().splitlines()[-1]
t("GET", "/api/admin/whoami", "fresh member session after enable", 200, "member")
t("POST", "/api/admin/people/%s/reset" % S["leader_id"], "admin mints a reset link", 201, "claude", check=lambda d: "/reset/" in d["url"])
t("POST", "/api/admin/people/%s/reset" % S["claude_id"], "owner may reset an admin", 201, "owner")
t("POST", "/api/admin/people/nope/reset", "unknown person", 404, "claude")
t("GET", "/api/admin/roster", "member cannot", 403, "member")
t("POST", "/api/admin/roster/households", "no parent", 422, "leader", body={"email": "x@y.test"})
hh = t("POST", "/api/admin/roster/households", "create family", 201, "leader", body={"parent_name": "ZZ Api Family", "email": "zzfam@example.test"})
if lid:
t("POST", "/api/admin/roster/households", "import lead", 201, "leader", body={"lead_id": lid}, check=lambda d: d["source_lead_id"] == lid)
t("POST", "/api/admin/roster/households", "import twice", 409, "leader", body={"lead_id": lid})
sc = t("POST", "/api/admin/roster/households/%s/scouts" % hh["id"], "add scout", 201, "leader", body={"first_name": "Sam", "unit_id": S["pack"], "den": "Bear", "bsa_member_id": "1234567"})
t("POST", "/api/admin/roster/households/%s/scouts" % hh["id"], "bad bsa id", 422, "leader", body={"first_name": "X", "unit_id": S["pack"], "bsa_member_id": "12a"})
t("PATCH", "/api/admin/roster/scouts/" + sc["id"], "edit scout", 200, "leader", body={"den": "Webelos"}, check=lambda d: d["den"] == "Webelos")
t("PUT", "/api/admin/roster/scouts/%s/checks/dues" % sc["id"], "mark dues", 200, "leader", body={"done": True}, check=lambda d: d["done_at"])
t("PUT", "/api/admin/roster/scouts/%s/checks/dob" % sc["id"], "bad item", 422, "leader", body={"done": True})
t("GET", "/api/admin/roster?unit=pack73", "roster by unit", 200, "leader", check=lambda d: any(h["id"] == hh["id"] for h in d["households"]))
t("GET", "/api/admin/roster/households/" + hh["id"], "one family", 200, "leader", check=lambda d: d["scouts"][0]["checks"]["dues"]["done_by"] == "zz.leader@example.test")
t("PUT", "/api/admin/roster/households/%s/people" % hh["id"], "leader cannot link accounts", 403, "leader", body={"person_ids": [S["member_id"]]})
t("PUT", "/api/admin/roster/households/%s/people" % hh["id"], "unknown person id", 422, "claude", body={"person_ids": ["nope"]})
t("PUT", "/api/admin/roster/households/%s/people" % hh["id"], "admin links the member", 200, "claude", body={"person_ids": [S["member_id"]]})
t("GET", "/api/admin/family", "member sees own family", 200, "member", check=lambda d: d["households"][0]["parent_name"] == "ZZ Api Family")
t("GET", "/api/admin/family", "unlinked account sees none", 200, "leader", check=lambda d: d["households"] == [])
t("PATCH", "/api/admin/roster/households/" + hh["id"], "deactivate family", 200, "leader", body={"active": False})
t("GET", "/api/admin/family", "inactive family drops off", 200, "member", check=lambda d: d["households"] == [])
t("GET", "/api/admin/family", "anon", 401)
t("GET", "/api/admin/calendar", "member cannot", 403, "member")
t("GET", "/api/admin/calendar", "list", 200, "leader", check=lambda d: d["configured"] and len(d["events"]) >= 32 and all(e["mine"] for e in d["events"]))
t("POST", "/api/admin/calendar", "no title", 422, "leader", body={"date": "2036-06-06"})
ev = t("POST", "/api/admin/calendar", "create (2036)", 201, "leader", body={"title": "ZZ api probe", "unit": "troop", "date": "2036-06-06", "time": "18:30"}, check=lambda d: d["uid"].endswith("@site73.greenlanescouts73.org"))
t("PUT", "/api/admin/calendar/" + ev["uid"], "replace", 200, "leader", body={"title": "ZZ api probe moved", "unit": "pack", "date": "2036-06-07"})
t("PUT", "/api/admin/calendar/x@band.us", "foreign uid", 403, "leader", body={"title": "x", "date": "2036-06-07"})
t("DELETE", "/api/admin/calendar/" + ev["uid"], "delete", 200, "leader", check=lambda d: d["deleted"])
t("DELETE", "/api/admin/calendar/" + ev["uid"], "delete again", 404, "leader")
t("GET", "/api/admin/fbposts", "member cannot", 403, "member")
t("POST", "/api/admin/fbposts/ingest", "leader cannot ingest", 403, "leader", body={"id": "pack-73/x", "status": "scheduled"})
data = png(320, 200); sha = hashlib.sha256(data).hexdigest()
t("POST", "/api/admin/fbposts/ingest", "hash mismatch refused", 422, "claude", body={"id": "pack-73/zz-api", "status": "scheduled", "image": {"b64": base64.b64encode(data).decode(), "mime": "image/png", "sha256": "bad"}})
t("POST", "/api/admin/fbposts/ingest", "ingest with image", 200, "claude", body={"id": "pack-73/zz-api", "unit": "pack-73", "status": "scheduled", "fb_post_id": "141826306647186_ZZAPI", "message": "api test", "scheduled_for": "2036-09-10T12:00:00+00:00", "cancel_url": "https://scout-control.thewichersfamily.com/cancel?id=141826306647186_ZZAPI&sig=nope", "image": {"b64": base64.b64encode(data).decode(), "mime": "image/png", "sha256": sha}}, check=lambda d: d["image_sha256"] == sha)
t("GET", "/api/admin/fbposts?include_done=false", "listed as scheduled", 200, "leader", check=lambda d: any(p["id"] == "pack-73/zz-api" and p["state"] == "scheduled" for p in d["posts"]))
t("GET", "/api/admin/fbposts/pack-73/zz-api/image", "image anon", 401)
t("GET", "/api/admin/fbposts/pack-73/zz-api/image", "image gated", 200, "leader", check=lambda d: d[:8] == b"\x89PNG\r\n\x1a\n")
t("POST", "/api/admin/fbposts/pack-73/zz-api/cancel", "cancel: publisher refuses a bad signature", 502, "leader")
t("POST", "/api/admin/fbposts/pack-73/zz-api/reschedule", "reschedule: time too soon", 422, "leader", body={"message": "x", "scheduled_for": "2020-01-01T00:00:00+00:00"})
t("POST", "/api/admin/fbposts/pack-73/zz-api/reschedule", "reschedule: publisher refuses a bad signature", 502, "leader", body={"message": "x", "scheduled_for": "2036-09-11T12:00:00+00:00"})
t("POST", "/api/admin/fbposts/ingest", "past time reads as published", 200, "claude", body={"id": "pack-73/zz-old", "status": "scheduled", "scheduled_for": "2020-01-01T00:00:00+00:00"})
t("GET", "/api/admin/fbposts?include_done=false", "published dropped from open list", 200, "leader", check=lambda d: not any(p["id"] == "pack-73/zz-old" for p in d["posts"]))
t("POST", "/api/admin/fbposts/pack-73/zz-old/cancel", "cancel after time passed", 409, "leader")
json.dump(R, open(os.path.join(D, "results.json"), "w"))
fails = [r for r in R if not r[5]]
print("%d checks, %d failed" % (len(R), len(fails)))
for r in fails: print(" FAIL", r[0], r[1], "|", r[2], "| expected", r[3], "got", r[4], r[6])