thethreemagi
38bedaea88
48 routes plus /api/docs, 123 checks over real HTTP against a throwaway site on a copy of the live database, as anonymous, a member, a pack leader, an admin (the claude account), the owner, and the break-glass token from the LAN and from outside. Every refusal path the routes promise was exercised: 401, 403 by capability and by unit scope, 404, 409 state conflicts, 422 validation, and 502 when the publisher refuses a bad signature. Calendar probes went to the real Radicale store dated 2036 and were deleted; the store ended with 32 objects and none in the site namespace. API.md is generated from the router registry plus those results, one entry per route with its own description and what was tried. No route failed; all 48 approved.
222 lines
18 KiB
Python
222 lines
18 KiB
Python
"""Drive every admin API route over real HTTP. NOT a unit test: it needs a running
|
|
throwaway site on a COPY of the database (container name sw-test on arrstack_arr_net,
|
|
reachable at the URL in /tmp/apitest/base) and sessions for four accounts in
|
|
/tmp/apitest/setup.json - see projects/scout-website/state.md for the recipe.
|
|
Calendar checks write 2036 probes to the REAL Radicale store and delete them.
|
|
Writes results to /tmp/apitest/results.json; API.md is generated from that plus the
|
|
router registry. 123 checks on 2026-09-04, all passing."""
|
|
|
|
import json, urllib.request, urllib.error, urllib.parse, hashlib, struct, zlib, base64, datetime, sys
|
|
B = open("/tmp/apitest/base").read().strip()
|
|
S = json.load(open("/tmp/apitest/setup.json"))
|
|
R = []
|
|
|
|
def call(method, path, who=None, body=None, raw=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), ct
|
|
except urllib.error.HTTPError as e:
|
|
ct = e.headers.get("Content-Type", ""); b = e.read()
|
|
try: return e.code, json.loads(b), ct
|
|
except Exception: return e.code, b, ct
|
|
|
|
def t(method, path, case, expect, who=None, body=None, token=None, headers=None, check=None):
|
|
st, d, ct = call(method, path, who, body, token=token, headers=headers)
|
|
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"")
|
|
|
|
# ---- whoami / docs ----
|
|
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)
|
|
|
|
# ---- summary / leads ----
|
|
t("GET", "/api/admin/summary", "anon", 401)
|
|
t("GET", "/api/admin/summary", "member", 403, "member")
|
|
summ = 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")
|
|
|
|
# ---- announcements ----
|
|
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"})
|
|
|
|
# ---- nearby ----
|
|
nb = 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"})
|
|
|
|
# ---- units ----
|
|
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"})
|
|
|
|
# ---- settings ----
|
|
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"})
|
|
|
|
# ---- keys ----
|
|
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")
|
|
|
|
# ---- history ----
|
|
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"])
|
|
|
|
# ---- people ----
|
|
t("GET", "/api/admin/people", "leader cannot", 403, "leader")
|
|
pp = 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")
|
|
# disabling ended the member's sessions (correct); mint a fresh one for the rest of the run
|
|
import subprocess
|
|
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 works", 200, "member")
|
|
# disabling ended the member's sessions (correct); mint a fresh one for the rest of the run
|
|
import subprocess
|
|
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", "member: fresh 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")
|
|
|
|
# ---- roster / family ----
|
|
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:
|
|
imp = 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)
|
|
|
|
# ---- calendar (real store, 2036) ----
|
|
t("GET", "/api/admin/calendar", "member cannot", 403, "member")
|
|
cal = 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")
|
|
|
|
# ---- fbposts ----
|
|
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"}})
|
|
fp = 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("/tmp/apitest/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])
|