"""End-to-end integration test against a running LifeOS server. Exercises reset_demo + the seeded persona, so run the server in demo mode: Usage: LIFEOS_DEMO=1 python app.py (in another shell), then python tests/test_integration.py Note: this drives the real local model, so first run downloads the GGUFs and streaming is real-time (not instant). """ import json import os import sys import httpx BASE = os.environ.get("LIFEOS_URL", "http://127.0.0.1:7860") def call_stream(name: str, data: list) -> str: """POST /gradio_api/call/ then stream SSE; return final output[0].""" r = httpx.post(f"{BASE}/gradio_api/call/{name}", json={"data": data}, timeout=30) r.raise_for_status() event_id = r.json()["event_id"] final = None with httpx.stream("GET", f"{BASE}/gradio_api/call/{name}/{event_id}", timeout=120) as resp: event = None for line in resp.iter_lines(): if line.startswith("event:"): event = line.split(":", 1)[1].strip() elif line.startswith("data:"): payload = json.loads(line.split(":", 1)[1].strip()) if event in ("generating", "complete") and isinstance(payload, list) and payload: final = payload[0] if event == "error": raise RuntimeError(f"{name} errored: {payload}") assert final is not None, f"{name}: no output received" return final def main() -> None: root = httpx.get(BASE + "/", timeout=10) assert root.status_code == 200 and "LifeOS" in root.text, "frontend not served" css = httpx.get(BASE + "/static/style.css", timeout=10) js = httpx.get(BASE + "/static/app.js", timeout=10) assert css.status_code == 200 and js.status_code == 200, "static assets missing" print("PASS frontend served") mem = json.loads(call_stream("reset_demo", [])) assert mem["user_profile"]["first_name"] == "Awais" assert mem["user_profile"]["name"] == "Awais Aziz" print("PASS reset_demo") mem = json.loads(call_stream("get_memory", [])) assert len(mem["meals"]) >= 7 and len(mem["finances"]["subscriptions"]) == 7 print("PASS get_memory") mem = json.loads(call_stream("log_workout", ["run", 30])) assert mem["workouts"][-1]["type"] == "run" print("PASS log_workout") flyer_txt = open(os.path.join(os.path.dirname(__file__), "..", "data", "samples", "flyer.txt"), encoding="utf-8").read() r = httpx.post(f"{BASE}/upload/flyer_text", json={"text": flyer_txt}, timeout=30) deals = r.json()["deals"] assert len(deals) >= 10, f"only {len(deals)} deals extracted" print(f"PASS flyer_text ({len(deals)} deals)") pdf_path = os.path.join(os.path.dirname(__file__), "..", "data", "samples", "flyer.pdf") with open(pdf_path, "rb") as f: r = httpx.post(f"{BASE}/upload/flyer", files={"file": ("flyer.pdf", f, "application/pdf")}, timeout=60) assert len(r.json()["deals"]) >= 10, "PDF flyer extraction failed" print("PASS flyer PDF upload") csv_path = os.path.join(os.path.dirname(__file__), "..", "data", "samples", "transactions.csv") with open(csv_path, "rb") as f: r = httpx.post(f"{BASE}/upload/transactions", files={"file": ("transactions.csv", f, "text/csv")}, timeout=30) subs = r.json()["subscriptions"] assert len(subs) == 7, f"expected 7 subscriptions, got {len(subs)}" print("PASS transactions upload (7 subscriptions)") out = call_stream("food_recommend", [json.dumps(deals)]) assert len(out) > 50 print("PASS food_recommend stream") out = call_stream("health_recommend", []) assert len(out) > 50 print("PASS health_recommend stream") out = call_stream("money_review", [json.dumps(subs)]) assert len(out) > 50 print("PASS money_review stream") out = call_stream("chat", ["Plan my week under $80, high protein, run 3x", "[]", "", ""]) assert len(out) > 50 print("PASS chat stream") out = call_stream("chat", ["What should I cook?", "[]", "", json.dumps(["kitchen"])]) assert len(out) > 50 print("PASS chat stream with refs") out = call_stream("goal_chat", ["I want to buy a used car", "[]", "{}"]) assert len(out) > 20 print("PASS goal_chat stream") out = call_stream("meal_analyze", ["CHICKEN THIGHS 2.49/LB\nGREEK YOGURT 4.99", "lunch"]) assert len(out) > 20 print("PASS meal_analyze stream") plan = json.loads(call_stream("health_plan", [])) assert "recommendation_slots" in plan and "schedule" in plan print(f"PASS health_plan ({len(plan['recommendation_slots'])} slots)") mem = json.loads(call_stream("set_profile", [json.dumps({"city": "Toronto", "first_name": "Awais", "last_name": "Aziz", "postal_code": "M1B 2C3", "country": "Canada"})])) assert mem["user_profile"]["city"] == "Toronto" assert mem["user_profile"]["postal_code"] == "M1B 2C3" assert mem["user_profile"]["name"] == "Awais Aziz" print("PASS set_profile") mem = json.loads(call_stream("upsert_goal", [json.dumps({"title": "Laptop", "target_amount": 1500, "saved": 0, "deadline": "2027-01"})])) gid = [g for g in mem["goals"] if g["title"] == "Laptop"][0]["id"] mem = json.loads(call_stream("delete_goal", [gid])) assert not [g for g in mem["goals"] if g.get("id") == gid] print("PASS upsert_goal / delete_goal") mem = json.loads(call_stream("edit_memory", ["meals", json.dumps([{"date": "2026-06-01", "dish": "Toast", "ingredients": ["bread"]}])])) assert mem["meals"] == [{"date": "2026-06-01", "dish": "Toast", "ingredients": ["bread"]}] err = json.loads(call_stream("edit_memory", ["nope", "[]"])) assert "error" in err print("PASS edit_memory") notes = json.loads(call_stream("list_notes", [])) assert notes and {"id", "text", "kind"} <= set(notes[0]) nid = notes[0]["id"] assert json.loads(call_stream("update_note", [nid, "updated note text"]))["ok"] is True notes2 = json.loads(call_stream("list_notes", [])) assert [n for n in notes2 if n["id"] == nid][0]["text"] == "updated note text" assert json.loads(call_stream("delete_note", [nid]))["ok"] is True assert not [n for n in json.loads(call_stream("list_notes", [])) if n["id"] == nid] print("PASS notes list/update/delete") call_stream("reset_demo", []) print("\nALL INTEGRATION TESTS PASSED") if __name__ == "__main__": sys.exit(main())