c3f0370964
- script_template.py: add server_default to ALL NOT NULL columns so
Base.metadata.create_all matches Alembic behavior for raw SQL INSERTs
- test_session_branches_api.py: fork_reason needs 5+ chars ("test" → "testing fork")
- test_scripts.py: engineers CAN create templates (assert 201, not 403)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""API endpoint tests for session branches."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.models.ai_session import AISession
|
|
from app.models.ai_session_step import AISessionStep
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_branches_empty(client: AsyncClient, test_user, auth_headers, test_db):
|
|
"""GET /ai-sessions/{id}/branches returns empty for non-branching 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.commit()
|
|
|
|
resp = await client.get(
|
|
f"/api/v1/ai-sessions/{session.id}/branches",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["branches"] == []
|
|
assert data["active_branch_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_fork(client: AsyncClient, test_user, auth_headers, test_db):
|
|
"""POST /ai-sessions/{id}/branches/fork creates branches."""
|
|
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"}],
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
step = AISessionStep(
|
|
session_id=session.id, step_order=0, step_type="question",
|
|
content={"text": "test"}, confidence_at_step=0.5,
|
|
)
|
|
test_db.add(step)
|
|
await test_db.commit()
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/ai-sessions/{session.id}/branches/fork",
|
|
headers=auth_headers,
|
|
json={
|
|
"fork_reason": "Two possible causes",
|
|
"options": [
|
|
{"label": "Network issue", "description": "Check connectivity"},
|
|
{"label": "DNS problem", "description": "Check DNS"},
|
|
],
|
|
},
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert len(data["options"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_branch(client: AsyncClient, test_user, auth_headers, test_db):
|
|
"""POST /ai-sessions/{id}/branches/{bid}/switch changes active branch."""
|
|
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"}],
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
step = AISessionStep(
|
|
session_id=session.id, step_order=0, step_type="question",
|
|
content={"text": "test"}, confidence_at_step=0.5,
|
|
)
|
|
test_db.add(step)
|
|
await test_db.commit()
|
|
|
|
# Create fork first
|
|
fork_resp = await client.post(
|
|
f"/api/v1/ai-sessions/{session.id}/branches/fork",
|
|
headers=auth_headers,
|
|
json={
|
|
"fork_reason": "testing fork",
|
|
"options": [
|
|
{"label": "A", "description": "a"},
|
|
{"label": "B", "description": "b"},
|
|
],
|
|
},
|
|
)
|
|
assert fork_resp.status_code == 201, fork_resp.text
|
|
fork_data = fork_resp.json()
|
|
branch_b_id = fork_data["options"][1]["branch_id"]
|
|
|
|
# Switch to branch B
|
|
resp = await client.post(
|
|
f"/api/v1/ai-sessions/{session.id}/branches/{branch_b_id}/switch",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["active_branch_id"] == branch_b_id
|