f84b868d13
Unified park/escalate handoff management with snapshot generation, AI diagnostic assessment for escalations (via _call_ai), claim workflow that reactivates sessions, PSA push via existing psa_documentation_service, and team queue query. Dual-writes to ai_sessions.escalation_package for backward compatibility. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
"""Integration tests for HandoffManager service."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.models.ai_session import AISession
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_park_handoff(client: AsyncClient, test_user, auth_headers, test_db):
|
|
"""Parking a session creates a handoff with snapshot."""
|
|
session = AISession(
|
|
user_id=test_user["user_data"]["id"],
|
|
account_id=test_user["user_data"]["account_id"],
|
|
session_type="guided",
|
|
intake_type="free_text",
|
|
intake_content={"text": "test"},
|
|
status="active",
|
|
confidence_tier="discovery",
|
|
conversation_messages=[{"role": "user", "content": "help me"}],
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
from app.services.handoff_manager import HandoffManager
|
|
manager = HandoffManager(test_db)
|
|
|
|
handoff = await manager.create_handoff(
|
|
session_id=session.id,
|
|
intent="park",
|
|
engineer_notes="Waiting for client to provide logs",
|
|
user_id=test_user["user_data"]["id"],
|
|
)
|
|
|
|
assert handoff is not None
|
|
assert handoff.intent == "park"
|
|
assert handoff.engineer_notes == "Waiting for client to provide logs"
|
|
assert handoff.snapshot is not None
|
|
|
|
# Session should be paused
|
|
await test_db.refresh(session)
|
|
assert session.status == "paused"
|
|
assert session.handoff_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_escalate_handoff(client: AsyncClient, test_user, auth_headers, test_db):
|
|
"""Escalating creates handoff + dual-writes to escalation_package."""
|
|
session = AISession(
|
|
user_id=test_user["user_data"]["id"],
|
|
account_id=test_user["user_data"]["account_id"],
|
|
session_type="guided",
|
|
intake_type="free_text",
|
|
intake_content={"text": "test"},
|
|
status="active",
|
|
confidence_tier="discovery",
|
|
conversation_messages=[],
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
from app.services.handoff_manager import HandoffManager
|
|
manager = HandoffManager(test_db)
|
|
|
|
handoff = await manager.create_handoff(
|
|
session_id=session.id,
|
|
intent="escalate",
|
|
engineer_notes="Need senior help",
|
|
user_id=test_user["user_data"]["id"],
|
|
)
|
|
|
|
assert handoff.intent == "escalate"
|
|
|
|
# Dual-write check
|
|
await test_db.refresh(session)
|
|
assert session.status == "escalated"
|
|
assert session.escalation_package is not None
|
|
assert "branch_map" in session.escalation_package or "snapshot" in session.escalation_package
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_session(client: AsyncClient, test_user, test_admin, auth_headers, test_db):
|
|
"""Claiming a handoff sets claimed_by and reactivates session."""
|
|
session = AISession(
|
|
user_id=test_user["user_data"]["id"],
|
|
account_id=test_user["user_data"]["account_id"],
|
|
session_type="guided",
|
|
intake_type="free_text",
|
|
intake_content={"text": "test"},
|
|
status="active",
|
|
confidence_tier="discovery",
|
|
conversation_messages=[],
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
from app.services.handoff_manager import HandoffManager
|
|
manager = HandoffManager(test_db)
|
|
|
|
handoff = await manager.create_handoff(
|
|
session_id=session.id,
|
|
intent="escalate",
|
|
engineer_notes="Need help",
|
|
user_id=test_user["user_data"]["id"],
|
|
)
|
|
|
|
claimed = await manager.claim_session(
|
|
handoff_id=handoff.id,
|
|
claiming_user_id=test_admin["user_data"]["id"],
|
|
)
|
|
|
|
assert claimed.claimed_by == test_admin["user_data"]["id"]
|
|
assert claimed.claimed_at is not None
|
|
|
|
await test_db.refresh(session)
|
|
assert session.status == "active"
|