214 lines
13 KiB
Python
214 lines
13 KiB
Python
"""End-to-end check of the Phase 1 API with the model replaced by canned results. No GPU or network needed.
|
|
|
|
Run: python tests/test_phase1.py
|
|
"""
|
|
import io
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
TMP = tempfile.mkdtemp(prefix="parentcalendar-test-")
|
|
os.environ.update(DATA_DIR=TMP, PUBLIC_URL="https://calendar.example.com", EXTRACT_IGNORE="")
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
from PIL import Image # noqa: E402
|
|
|
|
from parentcalendar import app as app_module, db, extraction, scans # noqa: E402
|
|
|
|
app_module.START_WORKER = False
|
|
|
|
|
|
def png(color):
|
|
buf = io.BytesIO()
|
|
Image.new("RGB", (400, 300), color).save(buf, "PNG")
|
|
return buf.getvalue()
|
|
|
|
|
|
def ev(date, title, **kw):
|
|
return {"date": date, "end_date": None, "all_day": True, "start_time": None, "end_time": None, "title": title,
|
|
"notes": None, "location": None, "kind": "event", "applies": True, "source_text": title,
|
|
"confidence": 0.9, "cell_bbox": [0, 0.2, 1, 0.2], "flags": [], **kw}
|
|
|
|
|
|
CANNED = [] # the next fake extraction results, one per page
|
|
|
|
|
|
def fake_extract(img, hint, double_read=None, ignore=None):
|
|
result = CANNED.pop(0)
|
|
return {"model": "fake", "method": "page", "seconds": 0.1, "usage": {"input_tokens": 1, "output_tokens": 1},
|
|
"result": {"date_range": {"start": "2026-10-01", "end": "2026-10-31"}, "menu_items": [], **result},
|
|
"image": img}
|
|
|
|
|
|
extraction.extract_image = fake_extract
|
|
PASTED = [] # the texts fake_text was given
|
|
extraction.extract_text = lambda text, hint, ignore=None: PASTED.append((text, hint["month"])) or {
|
|
**fake_extract(None, hint), "method": "text"}
|
|
|
|
|
|
def upload(c, source_id, image, result, month="2026-10"):
|
|
r = c.post("/api/scans", data={"source_id": source_id, "month": month}, files=[("files", ("p.png", image, "image/png"))])
|
|
assert r.status_code == 200, r.text
|
|
CANNED.append(result)
|
|
with db.connect() as conn:
|
|
scans.process_scan(conn, r.json()["id"])
|
|
return c.get(f"/api/scans/{r.json()['id']}").json()
|
|
|
|
|
|
def test():
|
|
with TestClient(app_module.app) as c:
|
|
# Household setup
|
|
s = c.post("/api/children", json={"display_name": "Sam", "color": "#ff8800"}).json()
|
|
sam = s["children"][0]["id"]
|
|
s = c.post("/api/calendars", json={"name": "Sam's school", "child_ids": [sam], "alarm_minutes": 60}).json()
|
|
cal = s["calendars"][0]["id"]
|
|
assert len(s["calendars"][0]["feed_token"]) >= 32 # 192 random bits
|
|
s = c.post("/api/sources", json={"name": "District", "child_ids": [sam]}).json()
|
|
district = s["sources"][0]["id"]
|
|
s = c.post("/api/sources", json={"name": "Lunch", "layout_type": "menu", "child_ids": [sam]}).json()
|
|
menu_src = s["sources"][1]["id"]
|
|
assert c.put("/api/user", json={"timezone": "America/Chicago"}).json()["timezone"] == "America/Chicago"
|
|
assert c.put("/api/user", json={"timezone": "Mars/Base"}).status_code == 422
|
|
|
|
# Upload: bad files are rejected; the same image again returns the same scan (no second model run)
|
|
bad = c.post("/api/scans", data={"source_id": district, "month": "2026-10"}, files=[("files", ("x.png", b"nope", "image/png"))])
|
|
assert bad.status_code == 400
|
|
scan = upload(c, district, png("white"), {"events": [
|
|
ev("2026-10-02", "Spirit Day"),
|
|
ev("2026-10-20", "PTO", all_day=False, start_time="16:00", location="Library"),
|
|
ev("2026-10-29", "End of Q.1", flags=["second read differs: date"]),
|
|
ev("2026-10-14", "Grade 5 field trip", applies=False),
|
|
ev("2026-10-30", "No School", kind="no_school"),
|
|
]})
|
|
again = c.post("/api/scans", data={"source_id": district, "month": "2026-10"},
|
|
files=[("files", ("p.png", png("white"), "image/png"))]).json()
|
|
assert again["id"] == scan["id"]
|
|
assert scan["status"] == "parsed" and len(scan["events"]) == 5
|
|
by = {e["title"]: e for e in scan["events"]}
|
|
assert by["PTO"]["child_ids"] == [sam] and by["End of Q.1"]["flags"] == ["second read differs: date"]
|
|
assert by["Grade 5 field trip"]["include"] is False # not for this child: unchecked
|
|
assert c.get(f"/api/scans/{scan['id']}/pages/1").status_code == 200
|
|
|
|
# Review: edit, validation, undo, bulk, add
|
|
r = c.patch(f"/api/events/{by['End of Q.1']['id']}", json={"date": "2026-10-28", "end_date": "2026-10-20"})
|
|
assert r.status_code == 422 # end before start
|
|
c.patch(f"/api/events/{by['End of Q.1']['id']}", json={"date": "2026-10-28"})
|
|
c.patch(f"/api/events/{by['Spirit Day']['id']}", json={"title": "Spirit Day!"})
|
|
scan = c.post(f"/api/scans/{scan['id']}/undo").json() # undoes the title only
|
|
titles = {e["title"]: e for e in scan["events"]}
|
|
assert "Spirit Day" in titles and titles["End of Q.1"]["date"] == "2026-10-28" and scan["status"] == "reviewed"
|
|
c.post("/api/events/bulk", json={"ids": [by["Spirit Day"]["id"], by["PTO"]["id"]], "delete": True})
|
|
scan = c.post(f"/api/scans/{scan['id']}/undo").json() # both come back together
|
|
assert all(e["status"] == "draft" for e in scan["events"])
|
|
added = c.post("/api/events", json={"scan_id": scan["id"], "date": "2026-10-09", "title": "Spirit Day",
|
|
"child_ids": [sam]}).json()
|
|
assert added["status"] == "draft" and added["all_day"] is True
|
|
|
|
# Publish: checked drafts are confirmed, the unchecked one is dropped
|
|
scan = c.post(f"/api/scans/{scan['id']}/publish").json()
|
|
assert scan["status"] == "published"
|
|
assert sorted(e["title"] for e in scan["events"]) == ["End of Q.1", "No School", "PTO", "Spirit Day", "Spirit Day"]
|
|
assert c.post(f"/api/scans/{scan['id']}/publish").status_code == 409
|
|
|
|
# Export: first download has everything; the next only what changed
|
|
ics1 = c.get(f"/api/calendars/{cal}/export.ics").text
|
|
assert ics1.count("BEGIN:VEVENT") == 5 and "TZID=America/Chicago:20261020T160000" in ics1
|
|
assert c.get(f"/api/calendars/{cal}/export.ics").text.count("BEGIN:VEVENT") == 0
|
|
pto = next(e for e in scan["events"] if e["title"] == "PTO")
|
|
assert c.patch(f"/api/events/{pto['id']}", json={"end_time": "17:00"}).json()["sequence"] == 1
|
|
ics2 = c.get(f"/api/calendars/{cal}/export.ics").text
|
|
assert ics2.count("BEGIN:VEVENT") == 1 and pto["uid"] in ics2.replace("\r\n ", "") and "SEQUENCE:1" in ics2
|
|
c.delete(f"/api/events/{pto['id']}")
|
|
assert "STATUS:CANCELLED" in c.get(f"/api/calendars/{cal}/export.ics").text
|
|
assert c.get(f"/api/calendars/{cal}/export.ics?everything=true").text.count("BEGIN:VEVENT") == 4
|
|
|
|
# Feed: everything plus the deletion, with an ETag; a reset link retires the old one
|
|
token = c.get("/api/state").json()["calendars"][0]["feed_token"]
|
|
r = c.get(f"/feed/{token}.ics")
|
|
assert r.status_code == 200 and r.text.count("BEGIN:VEVENT") == 5 and "STATUS:CANCELLED" in r.text
|
|
assert c.get(f"/feed/{token}.ics", headers={"If-None-Match": r.headers["etag"]}).status_code == 304
|
|
assert c.get("/feed/nope.ics").status_code == 404
|
|
new = c.post(f"/api/calendars/{cal}/feed_token").json()["calendars"][0]["feed_token"]
|
|
assert new != token and c.get(f"/feed/{token}.ics").status_code == 404 and c.get(f"/feed/{new}.ics").status_code == 200
|
|
|
|
# A revised calendar: the matching event keeps its UID and is updated, not duplicated
|
|
spirit = next(e for e in scan["events"] if e["title"] == "Spirit Day" and e["date"] == "2026-10-02")
|
|
rescan = upload(c, district, png("gray"), {"events": [
|
|
ev("2026-10-02", "Spirit Day", notes="Wear purple"), ev("2026-10-23", "Spirit Day")]})
|
|
m = {e["date"]: e for e in rescan["events"]}
|
|
assert m["2026-10-02"]["match"] == "updated" and m["2026-10-02"]["replaces_id"] == spirit["id"]
|
|
assert m["2026-10-23"]["match"] == "new"
|
|
c.post(f"/api/scans/{rescan['id']}/publish")
|
|
events = c.get("/api/events", params={"calendar_id": cal}).json()
|
|
spirits = [e for e in events if e["date"] == "2026-10-02"]
|
|
assert len(spirits) == 1 and spirits[0]["uid"] == spirit["uid"] and spirits[0]["notes"] == "Wear purple"
|
|
assert spirits[0]["sequence"] == 1
|
|
|
|
# Lunch: menu days, greyed closures, picks as events
|
|
menu = upload(c, menu_src, png("yellow"), {"events": [], "menu_items": [
|
|
{"date": "2026-10-05", "items": ["Pork Chopette", "Corn"], "no_lunch": False, "flags": [],
|
|
"reads": [{"items": ["Pork Chopette", "Corn"], "no_lunch": False},
|
|
{"items": ["Pork Chop", "Corn"], "no_lunch": False}]},
|
|
{"date": "2026-10-30", "items": [], "no_lunch": True, "flags": []}]})
|
|
assert [r["items"][0] for r in menu["menu_items"][0]["reads"]] == ["Pork Chopette", "Pork Chop"] # both reads kept
|
|
assert menu["menu_items"][1]["reads"] is None
|
|
c.post(f"/api/scans/{menu['id']}/publish")
|
|
# After publishing, a mistaken "no lunch" keeps the food, and unticking it brings the day back
|
|
item = menu["menu_items"][0]["id"]
|
|
r = c.patch(f"/api/menu_items/{item}", json={"items": ["Pork Chopette", "Corn"], "no_lunch": True}).json()
|
|
assert r["no_lunch"] is True and r["items"] == ["Pork Chopette", "Corn"], r
|
|
days = {d["date"]: d for d in c.get("/api/lunch", params={"child_id": sam, "month": "2026-10"}).json()}
|
|
assert days["2026-10-05"]["greyed"] is True
|
|
c.patch(f"/api/menu_items/{item}", json={"items": ["Pork Chopette", "Corn"], "no_lunch": False})
|
|
days = {d["date"]: d for d in c.get("/api/lunch", params={"child_id": sam, "month": "2026-10"}).json()}
|
|
assert days["2026-10-05"]["greyed"] is False and days["2026-10-30"]["greyed"] is True
|
|
assert days["2026-10-30"]["closed"] == "No School"
|
|
assert c.post("/api/lunch", json={"child_id": sam, "date": "2026-10-30", "picked": True}).status_code == 409
|
|
days = c.post("/api/lunch", json={"child_id": sam, "date": "2026-10-05", "picked": True}).json()
|
|
assert next(d for d in days if d["date"] == "2026-10-05")["picked"] is True
|
|
pack = [e for e in c.get("/api/events", params={"calendar_id": cal}).json() if e["kind"] == "pack_lunch"]
|
|
assert len(pack) == 1 and pack[0]["title"] == "Pack lunch — Sam" and pack[0]["all_day"] is False and pack[0]["start_time"] == "08:00" and pack[0]["notes"] == "Pork Chopette"
|
|
c.post("/api/lunch", json={"child_id": sam, "date": "2026-10-05", "picked": False})
|
|
c.post("/api/lunch", json={"child_id": sam, "date": "2026-10-05", "picked": True})
|
|
pack2 = [e for e in c.get("/api/events", params={"calendar_id": cal}).json() if e["kind"] == "pack_lunch"]
|
|
assert pack2[0]["uid"] == pack[0]["uid"] and pack2[0]["sequence"] == 2 # same event, re-picked
|
|
# Select all skips the no-school day instead of failing; clear all unpicks
|
|
c.post("/api/lunch", json={"child_id": sam, "date": "2026-10-05", "picked": False})
|
|
days = {d["date"]: d for d in c.post("/api/lunch/bulk", json={
|
|
"child_id": sam, "dates": ["2026-10-05", "2026-10-30"], "picked": True}).json()}
|
|
assert days["2026-10-05"]["picked"] is True and days["2026-10-30"]["picked"] is False
|
|
days = {d["date"]: d for d in c.post("/api/lunch/bulk", json={
|
|
"child_id": sam, "dates": ["2026-10-05"], "picked": False}).json()}
|
|
assert days["2026-10-05"]["picked"] is False
|
|
assert c.post("/api/lunch/bulk", json={"child_id": sam, "dates": [], "picked": True}).status_code == 422
|
|
|
|
# A failed extraction marks the scan failed and can be retried
|
|
r = c.post("/api/scans", data={"source_id": district, "month": "2026-11"},
|
|
files=[("files", ("p.png", png("blue"), "image/png"))]).json()
|
|
with db.connect() as conn:
|
|
conn.execute("UPDATE scan SET status = 'failed', error = 'boom' WHERE id = ?", (r["id"],))
|
|
conn.commit()
|
|
assert c.post(f"/api/scans/{r['id']}/retry").json()["status"] == "queued"
|
|
|
|
# Pasted text: read as text, no photo pages; blank text is refused; the same text again is the same scan
|
|
assert c.post("/api/scans/text", json={"source_id": district, "month": "2026-11", "text": " "}).status_code == 400
|
|
r = c.post("/api/scans/text", json={"source_id": district, "month": "2026-11", "text": " 5th-Bring a G sound "})
|
|
assert r.status_code == 200 and r.json()["text"] == "5th-Bring a G sound" and r.json()["image_keys"] == [], r.text
|
|
CANNED.append({"events": [ev("2026-11-05", "Bring a G sound", cell_bbox=None)]})
|
|
with db.connect() as conn:
|
|
scans.process_scan(conn, r.json()["id"])
|
|
pasted = c.get(f"/api/scans/{r.json()['id']}").json()
|
|
assert PASTED == [("5th-Bring a G sound", 11)] and pasted["status"] == "parsed", pasted
|
|
assert [e["title"] for e in pasted["events"]] == ["Bring a G sound"] and pasted["events"][0]["child_ids"] == [sam]
|
|
assert c.get(f"/api/scans/{pasted['id']}/pages/1").status_code == 404
|
|
again = c.post("/api/scans/text", json={"source_id": district, "month": "2026-11", "text": "5th-Bring a G sound"})
|
|
assert again.json()["id"] == pasted["id"]
|
|
print("phase 1 api test ok")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test()
|