"""In-process smoke test of the full discovery flow. Uses httpx ASGITransport to drive the FastAPI app without a network server, and patches DiscoveryExtractor so no real Anthropic call is made. Run: .venv/Scripts/python.exe smoke_test.py """ import asyncio import os os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///./data/smoke.db" os.environ.setdefault("JWT_SECRET", "smoke-jwt-secret") os.environ.setdefault("IMPACTFLOW_API_KEY", "smoke-api-key") from httpx import ASGITransport, AsyncClient # noqa: E402 from app import models # noqa: E402,F401 from app.database import Base, engine # noqa: E402 import app.routers.discovery as disc # noqa: E402 class FakeExtractor: def __init__(self, api_key, model="fake"): pass async def extract(self, responses): assert "friction" in responses return { "triad": "gut", "probable_type": 8, "wing": 9, "instinctual_variant": "sp", "instinctual_stack": "sp/so/sx", "love_summary": "You love hands-on, high-stakes work.", "strength_summary": "You take charge and see the whole board.", "mission_summary": "People need someone who will act and protect.", "vocation_summary": "You can be paid to lead and build.", "overlap_narrative": "You come alive where action meets protection.", "confidence": { "triad": "high", "type": "medium", "variant": "medium", "ikigai": "high", }, "extraction_notes": "", } disc.DiscoveryExtractor = FakeExtractor from app.main import app # noqa: E402 AUTH = {"X-API-Key": os.environ["IMPACTFLOW_API_KEY"]} async def main(): # Fresh DB every run so this is deterministic. db_path = "./data/smoke.db" if os.path.exists(db_path): os.remove(db_path) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) transport = ASGITransport(app=app) async with AsyncClient( transport=transport, base_url="http://test" ) as c: r = await c.get("/health") assert r.status_code == 200 and r.json() == {"status": "ok"}, r.text print("health:", r.json()) # Auth: unauthenticated calls must 401. r = await c.get("/api/me") assert r.status_code == 401, r.text print("unauth /api/me -> 401 OK") r = await c.post("/discovery/start") assert r.status_code == 401, r.text print("unauth /discovery/start -> 401 OK") # Auth: X-API-Key resolves to the synthetic admin user. r = await c.get("/api/me", headers=AUTH) assert r.status_code == 200, r.text me = r.json() assert me["role"] == "admin" assert me["email"] == "api-key@impactflow.local" print("authed /api/me OK ->", me["display_name"]) # Bogus JWT must 401. r = await c.get( "/api/me", headers={"Authorization": "Bearer not-a-real-jwt"}, ) assert r.status_code == 401, r.text print("bogus JWT -> 401 OK") # /api/auth/login should redirect to Google. r = await c.get("/api/auth/login", follow_redirects=False) assert r.status_code in (302, 303, 307), r.text assert "accounts.google.com" in r.headers.get("location", "") print("OAuth login redirects to:", r.headers["location"][:60], "...") # Discovery flow under the API-key admin user. r = await c.post("/discovery/start", headers=AUTH) assert r.status_code == 200, r.text conv_id = r.json()["conversation_id"] print("start -> conversation_id:", conv_id) body = { "prompt_alive": "I felt alive leading a flood rescue.", "prompt_friction": "I confronted a manager cutting workers' hours.", "prompt_pull": "I'm always building and fixing things.", "prompt_recognition": "Recognized for standing up for my crew.", "prompt_future": "I'd build a trades school for written-off kids.", } r = await c.put( f"/discovery/{conv_id}/respond", json=body, headers=AUTH ) assert r.status_code == 200, r.text assert r.json()["status"] == "responses_saved" print("respond:", r.json()) r = await c.get( f"/discovery/conversation/{conv_id}", headers=AUTH ) assert r.status_code == 200, r.text assert r.json()["prompt_friction"] == body["prompt_friction"] print("conversation fetched, friction stored OK") r = await c.post(f"/discovery/{conv_id}/complete", headers=AUTH) assert r.status_code == 200, r.text profile = r.json() assert profile["triad"] == "gut" assert profile["confidence"]["ikigai"] == "high" assert profile["locked"] is False print( "complete -> profile triad:", profile["triad"], "locked:", profile["locked"], ) r = await c.get("/discovery/profile/me", headers=AUTH) assert r.status_code == 200, r.text assert r.json()["overlap_narrative"] print("get my profile OK") r = await c.put("/discovery/profile/me/confirm", headers=AUTH) assert r.status_code == 200, r.text assert r.json()["status"] == "locked" print("confirm:", r.json()) r = await c.get("/discovery/profile/me", headers=AUTH) assert r.json()["locked"] is True print("profile now locked:", r.json()["locked"]) # 404 paths r = await c.post("/discovery/does-not-exist/complete", headers=AUTH) assert r.status_code == 404 print("404 handling OK") # Activity tracking — at this point we've made several authed, # 2xx requests, so the activity log should have entries. r = await c.get("/api/activity?limit=50", headers=AUTH) assert r.status_code == 200, r.text events = r.json() assert len(events) > 0, "expected activity log entries" assert all(e["source"] == "mcp" for e in events), ( "API key requests must be tagged source=mcp" ) print(f"activity feed has {len(events)} entries, all source=mcp OK") r = await c.get("/api/activity/summary?days=7", headers=AUTH) assert r.status_code == 200, r.text summary = r.json() assert summary["total"] > 0 assert summary["mcp_count"] > 0 print("activity summary:", summary["total"], "events") r = await c.get("/api/me/stats", headers=AUTH) assert r.status_code == 200, r.text stats = r.json() assert stats["conversations"] == 1 assert stats["profiles"] == 1 assert stats["locked_profiles"] == 1 print("me/stats:", stats) print("\nALL SMOKE CHECKS PASSED") if __name__ == "__main__": asyncio.run(main())