chihlasm/resolutionflow
1
0
Fork 0
Code Issues 23 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

test(l1): rewrite AI-build API tests on proven register/login/subscription helpers

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Michael Chihlas
2026-05-30 19:33:36 -04:00
parent 633a208742
commit b57089d523
1 changed files with 74 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
backend/tests/test_l1_api_ai_build.py
View File

@@ -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