""" End-to-end pipeline test through the real FastAPI routes. Only the two THIRD-PARTY boundaries are mocked — the LLM extraction (`extract.analyze_batch`) and the Nominatim geocoder (`geocode.geocode_row`). Everything else runs for real: the HTTP routes, prefilter, response parsing, CSV/KML/GeoJSON export, the SSE progress stream, and the results browser. This is the test that would have caught the bugs that only showed up on deploy: - the `TemplateResponse(name, {...})` 500 (caught by `GET /`) - the chatbot `get_prompt` KeyError 500 (caught by `POST /chatbot-prepare`) Run: python3 tests/test_e2e.py (needs httpx for Starlette's TestClient) """ import json import shutil import sys import time import xml.etree.ElementTree as ET 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: # pragma: no cover - environment without httpx print(f"SKIP: TestClient unavailable ({exc}). `pip install httpx` to run the e2e test.") sys.exit(0) from pipeline import extract as extract_mod from pipeline import geocode as geocode_mod from pipeline.extract import DEFAULT_MODEL 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) # ── Sample Instagram export: a place, a gym (prefiltered out), a cafe ────────── SAMPLE = [ {"timestamp": 1700000000, "label_values": [ {"label": "URL", "value": "https://www.instagram.com/reel/PLACE1/"}, {"label": "Caption", "value": "Best tonkotsu ramen at Ichiran in Tokyo 🍜 must try"}]}, {"timestamp": 1700000100, "label_values": [ {"label": "URL", "value": "https://www.instagram.com/reel/GYM1/"}, {"label": "Caption", "value": "Crushing leg day at the gym 💪 squats and deadlifts #fitness"}]}, {"timestamp": 1700000200, "label_values": [ {"label": "URL", "value": "https://www.instagram.com/reel/CAFE1/"}, {"label": "Caption", "value": "Cozy matcha latte at Blue Bottle Coffee, Kyoto"}]}, ] SAMPLE_BYTES = json.dumps(SAMPLE).encode() # ── Third-party mocks (the ONLY things stubbed) ─────────────────────────────── def _place(name, city, country, category, cuisine) -> dict: 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, cuisine=cuisine) return d def fake_analyze_batch(client, posts, model=DEFAULT_MODEL, provider="anthropic", ollama_url=""): """Stand in for the LLM: deterministic extraction (prefiltered posts never arrive).""" out = [] for p in posts: cap = (p.get("caption") or "").lower() if "ramen" in cap or "ichiran" in cap: out.append(_place("Ichiran", "Tokyo", "Japan", "Restaurant", "Japanese")) elif "coffee" in cap or "matcha" in cap: out.append(_place("Blue Bottle Coffee", "Kyoto", "Japan", "Cafe", "Coffee")) else: out.append(None) return out _FAKE_COORDS = {"Ichiran": (35.6595, 139.7005), "Blue Bottle Coffee": (35.0116, 135.7681)} def fake_geocode_row(name, city, state, country, address): """Stand in for Nominatim: canned coordinates, no network.""" return _FAKE_COORDS.get(name, (None, None)) def wait_done(client, job_id, timeout=30) -> dict: """Consume the real SSE progress stream until the job ends.""" deadline = time.time() + timeout with client.stream("GET", f"/progress/{job_id}") as resp: for line in resp.iter_lines(): if line and line.startswith("data:"): state = json.loads(line[5:]) if state.get("step") in ("done", "error"): return state if time.time() > deadline: return {"step": "timeout"} return {"step": "closed"} # ── Run ─────────────────────────────────────────────────────────────────────── _orig_ab, _orig_gr = extract_mod.analyze_batch, geocode_mod.geocode_row extract_mod.analyze_batch = fake_analyze_batch geocode_mod.geocode_row = fake_geocode_row created_jobs: list[str] = [] try: with TestClient(webapp.app) as client: # 1) Index route renders (regression: TemplateResponse signature 500). check("GET / returns 200", client.get("/").status_code == 200) # 2) Full upload → extract → geocode → export, via the real routes. up = client.post( "/upload", files={"file": ("saved_posts.json", SAMPLE_BYTES, "application/json")}, data={"provider": "anthropic", "model": DEFAULT_MODEL, "api_key": "sk-ant-dummy"}, ) check("POST /upload accepted", up.status_code == 200 and "job_id" in up.json()) job = up.json()["job_id"] created_jobs.append(job) st = wait_done(client, job) check("upload pipeline reached 'done'", st.get("step") == "done") rows = client.get(f"/results/{job}").json().get("rows", []) check("2 places extracted (gym was prefiltered)", len(rows) == 2) check("Ichiran extracted and geocoded", any(r["name"] == "Ichiran" and r["geocoded"] for r in rows)) kml = client.get(f"/download/{job}").text check("KML download has a placemark for Ichiran", "" in kml and "Ichiran" in kml) try: ET.fromstring(kml) check("KML download is well-formed XML", True) except ET.ParseError as exc: check(f"KML download is well-formed XML ({exc})", False) gj = client.get(f"/download/{job}/geojson").text check("GeoJSON download has the Ichiran feature", '"Feature"' in gj and "Ichiran" in gj) # 3) Free-chatbot path (regression: get_prompt KeyError 500). prep = client.post( "/chatbot-prepare", files={"file": ("saved_posts.json", SAMPLE_BYTES, "application/json")}, ) check("POST /chatbot-prepare returns prompt", prep.status_code == 200 and "prompt" in prep.json() and "{post_count}" not in prep.json()["prompt"]) cjob = prep.json()["job_id"] created_jobs.append(cjob) reply = [] for e in prep.json()["export_posts"]: cap = (e["caption"] or "").lower() if "ramen" in cap: reply.append({"post_number": e["post_number"], "is_place": True, "name": "Ichiran", "city": "Tokyo", "country": "Japan", "category": "Restaurant"}) else: reply.append({"post_number": e["post_number"], "is_place": False}) proc = client.post(f"/chatbot-process/{cjob}", json={"responses": [json.dumps(reply)]}) check("POST /chatbot-process accepted", proc.status_code == 200) st2 = wait_done(client, cjob) check("chatbot pipeline reached 'done'", st2.get("step") == "done") check("chatbot produced at least one place", len(client.get(f"/results/{cjob}").json().get("rows", [])) >= 1) # 4) /import — CSV path (skip extraction, jump straight to tabs). import csv as _csv, io as _io csv_buf = _io.StringIO() from pipeline.extract import FIELDNAMES w = _csv.DictWriter(csv_buf, fieldnames=FIELDNAMES) w.writeheader() w.writerow({k: "" for k in FIELDNAMES} | { "name": "Blue Bottle Coffee", "city": "Kyoto", "country": "Japan", "category": "Cafe", "lat": "35.0116", "lng": "135.7681", "status": "unvisited", }) csv_bytes = csv_buf.getvalue().encode() imp = client.post( "/import", files={"file": ("places_full.csv", csv_bytes, "text/csv")}, ) check("POST /import (CSV) returns 200", imp.status_code == 200) imp_data = imp.json() check("POST /import returns job_id", "job_id" in imp_data) check("POST /import reports 1 row", imp_data.get("rows") == 1) imp_job = imp_data["job_id"] created_jobs.append(imp_job) imp_rows = client.get(f"/results/{imp_job}").json().get("rows", []) check("GET /results after CSV import returns 1 row", len(imp_rows) == 1) check("imported row has correct name", imp_rows and imp_rows[0]["name"] == "Blue Bottle Coffee") check("imported row has coordinates", imp_rows and imp_rows[0]["geocoded"] is True) # 5) /import — KML path (minimal KML with one placemark in a folder). kml_str = ( '' '' '' 'Japan' 'Tokyo' '' 'Ichiran Shibuya' '' 'Highlight: Tonkotsu ramen
via @foodie
' 'View on Instagram ↗]]>' '139.7005,35.6595,0' '' '' '' ) kml_bytes = kml_str.encode() kimp = client.post( "/import", files={"file": ("places_map.kml", kml_bytes, "application/vnd.google-earth.kml+xml")}, ) check("POST /import (KML) returns 200", kimp.status_code == 200) kimp_data = kimp.json() check("POST /import (KML) reports 1 row", kimp_data.get("rows") == 1) kimp_job = kimp_data["job_id"] created_jobs.append(kimp_job) kimp_rows = client.get(f"/results/{kimp_job}").json().get("rows", []) check("GET /results after KML import returns 1 row", len(kimp_rows) == 1) check("KML-imported row has name", kimp_rows and kimp_rows[0]["name"] == "Ichiran Shibuya") check("KML-imported row has coordinates", kimp_rows and kimp_rows[0]["geocoded"] is True) check("KML-imported row has category", kimp_rows and kimp_rows[0]["category"] == "Restaurant") # 6) /extract-url — single post URL → multi-venue & person-handle cases. # Mirrors the TODO test cases: every venue must be extracted AND pinned # separately; a person @handle must be excluded. The caption fetch # (fetch_post_metadata) is mocked alongside the LLM and geocoder. from pipeline import transcribe as transcribe_mod def run_url(url, batch_fn, coords): """POST /extract-url with a stubbed caption fetch, LLM, and geocoder.""" orig_fpm = transcribe_mod.fetch_post_metadata orig_ab2 = extract_mod.analyze_batch orig_gr2 = geocode_mod.geocode_row transcribe_mod.fetch_post_metadata = lambda u, *a, **k: { "caption": "see post", "creator": "guide", "location": None} extract_mod.analyze_batch = batch_fn geocode_mod.geocode_row = ( lambda name, city, state, country, address: coords.get(name, (None, None))) try: r = client.post("/extract-url", data={ "url": url, "provider": "anthropic", "model": DEFAULT_MODEL, "api_key": "sk-ant-dummy"}) jid = r.json()["job_id"] created_jobs.append(jid) wait_done(client, jid) return client.get(f"/results/{jid}").json().get("rows", []) finally: transcribe_mod.fetch_post_metadata = orig_fpm extract_mod.analyze_batch = orig_ab2 geocode_mod.geocode_row = orig_gr2 # TC1: caption tags two venue handles → both extracted and pinned separately. two_rows = run_url( "https://www.instagram.com/p/MULTI2/", lambda *a, **k: [_place("ULT Coffee", "Osaka", "Japan", "Cafe", "Coffee"), _place("Koffee Mameya", "Tokyo", "Japan", "Cafe", "Coffee")], {"ULT Coffee": (34.6937, 135.5023), "Koffee Mameya": (35.6595, 139.7005)}, ) check("URL multi-venue: 2 rows extracted", len(two_rows) == 2) check("URL multi-venue: both pinned separately", len(two_rows) == 2 and all(r["geocoded"] for r in two_rows)) check("URL multi-venue: both venue names present", {r["name"] for r in two_rows} == {"ULT Coffee", "Koffee Mameya"}) # TC3: editorial/list post tagging 5+ venues → all returned & geocoded, none dropped. names5 = ["Bar Basso", "Camparino", "Nottingham Forest", "Mag Cafe", "1930"] five_rows = run_url( "https://www.instagram.com/p/FIVE/", lambda *a, **k: [_place(n, "Milan", "Italy", "Bar", "Cocktail") for n in names5], {n: (45.46 + i * 0.01, 9.18 + i * 0.01) for i, n in enumerate(names5)}, ) check("URL 5-venue list: all 5 rows present (none dropped)", len(five_rows) == 5) check("URL 5-venue list: every venue geocoded", len(five_rows) == 5 and all(r["geocoded"] for r in five_rows)) check("URL 5-venue list: names round-trip intact", {r["name"] for r in five_rows} == set(names5)) # TC4: one handle is a person account (LLM marks is_place:false) → excluded. person_rows = run_url( "https://www.instagram.com/p/PERSON/", lambda *a, **k: [_place("Tartine Bakery", "San Francisco", "USA", "Bakery", "Pastry"), None, # the person @handle _place("Four Barrel Coffee", "San Francisco", "USA", "Cafe", "Coffee")], {"Tartine Bakery": (37.7614, -122.4241), "Four Barrel Coffee": (37.7670, -122.4218)}, ) check("URL person-handle: person excluded, 2 venues only", len(person_rows) == 2) check("URL person-handle: only the real venues remain", {r["name"] for r in person_rows} == {"Tartine Bakery", "Four Barrel Coffee"}) finally: extract_mod.analyze_batch = _orig_ab geocode_mod.geocode_row = _orig_gr for j in created_jobs: shutil.rmtree(webapp.JOBS_DIR / j, ignore_errors=True) # ── Report ──────────────────────────────────────────────────────────────────── print() if failures: print(f"FAIL — {len(failures)} e2e check(s) failed:") for f in failures: print(f" ✗ {f}") sys.exit(1) print("PASS — full pipeline e2e (upload + chatbot) works through the real routes.") sys.exit(0)