thethreemagi/scout-website
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ea22ed53e7ba253b6c2a6e193246f010f81b864f
scout-website/tests/smoke_admin.py
T
thethreemagi ea22ed53e7 fb_posts: past its publish time, a scheduled post reads as published 
Facebook does not call back and the reconciliation was never built, so a
row that stayed scheduled forever was a lie of omission. Decided by Mike:
once the time passes with nothing else reported, the post is published.
status keeps what the publisher last said; state is what a reader acts
on, and the open-only list and cancel use state. Cancel on a post whose
time has passed says so.
2026-09-04 19:53:08 -04:00

435 lines
29 KiB
Python
Raw Blame History

"""
smoke_admin.py - the P2 admin write paths against a throwaway DB.
Runs in-process with no container, no network and no dependencies beyond the
stdlib, so it can be run before anything is committed.
STORE_DB=/tmp/x.db python3 tests/smoke_admin.py
It covers the rules that are expensive to get wrong and invisible when they
are: nearby rows bump verified_at on every save and deactivate rather than
delete, unit edits are meeting fields only and unit-scoped, and settings
accept only registered keys and fall back to the code default on anything
absent or mangled.
"""
import os, sys, tempfile
DB = os.environ.get("STORE_DB") or os.path.join(tempfile.mkdtemp(), "smoke.db")
os.environ["STORE_DB"] = DB
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "app"))
import identity as I
import nearby as N
import store as S
PASS = FAIL = 0
def check(label, cond):
global PASS, FAIL
if cond:
PASS += 1
print(" ok %s" % label)
else:
FAIL += 1
print(" FAIL %s" % label)
def raises(label, status, exc, fn, *a, **kw):
try:
fn(*a, **kw)
except exc as e:
check("%s -> %d" % (label, status), e.status == status)
return
except Exception as e:
check("%s -> %d (got %r)" % (label, status, e), False)
return
check("%s -> %d (no error raised)" % (label, status), False)
print("db: %s\n" % DB)
S.init()
I.init()
print("nearby: writes are validated")
raises("bad unit_type", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "trop", "unit_number": "1"})
raises("missing unit_number", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack"})
raises("bad serves", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1", "serves": "everyone"})
raises("http link", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1", "link_url": "http://x.test"})
raises("link with an attribute breakout", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1",
"link_url": 'https://x.test" onmouseover="alert(1)'})
raises("contact without @", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1", "contact": "call Steve"})
raises("contact with an attribute breakout", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1",
"contact": 'a@b.test" onfocus="alert(1)'})
check("renderer escapes quotes as a second line", N._esc('a"b') == "a"b")
raises("unknown field rejected, not dropped", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1", "unit_typo": "pack"})
raises("verified_at must be a date", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1", "verified_at": "yesterday"})
raises("sort_order must be an int", 422, S.NearbyRejected, S.create_nearby,
{"unit_type": "pack", "unit_number": "1", "sort_order": "soon"})
print("\nnearby: create, update, verified_at")
a = S.create_nearby({"unit_type": "pack", "unit_number": "244", "town": "Harleysville",
"area": "North Penn", "serves": "family"})
check("row created", a and a["unit_number"] == "244")
check("active and sort_order defaulted", a["active"] == 1 and a["sort_order"] == 100)
check("verified_at defaults to today", a["verified_at"] == S._today())
b = S.create_nearby({"unit_type": "troop", "unit_number": "27", "area": "North Penn",
"verified_at": "2026-08-01"})
check("explicit verified_at kept", b["verified_at"] == "2026-08-01")
b2 = S.update_nearby(b["id"], {"town": "Lansdale"})
check("partial update lands", b2["town"] == "Lansdale" and b2["unit_number"] == "27")
check("saving bumps verified_at", b2["verified_at"] == S._today())
b3 = S.update_nearby(b["id"], {"notes": "meets in the annexe", "verified_at": "2026-08-15"})
check("explicit verified_at wins on update", b3["verified_at"] == "2026-08-15")
check("update of missing row is None", S.update_nearby("nope", {"town": "x"}) is None)
raises("empty update", 422, S.NearbyRejected, S.update_nearby, b["id"], {})
print("\nnearby: deactivate, never delete")
check("deactivate", S.deactivate_nearby(a["id"]))
check("second deactivate is a no-op", not S.deactivate_nearby(a["id"]))
check("row survives", S.get_nearby(a["id"])["active"] == 0)
check("default list hides it", all(r["id"] != a["id"] for r in S.list_nearby()))
check("include_inactive shows it",
any(r["id"] == a["id"] for r in S.list_nearby(include_inactive=True)))
back = S.update_nearby(a["id"], {"active": 1})
check("reactivate via update", back["active"] == 1)
groups = N.listing()
check("public page groups the live rows",
len(groups) == 1 and groups[0][0] == "North Penn" and len(groups[0][1]) == 2)
print("\nunits: meeting fields only, structured")
pack = I.get_unit("pack73")
check("seed row present", pack and pack["meets_weekday"] == 2)
u = I.update_unit_meets("pack73", {"meets_time": "18:30"})
check("time updated", u["meets_time"] == "18:30")
u = I.update_unit_meets(pack["id"], {"meets_weekday": 4, "meets_at": None})
check("update by id, null allowed", u["meets_weekday"] == 4 and u["meets_at"] is None)
check("other fields untouched", u["display_name"] == pack["display_name"])
raises("weekday 8", 422, I.IdentityError, I.update_unit_meets, "pack73", {"meets_weekday": 8})
raises("weekday as bool", 422, I.IdentityError, I.update_unit_meets, "pack73", {"meets_weekday": True})
raises("display time string", 422, I.IdentityError, I.update_unit_meets, "pack73", {"meets_time": "7pm"})
raises("renaming is not a meeting edit", 422, I.IdentityError, I.update_unit_meets,
"pack73", {"display_name": "Pack 99"})
raises("unknown unit", 404, I.IdentityError, I.update_unit_meets, "pack99", {"meets_time": "18:00"})
raises("nothing to update", 422, I.IdentityError, I.update_unit_meets, "pack73", {})
print("\nsettings: typed keys, default fallback")
check("default when unset",
I.get_setting("nearby_source_name") == "Continental District unit list")
s = I.set_setting("nearby_source_url", "https://example.test/units", actor="a@example.test")
check("set and read back", I.get_setting("nearby_source_url") == "https://example.test/units")
check("set_setting reports itself", s["is_set"] and s["updated_by"] == "a@example.test")
raises("unknown key", 422, I.IdentityError, I.set_setting, "nearby_src_url", "https://x.test")
raises("blank value", 422, I.IdentityError, I.set_setting, "nearby_source_name", " ")
raises("http url", 422, I.IdentityError, I.set_setting, "nearby_source_url", "http://x.test")
cleared = I.set_setting("nearby_source_url", None)
check("null clears to default", not cleared["is_set"]
and I.get_setting("nearby_source_url") == "https://tinyurl.com/ContinentalScouts")
con = I.connect()
con.execute("INSERT INTO settings (key, value, updated_at) VALUES (?,?,?)",
("nearby_source_url", "ftp://mangled", I._now()))
con.commit(); con.close()
check("mangled row falls back to default",
I.get_setting("nearby_source_url") == "https://tinyurl.com/ContinentalScouts")
check("all_settings covers the registry",
{s["key"] for s in I.all_settings()} == set(I.SETTINGS_KEYS))
print("\ncapabilities behind the new routes")
leader = {"memberships": [{"unit_id": "u1", "role": "leader"}]}
member = {"memberships": [{"unit_id": "u1", "role": "member"}]}
admin = {"global_role": "admin", "memberships": []}
check("leader can edit nearby", I.can(leader, "nearby:write"))
check("member cannot", not I.can(member, "nearby:write"))
check("leader edits own unit", I.can(leader, "unit:write_own", "u1"))
check("but not the other one", not I.can(leader, "unit:write_own", "u2"))
check("leader cannot touch settings", not I.can(leader, "settings:write"))
check("admin spans units", I.can(admin, "unit:write_own", "u1") and I.can(admin, "unit:write_own", "u2"))
check("admin holds settings:write", I.can(admin, "settings:write"))
print("\nlan rule")
import admin_api as A
class _Req:
def __init__(self, ip, cookie=None, bearer=None):
self.headers = {"x-forwarded-for": ip} if ip else {}
if bearer: self.headers["authorization"] = "Bearer " + bearer
self.cookies = {"s73_session": cookie} if cookie else {}
self.client = None
for ip, want in (("10.0.0.55", True), ("10.0.1.20", True), ("127.0.0.1", True), ("172.19.0.31", True),
("108.36.248.87", False), ("192.168.1.9", False), ("", False), ("garbage", False)):
check("client_is_lan(%r) is %s" % (ip, want), A.client_is_lan(_Req(ip)) == want)
check("forged first hop does not make an outsider LAN: last hop wins",
A.client_is_lan(_Req("10.0.0.1, 108.36.248.87")) is False)
check("and a forged outside hop does not make a LAN caller outside",
A.client_is_lan(_Req("203.0.113.9, 10.0.0.55")) is True)
check("api_keys_from defaults to lan", I.get_setting("api_keys_from") == "lan")
raises("api_keys_from rejects other values", 422, I.IdentityError, I.set_setting, "api_keys_from", "vpn")
# a key from outside is refused while lan, allowed once anywhere, admin token never from outside
A.ADMIN_TOKEN = "t"
import uuid as _uuid
con = I.connect()
pid = str(_uuid.uuid4())
con.execute("INSERT INTO people (id, email, full_name, global_role, created_at) VALUES (?,?,?,?,?)",
(pid, "lanrule@example.test", "Lan Rule", "admin", I._now())); con.commit(); con.close()
person = I.get_person(pid)
full, _row = I.mint_api_key(person, "outside", ["leads:read"])
def _hits(req, cap="leads:read"):
try:
A._auth(req, None, cap); return 200
except Exception as e:
return getattr(e, "status_code", 500)
check("key from the LAN passes", _hits(_Req("10.0.0.55", bearer=full)) == 200)
check("key from outside is 403 while lan", _hits(_Req("108.36.248.87", bearer=full)) == 403)
I.set_setting("api_keys_from", "anywhere", actor="test")
check("key from outside passes once anywhere", _hits(_Req("108.36.248.87", bearer=full)) == 200)
check("key still cannot exceed its scopes from anywhere", _hits(_Req("108.36.248.87", bearer=full), "nearby:write") == 403)
I.set_setting("api_keys_from", None, actor="test")
check("clearing the setting restores lan", I.get_setting("api_keys_from") == "lan"
and _hits(_Req("108.36.248.87", bearer=full)) == 403)
tok = I.start_session(pid)
check("a session from outside is never restricted", _hits(_Req("108.36.248.87", cookie=tok)) == 200)
def _tok(req):
try:
A._auth(req, "t", "leads:read"); return 200
except Exception as e:
return getattr(e, "status_code", 500)
check("admin token from the LAN passes", _tok(_Req("10.0.0.55")) == 200)
check("admin token from outside is 403, regardless of the setting", _tok(_Req("108.36.248.87")) == 403)
print("\ncalendar write-back")
import calendar_write as C
raises("title required", 422, C.CalendarRejected, C.clean, {"date": "2026-10-03"})
raises("date required", 422, C.CalendarRejected, C.clean, {"title": "x"})
raises("bad unit", 422, C.CalendarRejected, C.clean, {"title": "x", "date": "2026-10-03", "unit": "den"})
raises("end before start", 422, C.CalendarRejected, C.clean, {"title": "x", "date": "2026-10-03", "end": "2026-10-02"})
raises("bad time", 422, C.CalendarRejected, C.clean, {"title": "x", "date": "2026-10-03", "time": "1pm"})
raises("end_time without time", 422, C.CalendarRejected, C.clean, {"title": "x", "date": "2026-10-03", "end_time": "14:00"})
raises("end_time before time same day", 422, C.CalendarRejected, C.clean, {"title": "x", "date": "2026-10-03", "time": "14:00", "end_time": "13:00"})
ev = C.clean({"title": "Pack Hike; again", "unit": "Pack", "date": "2026-10-03", "end": "2026-10-03", "time": "13:00",
"location": "Foy Park", "badge": "", "description": "Bring water,\nand a hat"})
check("clean normalises: unit lower, same-day end dropped, blank badge absent",
ev["unit"] == "pack" and ev["end"] is None and ev["badge"] is None and ev["time"] == "13:00")
uid = "t1" + C.UID_SUFFIX
ics = C.build_ics(uid, ev, stamp="20260904T000000Z").decode()
check("timed event carries VTIMEZONE and TZID start, 90-minute default end",
"BEGIN:VTIMEZONE" in ics and "DTSTART;TZID=America/New_York:20261003T130000" in ics
and "DTEND;TZID=America/New_York:20261003T143000" in ics)
check("text is escaped and newlines encoded", "SUMMARY:Pack Hike\\; again" in ics and "DESCRIPTION:Bring water\\,\\nand a hat" in ics)
check("unit is CATEGORIES, crlf line ends", "CATEGORIES:pack" in ics and ics.endswith("END:VCALENDAR\r\n")
and ics.count("\n") == ics.count("\r\n"))
allday = C.build_ics(uid, C.clean({"title": "Fall Campout", "unit": "both", "date": "2026-10-17", "end": "2026-10-18", "badge": "OVERNIGHT"}), stamp="20260904T000000Z").decode()
check("all-day multi-day: VALUE=DATE with exclusive end, badge, no VTIMEZONE",
"DTSTART;VALUE=DATE:20261017" in allday and "DTEND;VALUE=DATE:20261019" in allday
and "X-SCOUT73-BADGE:OVERNIGHT" in allday and "VTIMEZONE" not in allday)
multi = C.build_ics(uid, C.clean({"title": "x", "date": "2026-10-17", "end": "2026-10-18", "time": "16:00"}), stamp="20260904T000000Z").decode()
check("timed multi-day with no end_time ends at noon on the end date, as seed.py did",
"DTEND;TZID=America/New_York:20261018T120000" in multi)
longt = C.build_ics(uid, C.clean({"title": "A" * 120, "date": "2026-10-03"}), stamp="20260904T000000Z").decode()
check("long lines are folded at 75 octets", all(len(l.encode()) <= 75 for l in longt.split("\r\n")))
check("ownership by suffix", C.owns("x" + C.UID_SUFFIX) and not C.owns("abc@greenlanescouts73.org")
and not C.owns("x@band.us") and not C.owns(None))
check("new uids are owned and unique", C.owns(C.new_uid()) and C.new_uid() != C.new_uid())
check("slug is stable and marked", C.slug("a" + C.UID_SUFFIX).startswith("site-") and C.slug("a" + C.UID_SUFFIX) == C.slug("a" + C.UID_SUFFIX))
# fail closed and never touch a foreign uid, with the transport watched
calls = []
class _Resp:
status = 201
def __enter__(self): return self
def __exit__(self, *a): return False
def _fake(req, timeout=None):
calls.append((req.get_method(), req.full_url, req.get_header("Authorization") is not None)); return _Resp()
C._urlopen = _fake
C.RADICALE_URL = ""; C.RADICALE_USER = ""; C.RADICALE_PASS = ""
raises("unconfigured put is 503", 503, C.CalendarRejected, C.put_event, uid, ev)
check("and nothing was sent", calls == [])
C.RADICALE_URL = "http://radicale.test/scouts/site73/"; C.RADICALE_USER = "scoutsite"; C.RADICALE_PASS = "p"
raises("foreign uid put is 403", 403, C.CalendarRejected, C.put_event, "abc@greenlanescouts73.org", ev)
raises("foreign uid delete is 403", 403, C.CalendarRejected, C.delete_event, "x@band.us")
check("still nothing sent for foreign uids", calls == [])
check("owned put goes to the slug with auth", C.put_event(uid, ev) == 201 and calls[-1][0] == "PUT"
and calls[-1][1] == "http://radicale.test/scouts/site73/" + C.slug(uid) and calls[-1][2])
check("owned delete", C.delete_event(uid) == 201 and calls[-1][0] == "DELETE")
print("\nseeded events are the site's too")
check("both namespaces owned, nothing else", C.owns("x" + C.UID_SUFFIX) and C.owns("site73-abc123@greenlanescouts73.org")
and not C.owns("abc@greenlanescouts73.org") and not C.owns("x@band.us") and not C.owns("site73-x@site73.example"))
check("object name per namespace", C.object_name("a" + C.UID_SUFFIX) == C.slug("a" + C.UID_SUFFIX)
and C.object_name("site73-abc@greenlanescouts73.org") == "site73-abc%40greenlanescouts73.org.ics")
calls.clear(); C.put_event("site73-abc@greenlanescouts73.org", ev)
check("seeded put goes to the seeder's object name", calls[-1][1].endswith("/site73-abc%40greenlanescouts73.org.ics"))
print("\naction log")
class _Sess:
def __init__(self, tok):
self.headers = {"x-forwarded-for": "10.0.0.5"}; self.cookies = {"s73_session": tok}; self.client = None
req = _Sess(I.start_session(pid))
n0 = len(I.list_events(limit=500))
A.update_unit(req, "pack73", {"meets_time": "18:15"}, None)
A.update_unit(req, "pack73", {"meets_time": "18:00"}, None)
A.put_setting(req, "nearby_source_name", {"value": "Test source"}, None)
A.put_setting(req, "nearby_source_name", {"value": None}, None)
nb = A.create_nearby(req, {"unit_type": "pack", "unit_number": "ZZ1", "town": "A"}, None)
A.update_nearby(req, nb["id"], {"town": "B"}, None)
A.deactivate_nearby(req, nb["id"], None)
an = A.create_announcement(req, {"message": "log me", "ends_at": "2036-01-02T00:00:00+00:00"}, None)
A.revoke_announcement(req, an["id"], None)
ev_rows = I.list_events(limit=500)
kinds = [e["kind"] for e in ev_rows[:len(ev_rows) - n0]]
check("every P2 write lands in the log", set(kinds) >= {"unit.updated", "setting.updated", "nearby.created",
"nearby.updated", "nearby.deactivated", "announcement.created", "announcement.revoked"})
check("attributed to the person", all(e["actor_email"] == "lanrule@example.test" for e in ev_rows[:len(ev_rows) - n0]))
unit_ev = [e for e in ev_rows if e["kind"] == "unit.updated"][-1]
check("diff is before -> after", "-> '18:15'" in unit_ev["detail"] and "meets_time:" in unit_ev["detail"])
nb_ev = [e for e in ev_rows if e["kind"] == "nearby.updated"][0]
check("nearby diff names the field", "town: 'A' -> 'B'" in nb_ev["detail"])
set_ev = [e for e in ev_rows if e["kind"] == "setting.updated"][0]
check("clearing a setting is said so", "cleared to default" in set_ev["detail"])
check("kind prefix filter and limit", all(e["kind"].startswith("nearby.") for e in I.list_events(kind_prefix="nearby."))
and len(I.list_events(limit=2)) == 2)
check("history is admin and above, and not scopable on a key", "history:read" in I.CAPS["admin"]
and "history:read" not in I.CAPS["leader"] and "history:read" in I.KEY_UNSCOPABLE)
print("\nlead claims")
lid = S.record_lead({"parent_name": "Claim Test", "email": "claim@example.test", "interested_in": "Pack 73"}) if hasattr(S, "record_lead") else None
if lid is None:
con = S.connect(); lid = "lead-claim-test"
con.execute("INSERT INTO join_leads (id, submitted_at, recorded_at, parent_name, email, payload) VALUES (?,?,?,?,?,?)",
(lid, I._now(), I._now(), "Claim Test", "claim@example.test", "{}")); con.commit(); con.close()
check("unclaimed count sees it", S.unclaimed_count() >= 1 and S.summary()["unclaimed"] >= 1)
c = S.claim_lead(lid, "p-a", "a@example.test")
check("claim recorded and listed on the lead", c["email"] == "a@example.test" and S.get_lead(lid)["claim"]["email"] == "a@example.test"
and [r for r in S.list_leads(limit=500) if r["id"] == lid][0]["claim"]["email"] == "a@example.test")
raises("claiming again yourself", 409, S.ClaimRejected, S.claim_lead, lid, "p-a", "a@example.test")
raises("someone else cannot take it over silently", 409, S.ClaimRejected, S.claim_lead, lid, "p-b", "b@example.test")
raises("someone else cannot release it", 403, S.ClaimRejected, S.release_lead, lid, "p-b", "b@example.test")
check("an owner can release it", S.release_lead(lid, "p-b", "b@example.test", can_release_any=True)["email"] == "a@example.test"
and S.get_lead(lid)["claim"] is None)
raises("release when nobody has it", 409, S.ClaimRejected, S.release_lead, lid, "p-a", "a@example.test")
S.claim_lead(lid, "p-b", "b@example.test"); S.release_lead(lid, "p-b", "b@example.test")
check("history keeps every claim", len(S.get_lead(lid)["claim_history"]) == 2 and all(h["released_at"] for h in S.get_lead(lid)["claim_history"]))
check("nothing on the lead itself changed", "status" not in S.get_lead(lid) and "claimed" not in S.get_lead(lid))
check("unknown lead is None", S.claim_lead("nope", "p", "e") is None and S.release_lead("nope", "p", "e") is None)
con = S.connect(); con.execute("DELETE FROM lead_claims WHERE lead_id=?", (lid,)); con.execute("DELETE FROM join_leads WHERE id=?", (lid,)); con.commit(); con.close()
print("\nroster")
pk = I.get_unit("pack73"); tr = I.get_unit("troop73")
raises("household needs a parent name", 422, S.RosterRejected, S.create_household, {"email": "x@y.test"})
raises("bad email", 422, S.RosterRejected, S.create_household, {"parent_name": "P", "email": "nope"})
hid = S.create_household({"parent_name": "Pat Parent", "email": "pat@example.test", "phone": "555"}, created_by="me")
check("household created", S.get_household(hid, "2026-2027")["parent_name"] == "Pat Parent")
raises("scout needs a name and unit", 422, S.RosterRejected, S.add_scout, hid, {"unit_id": pk["id"]})
raises("bsa id digits only", 422, S.RosterRejected, S.add_scout, hid, {"first_name": "Sam", "unit_id": pk["id"], "bsa_member_id": "12a"})
raises("unknown unit", 422, S.RosterRejected, S.add_scout, hid, {"first_name": "Sam", "unit_id": "nope"})
sc = S.add_scout(hid, {"first_name": "Sam", "last_name": "Parent", "unit_id": pk["id"], "den": "Bear", "bsa_member_id": "1234567"})
sc2 = S.add_scout(hid, {"first_name": "Alex", "unit_id": tr["id"]})
check("two scouts, two units, one household", len(S.get_household(hid, "2026-2027")["scouts"]) == 2)
check("roster by unit filters households", [h["id"] for h in S.list_roster("2026-2027", pk["id"])] == [hid]
and S.list_roster("2026-2027", "no-such-unit") == [])
raises("check item vocabulary", 422, S.RosterRejected, S.set_check, sc["id"], "2026-2027", "dob", True)
c = S.set_check(sc["id"], "2026-2027", "dues", True, done_by="me", note="cash")
check("check recorded with who and when", c["done_at"] and c["done_by"] == "me" and c["note"] == "cash")
check("check visible on the roster for that year only",
S.get_household(hid, "2026-2027")["scouts"][1 if S.get_household(hid, "2026-2027")["scouts"][0]["first_name"] == "Alex" else 0]["checks"].get("dues")
and not any(s["checks"] for s in S.get_household(hid, "2027-2028")["scouts"]))
check("clearing a check removes it", S.set_check(sc["id"], "2026-2027", "dues", False)["done_at"] is None)
before, after = S.update_scout(sc["id"], {"den": "Webelos"})
check("scout update returns before and after", before["den"] == "Bear" and after["den"] == "Webelos")
check("scout deactivate hides from the default list", S.update_scout(sc2["id"], {"active": False})[1]["active"] == 0
and len([s for h in S.list_roster("2026-2027") if h["id"] == hid for s in h["scouts"]]) == 1)
con = S.connect(); con.execute("INSERT INTO join_leads (id, submitted_at, recorded_at, parent_name, email, phone, children, payload) VALUES (?,?,?,?,?,?,?,?)",
("lead-import-test", I._now(), I._now(), "Lee Lead", "lee@example.test", "555-1", "Kim, 7 and Jo, 9", "{}")); con.commit(); con.close()
h2 = S.import_lead("lead-import-test", created_by="me")
hh = S.get_household(h2, "2026-2027")
check("lead imported as a household, contact copied, children as a note, lead linked",
hh["parent_name"] == "Lee Lead" and hh["email"] == "lee@example.test" and "Kim, 7" in hh["notes"] and hh["source_lead_id"] == "lead-import-test")
raises("importing the same lead twice", 409, S.RosterRejected, S.import_lead, "lead-import-test")
check("the lead itself is untouched", S.get_lead("lead-import-test")["parent_name"] == "Lee Lead" and "household" not in S.get_lead("lead-import-test"))
check("no health data columns exist anywhere",
not any(c for t in ("households", "scouts", "roster_checks") for c in [x[1] for x in S.connect().execute("PRAGMA table_info(%s)" % t)]
if any(k in c for k in ("dob", "birth", "address", "medical", "health_data", "allerg"))))
check("roster:write is not scopable on a key", "roster:write" in I.KEY_UNSCOPABLE and "roster:write" in I.CAPS["leader"])
con = S.connect()
for t, k in (("roster_checks", "scout_id IN (SELECT id FROM scouts WHERE household_id IN (?,?))"), ("scouts", "household_id IN (?,?)"), ("households", "id IN (?,?)")):
con.execute("DELETE FROM %s WHERE %s" % (t, k), (hid, h2))
con.execute("DELETE FROM join_leads WHERE id='lead-import-test'"); con.commit(); con.close()
print("\nfamily link")
hid = S.create_household({"parent_name": "Link Parent"}, created_by="me")
S.add_scout(hid, {"first_name": "Lin", "unit_id": pk["id"], "den": "Wolf"})
check("no link, no families", S.households_for_person("p-x", "2026-2027") == [])
check("link set", S.set_household_people(hid, ["p-x", "p-y", "p-x"]) == ["p-x", "p-y"] and set(S.household_people(hid)) == {"p-x", "p-y"})
fam = S.households_for_person("p-x", "2026-2027")
check("a linked person sees the household with scouts and checks", len(fam) == 1 and fam[0]["scouts"][0]["first_name"] == "Lin" and "checks" in fam[0]["scouts"][0])
check("unlinked person sees nothing", S.households_for_person("p-z", "2026-2027") == [])
S.update_household(hid, {"active": False})
check("an inactive household drops off the family view", S.households_for_person("p-x", "2026-2027") == [])
check("unknown household is None", S.set_household_people("nope", ["p-x"]) is None)
con = S.connect(); con.execute("DELETE FROM household_people WHERE household_id=?", (hid,)); con.execute("DELETE FROM scouts WHERE household_id=?", (hid,)); con.execute("DELETE FROM households WHERE id=?", (hid,)); con.commit(); con.close()
print("\nfacebook posts")
import hashlib as _h, struct as _st, zlib as _z
def _png(w, h):
def chunk(t, d): return _st.pack(">I", len(d)) + t + d + _st.pack(">I", _z.crc32(t + d) & 0xffffffff)
return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", _st.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)) + chunk(b"IEND", b"")
png = _png(640, 480); sha = _h.sha256(png).hexdigest()
raises("id shape", 422, S.FbRejected, S.upsert_fb_post, {"id": "nope", "status": "scheduled"})
raises("status vocabulary", 422, S.FbRejected, S.upsert_fb_post, {"id": "pack-73/x", "status": "posted"})
raises("image hash mismatch refused", 422, S.FbRejected, S.upsert_fb_post, {"id": "pack-73/x", "status": "scheduled"}, (png, "image/png", "deadbeef"))
raises("image mime", 422, S.FbRejected, S.upsert_fb_post, {"id": "pack-73/x", "status": "scheduled"}, (png, "image/gif", sha))
r = S.upsert_fb_post({"id": "pack-73/2026-09-10-hike", "status": "scheduled", "page_id": "141", "fb_post_id": "141_9",
"message": "Hike Saturday", "scheduled_for": "2026-09-10T12:00:00+00:00",
"cancel_url": "https://x.test/cancel?id=141_9&sig=abc", "image_ref": "scouting-comms/generated/hike.png"},
(png, "image/png", sha))
check("row stored with content-addressed image and size", r["image_sha256"] == sha and r["image_width"] == 640 and r["image_height"] == 480
and S.image_path(r) and S.image_path(r).name == sha + ".png")
r2 = S.upsert_fb_post({"id": "pack-73/2026-09-10-hike", "status": "scheduled", "message": "Hike Saturday, 1 PM"})
check("same id is one row, updated, image kept", r2["message"].endswith("1 PM") and r2["image_sha256"] == sha and r2["fb_post_id"] == "141_9")
check("listed, open-only filter", any(p["id"] == "pack-73/2026-09-10-hike" for p in S.list_fb_posts(include_done=False)))
c = S.mark_fb_cancelled("pack-73/2026-09-10-hike", "mike@example.test")
check("cancel recorded", c["status"] == "cancelled" and c["cancelled_by"] == "mike@example.test"
and not any(p["id"] == c["id"] for p in S.list_fb_posts(include_done=False)))
r3 = S.upsert_fb_post({"id": "troop-73/2026-09-12-canoe", "status": "cancelled", "cancelled_by": "ntfy button"})
check("publisher-reported cancel stamps cancelled_at", r3["cancelled_at"] and r3["cancelled_by"] == "ntfy button")
past = S.upsert_fb_post({"id": "pack-73/2020-01-01-old", "status": "scheduled", "scheduled_for": "2020-01-01T12:00:00+00:00"})
future = S.upsert_fb_post({"id": "pack-73/2036-01-01-new", "status": "scheduled", "scheduled_for": "2036-01-01T12:00:00+00:00"})
check("a scheduled post past its time reads as published; a future one stays scheduled",
S.get_fb_post("pack-73/2020-01-01-old")["state"] == "published" and S.get_fb_post("pack-73/2036-01-01-new")["state"] == "scheduled"
and S.get_fb_post("pack-73/2020-01-01-old")["status"] == "scheduled")
check("open-only list drops the past one and keeps the future one",
[p["id"] for p in S.list_fb_posts(include_done=False) if p["id"].startswith("pack-73/20")] == ["pack-73/2036-01-01-new"])
check("cancelled beats the clock", S.fb_state({"status": "cancelled", "scheduled_for": "2020-01-01T00:00:00+00:00"}) == "cancelled")
check("fbposts caps: read for leaders, ingest admin-only and scopable", "fbposts:read" in I.CAPS["leader"] and "fbposts:ingest" in I.CAPS["admin"]
and "fbposts:ingest" not in I.KEY_UNSCOPABLE)
con = S.connect(); con.execute("DELETE FROM fb_posts"); con.commit(); con.close()
import os as _os; _os.remove(S.image_path(r))
print("\napi docs registry")
import admin_api as A
reg = A.describe_routes()
check("registry is non-trivial", len(reg) >= 18)
missing = [d for d in reg if not d["capability"] and d["path"] != "/api/admin/whoami"]
check("every route's capability is read from its own _auth call (except whoami): %s"
% ", ".join(d["path"] for d in missing), not missing)
check("key routes gate on apikeys:own", all(d["capability"] == "apikeys:own" for d in reg if "/keys" in d["path"]))
check("docs carry the handler docstring", all(d["doc"] for d in reg if "/keys" in d["path"]))
class _R:
headers = {"authorization": "", "x-forwarded-for": "127.0.0.1"}; cookies = {}; client = None
os.environ["ADMIN_TOKEN"] = "t"; A.ADMIN_TOKEN = "t"
page = A.api_docs(_R(), "t").body.decode()
check("docs page renders every route", page.count("<tr><td><code>") == len(reg))
print("\n%d passed, %d failed" % (PASS, FAIL))
sys.exit(1 if FAIL else 0)
Reference in New Issue View Git Blame Copy Permalink