thethreemagi
b84fc7f064
POST /fbposts/{id}/reschedule {message, scheduled_for}. The site derives
the per-post signature from the stored cancel link and asks the publisher
to delete the post on Facebook and redraft the queue file; the row goes
to drafted with no Facebook id and no cancel link until the publisher's
next cycle reports the new one. Times must be at least 15 minutes out,
with an offset. An explicit null now clears fb_post_id on upsert; an
absent one still keeps it. tests/smoke_admin.py 160 -> 162.
440 lines
30 KiB
Python
440 lines
30 KiB
Python
"""
|
|
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")
|
|
rd = S.upsert_fb_post({"id": "pack-73/2036-01-01-new", "status": "drafted", "message": "edited", "fb_post_id": None, "cancel_url": None})
|
|
check("a redraft clears the facebook id and cancel link", rd["status"] == "drafted" and rd["fb_post_id"] is None and rd["cancel_url"] is None)
|
|
rd2 = S.upsert_fb_post({"id": "pack-73/2036-01-01-new", "status": "scheduled", "message": "edited", "fb_post_id": "141_NEW"})
|
|
rd3 = S.upsert_fb_post({"id": "pack-73/2036-01-01-new", "status": "scheduled", "message": "edited again"})
|
|
check("a report without an id keeps the id", rd2["fb_post_id"] == "141_NEW" and rd3["fb_post_id"] == "141_NEW")
|
|
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)
|