thethreemagi
d1dfa6df90
Moved to app/API.md so the build context carries it. The console renders it at /leaders/api. tests/smoke_admin.py 162 -> 164.
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. Output goes to app/API.md so it ships in the image and the console can render it.
|
|
Calendar checks write 2036 probes to the REAL Radicale store and delete them.
|
|
Writes results to /tmp/apitest/results.json; app/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])
|