Spaces:
Running
Running
| """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/<name> 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()) | |