From b57089d523dcf9c06d5e5cd9c2f56a565d4e08a6 Mon Sep 17 00:00:00 2001 From: Michael Chihlas Date: Sat, 30 May 2026 19:33:36 -0400 Subject: [PATCH] test(l1): rewrite AI-build API tests on proven register/login/subscription helpers KNOWN-RED (handoff): test_escalations_forbidden_for_l1_tech passes; the intake/ next-node tests still 403 'L1 access required' despite the DB role persisting as l1_tech (verified) and get_current_user reading role from the DB. The identical register->promote->subscribe->login helper works in test_l1_endpoints.py, so this is a test-harness/auth interaction needing interactive debugging in a clean shell. Co-Authored-By: Claude Opus 4.7 --- backend/tests/test_l1_api_ai_build.py | 156 ++++++++++++-------------- 1 file changed, 74 insertions(+), 82 deletions(-) diff --git a/backend/tests/test_l1_api_ai_build.py b/backend/tests/test_l1_api_ai_build.py index 04ffacdd..59fe0428 100644 --- a/backend/tests/test_l1_api_ai_build.py +++ b/backend/tests/test_l1_api_ai_build.py @@ -1,60 +1,73 @@ -"""Tests for the Phase 2A L1 AI-build API surface. +"""Integration tests for the Phase 2A L1 AI-build API surface. -Covers intake dispatch (match_or_build outcomes), the next-node endpoint, and -the engineer escalations list. The orchestrator and node generator are mocked — -this exercises the endpoint wiring, not the AI. +Covers intake dispatch (match_or_build outcomes), the next-node endpoint, and the +engineer escalations list. The orchestrator and node generator are mocked — this +exercises the endpoint wiring, not the AI. Auth/subscription follow the same +register → promote-role → ensure-subscription → login pattern as test_l1_endpoints. """ import uuid +from datetime import datetime, timezone from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient +from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession -from app.core.security import create_access_token, get_password_hash -from app.models.account import Account +from app.models.l1_walk_session import L1WalkSession +from app.models.subscription import Subscription from app.models.user import User -async def _seed_user(test_db: AsyncSession, *, account_role="l1_tech"): - account = Account(name="L1 Co", display_code=f"L1{uuid.uuid4().hex[:6].upper()}") - test_db.add(account) - await test_db.flush() - user = User( - account_id=account.id, - email=f"l1-{uuid.uuid4().hex[:8]}@example.com", - hashed_password=get_password_hash("pw"), - full_name="L1 Tech", - account_role=account_role, - is_active=True, - is_verified=True, +async def _register(client: AsyncClient, *, email: str) -> dict: + resp = await client.post( + "/api/v1/auth/register", + json={"email": email, "password": "TestPassword123!", "name": "Test User"}, ) - test_db.add(user) - await test_db.flush() - return account, user + assert resp.status_code in (200, 201), resp.text + return resp.json() -def _auth(user: User) -> dict: - return {"Authorization": f"Bearer {create_access_token(subject=str(user.id))}"} +async def _login(client: AsyncClient, *, email: str) -> dict: + resp = await client.post( + "/api/v1/auth/login/json", + json={"email": email, "password": "TestPassword123!"}, + ) + assert resp.status_code == 200, resp.text + return {"Authorization": f"Bearer {resp.json()['access_token']}"} + + +async def _ensure_subscription(db: AsyncSession, account_id: uuid.UUID) -> None: + await db.execute(delete(Subscription).where(Subscription.account_id == account_id)) + db.add(Subscription(account_id=account_id, plan="pro", status="active")) + await db.commit() + + +async def _make_user(client: AsyncClient, db: AsyncSession, *, email: str, account_role: str) -> dict: + """Register a user, promote to a role, ensure an active subscription, return headers + ids.""" + data = await _register(client, email=email) + uid = uuid.UUID(data["id"]) + acct_id = uuid.UUID(data["account_id"]) + user = (await db.execute(select(User).where(User.id == uid))).scalar_one() + user.account_role = account_role + await db.commit() + await _ensure_subscription(db, acct_id) + headers = await _login(client, email=email) # login AFTER role change + return {"headers": headers, "account_id": acct_id, "user_id": uid} @pytest.mark.asyncio -async def test_intake_build_outcome_creates_ai_build_session( - client: AsyncClient, test_db: AsyncSession -): +async def test_intake_build_outcome_creates_ai_build_session(client: AsyncClient, test_db: AsyncSession): """intake → match_or_build returns 'build' → an ai_build session is created.""" - account, user = await _seed_user(test_db) + info = await _make_user(client, test_db, email="aib_build@example.com", account_role="l1_tech") with patch( "app.api.endpoints.l1.match_or_build.match_or_build", new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build", "category": "printer"}), ): - r = await client.post( - "/api/v1/l1/intake", - json={"problem_statement": "printer jam"}, - headers=_auth(user), - ) - assert r.status_code == 200 + r = await client.post("/api/v1/l1/intake", + json={"problem_statement": "printer jam"}, headers=info["headers"]) + assert r.status_code == 200, r.text body = r.json() assert body["outcome"] == "build" assert body["session_kind"] == "ai_build" @@ -64,83 +77,64 @@ async def test_intake_build_outcome_creates_ai_build_session( @pytest.mark.asyncio async def test_intake_out_of_scope(client: AsyncClient, test_db: AsyncSession): """intake → 'out_of_scope' → no session, surfaced to the caller.""" - account, user = await _seed_user(test_db) + info = await _make_user(client, test_db, email="aib_oos@example.com", account_role="l1_tech") with patch( "app.api.endpoints.l1.match_or_build.match_or_build", new=AsyncMock(return_value={"outcome": "out_of_scope", "category": "unknown"}), ): - r = await client.post( - "/api/v1/l1/intake", - json={"problem_statement": "weird thing"}, - headers=_auth(user), - ) - assert r.status_code == 200 + r = await client.post("/api/v1/l1/intake", + json={"problem_statement": "weird"}, headers=info["headers"]) + assert r.status_code == 200, r.text body = r.json() assert body["outcome"] == "out_of_scope" - assert body["session_id"] is None + assert body.get("session_id") is None @pytest.mark.asyncio async def test_intake_suggest_returns_near_miss(client: AsyncClient, test_db: AsyncSession): """intake → 'suggest' → near_miss prompt, no session.""" - account, user = await _seed_user(test_db) + info = await _make_user(client, test_db, email="aib_sugg@example.com", account_role="l1_tech") near = {"flow_id": str(uuid.uuid4()), "flow_name": "VPN", "score": 0.66} with patch( "app.api.endpoints.l1.match_or_build.match_or_build", new=AsyncMock(return_value={"outcome": "suggest", "near_miss": near, "can_build": True}), ): - r = await client.post( - "/api/v1/l1/intake", - json={"problem_statement": "vpn"}, - headers=_auth(user), - ) - assert r.status_code == 200 - body = r.json() - assert body["outcome"] == "suggest" - assert body["near_miss"]["flow_name"] == "VPN" + r = await client.post("/api/v1/l1/intake", + json={"problem_statement": "vpn"}, headers=info["headers"]) + assert r.status_code == 200, r.text + assert r.json()["near_miss"]["flow_name"] == "VPN" @pytest.mark.asyncio async def test_next_node_returns_generated_node(client: AsyncClient, test_db: AsyncSession): """After a build intake, /next-node returns the node from advance_ai_build.""" - account, user = await _seed_user(test_db) + info = await _make_user(client, test_db, email="aib_next@example.com", account_role="l1_tech") with patch( "app.api.endpoints.l1.match_or_build.match_or_build", new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build", "category": "printer"}), ): - r = await client.post( - "/api/v1/l1/intake", - json={"problem_statement": "printer jam"}, - headers=_auth(user), - ) + r = await client.post("/api/v1/l1/intake", + json={"problem_statement": "printer jam"}, headers=info["headers"]) sid = r.json()["session_id"] with patch( "app.api.endpoints.l1.l1_session_service.advance_ai_build", new=AsyncMock(return_value={"node_type": "question", "id": "n1", "text": "Powered on?"}), ): - r2 = await client.post( - f"/api/v1/l1/sessions/{sid}/next-node", - json={}, - headers=_auth(user), - ) - assert r2.status_code == 200 + r2 = await client.post(f"/api/v1/l1/sessions/{sid}/next-node", + json={}, headers=info["headers"]) + assert r2.status_code == 200, r2.text assert r2.json()["node"]["node_type"] == "question" @pytest.mark.asyncio -async def test_escalations_lists_escalated_sessions_for_engineer( - client: AsyncClient, test_db: AsyncSession -): - """GET /l1/escalations returns escalated L1 sessions; requires engineer-or-above.""" - from datetime import datetime, timezone - from app.models.l1_walk_session import L1WalkSession - - account, engineer = await _seed_user(test_db, account_role="engineer") +async def test_escalations_lists_escalated_sessions_for_engineer(client: AsyncClient, test_db: AsyncSession): + """GET /l1/escalations returns escalated L1 sessions for an engineer-or-above user.""" + info = await _make_user(client, test_db, email="aib_eng@example.com", account_role="engineer") now = datetime.now(timezone.utc) sess = L1WalkSession( - account_id=account.id, - created_by_user_id=engineer.id, + account_id=info["account_id"], + created_by_user_id=info["user_id"], ticket_id="t-esc", ticket_kind="internal", session_kind="ai_build", @@ -150,17 +144,15 @@ async def test_escalations_lists_escalated_sessions_for_engineer( escalated_at=now, ) test_db.add(sess) - await test_db.flush() - - r = await client.get("/api/v1/l1/escalations", headers=_auth(engineer)) - assert r.status_code == 200 - rows = r.json() - assert any(row["id"] == str(sess.id) for row in rows) + await test_db.commit() + r = await client.get("/api/v1/l1/escalations", headers=info["headers"]) + assert r.status_code == 200, r.text + assert any(row["id"] == str(sess.id) for row in r.json()) @pytest.mark.asyncio async def test_escalations_forbidden_for_l1_tech(client: AsyncClient, test_db: AsyncSession): """An l1_tech (not engineer-or-above) is rejected from the escalations queue.""" - account, l1 = await _seed_user(test_db, account_role="l1_tech") - r = await client.get("/api/v1/l1/escalations", headers=_auth(l1)) - assert r.status_code == 403 + info = await _make_user(client, test_db, email="aib_l1@example.com", account_role="l1_tech") + r = await client.get("/api/v1/l1/escalations", headers=info["headers"]) + assert r.status_code == 403, r.text