""" POST /capture — single-post extraction for the iOS Shortcut path (Phase A). The Shortcut fetches /embed/captioned/ on-device and POSTs the HTML (or a ready caption, or a URL for local server-fetch). /capture runs the shared _places_from_caption and returns the extracted place(s). Optional lat/lng carry an Instagram location tag → coordinates without geocoding. Only the LLM is mocked. Run: python3 tests/test_capture.py """ import shutil import sys from pathlib import Path ROOT = Path(__file__).parent.parent sys.path.insert(0, str(ROOT)) try: from starlette.testclient import TestClient except Exception as exc: print(f"SKIP: TestClient unavailable ({exc}). `pip install httpx` to run.") sys.exit(0) from pipeline import extract as extract_mod from web import app as webapp failures: list[str] = [] def check(label: str, cond: bool) -> None: print(("PASS" if cond else "FAIL"), "-", label) if not cond: failures.append(label) def _place(name, city="Tokyo", country="Japan", category="Cafe"): d = {k: "UNKNOWN" for k in ("name", "city", "state", "country", "address", "cuisine", "price_range", "highlight", "occasion")} d.update(name=name, city=city, country=country, category=category) return d def fake_analyze_batch(client, posts, model=None, provider="anthropic", ollama_url=""): """Deterministic stand-in: a place when the caption looks place-y, else None.""" out = [] for p in posts: cap = (p.get("caption") or "").lower() if "ichiran" in cap or "ramen" in cap or "cafe" in cap or "koffee" in cap: out.append(_place("Koffee Mameya" if "koffee" in cap else "Ichiran")) else: out.append(None) return out _orig_ab = extract_mod.analyze_batch extract_mod.analyze_batch = fake_analyze_batch try: with TestClient(webapp.app) as client: # ── caption path ────────────────────────────────────────────────────── r = client.post("/capture", data={"caption": "Amazing @koffee_mameya cafe in Tokyo", "url": "https://instagram.com/p/aaa", "api_key": "sk-test-unused"}) check("caption path: 200", r.status_code == 200) body = r.json() check("caption path: 1 place", body.get("count") == 1) check("caption path: place name extracted", body["places"][0]["name"] == "Koffee Mameya") check("caption path: instagram_url carried", body["places"][0]["instagram_url"] == "https://instagram.com/p/aaa") # ── location tag → coords without geocoding ───────────────────────────── r2 = client.post("/capture", data={"caption": "Ichiran ramen", "url": "https://instagram.com/p/bbb", "lat": "35.6595", "lng": "139.7005", "api_key": "x"}) p2 = r2.json()["places"][0] check("location tag: lat applied", p2["lat"].startswith("35.6595")) check("location tag: lng applied", p2["lng"].startswith("139.7005")) # ── embed_html path (server parses the Caption block) ─────────────────── embed = ('

' 'timeouttokyo_' 'Best ramen — Ichiran in Tokyo

') r3 = client.post("/capture", data={"embed_html": embed, "url": "https://instagram.com/p/ccc", "api_key": "x"}) check("embed_html path: 200", r3.status_code == 200) check("embed_html path: place extracted", r3.json()["places"][0]["name"] == "Ichiran") # ── error: no caption ─────────────────────────────────────────────────── r4 = client.post("/capture", data={"api_key": "x"}) check("no caption → 422", r4.status_code == 422) # ── error: caption with no place ──────────────────────────────────────── r5 = client.post("/capture", data={"caption": "my gym workout routine today", "api_key": "x"}) check("non-place caption → 422", r5.status_code == 422) # ── inbox (Phase B): /capture?token enqueues, /capture-inbox drains+merges ─ created: list[str] = [] client.post("/capture", data={"caption": "@koffee cafe Tokyo", "url": "https://instagram.com/p/zzz", "token": "tok-test", "api_key": "x"}) drain = client.post("/capture-inbox/tok-test").json() if drain.get("job_id"): created.append(drain["job_id"]) check("inbox drain returns a job", bool(drain.get("job_id"))) check("inbox drain added == 1", drain.get("added") == 1) check("inbox drain added_urls correct", drain.get("added_urls") == ["https://instagram.com/p/zzz"]) rows = client.get(f"/results/{drain['job_id']}").json()["rows"] check("inbox job contains the captured place", any(r["instagram_url"] == "https://instagram.com/p/zzz" for r in rows)) # idempotent: a second drain is empty again = client.post("/capture-inbox/tok-test").json() check("second drain is empty", again.get("empty") is True) # drain with merge_into folds into an existing library (keep-existing-wins) client.post("/capture", data={"caption": "Ichiran ramen", "url": "https://instagram.com/p/aaa", "token": "tok-m", "api_key": "x"}) lib = ("name,city,state,country,address,category,cuisine,price_range,highlight," "occasion,lat,lng,creator,saved_at,instagram_url,status\n" "Old Spot,Paris,,France,,Cafe,,,,,48.85,2.35,,,https://instagram.com/p/old,visited\n") merged = client.post("/capture-inbox/tok-m", data={"merge_into": lib}).json() if merged.get("job_id"): created.append(merged["job_id"]) mrows = {r["instagram_url"]: r for r in client.get(f"/results/{merged['job_id']}").json()["rows"]} check("inbox merge: existing place kept", "https://instagram.com/p/old" in mrows) check("inbox merge: existing status preserved", mrows["https://instagram.com/p/old"]["status"] == "visited") check("inbox merge: captured place added", "https://instagram.com/p/aaa" in mrows) for j in created: shutil.rmtree(webapp.JOBS_DIR / j, ignore_errors=True) finally: extract_mod.analyze_batch = _orig_ab print() if failures: print(f"FAIL — {len(failures)} capture check(s) failed:") for f in failures: print(f" ✗ {f}") sys.exit(1) print("PASS — /capture extracts place(s) from a single shared post.") sys.exit(0)