impactflow_discovery/smoke_test.py

"""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())