Called by the inline Script Builder tab on Submit. Writes ai_drafted_script + ai_drafted_parameters to the fix without stamping applied_at (a draft is not an application — that's §5 of the Phase 9 spec). Bumps state_version so Resolve/Escalate preview bundles regenerate. 409 on terminal fix status. 404 on wrong session. 422 on empty script. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
"""Integration tests for PATCH /ai-sessions/{sid}/suggested-fixes/{fid}/script."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from uuid import UUID, uuid4
|
|
|
|
from app.models.ai_session import AISession
|
|
from app.models.session_suggested_fix import SessionSuggestedFix
|
|
|
|
|
|
async def _make_session_with_fix(
|
|
test_db, user, *, status: str = "proposed", with_script: bool = False,
|
|
) -> tuple[str, str]:
|
|
"""Create a pilot session + suggested fix for tests. Returns (sid, fid)."""
|
|
session = AISession(
|
|
id=uuid4(),
|
|
user_id=user["user_data"]["id"],
|
|
account_id=user["user_data"]["account_id"],
|
|
session_type="tshoot",
|
|
intake_type="psa_ticket",
|
|
intake_content={},
|
|
title="QA",
|
|
status="active",
|
|
confidence_tier="exploring",
|
|
confidence_score=0.0,
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
fix = SessionSuggestedFix(
|
|
id=uuid4(),
|
|
session_id=session.id,
|
|
account_id=user["user_data"]["account_id"],
|
|
title="QA: test fix",
|
|
description="desc",
|
|
confidence_pct=80,
|
|
status=status,
|
|
ai_drafted_script="pre-existing" if with_script else None,
|
|
)
|
|
test_db.add(fix)
|
|
await test_db.commit()
|
|
return str(session.id), str(fix.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_script_happy_path(
|
|
client: AsyncClient, test_user, auth_headers, test_db
|
|
):
|
|
sid, fid = await _make_session_with_fix(test_db, test_user)
|
|
r = await client.patch(
|
|
f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
|
|
json={"ai_drafted_script": "Write-Host 'hello'"},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ai_drafted_script"] == "Write-Host 'hello'"
|
|
assert body["applied_at"] is None # draft != apply
|
|
assert body["status"] == "proposed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_script_bumps_state_version(
|
|
client: AsyncClient, test_user, auth_headers, test_db
|
|
):
|
|
sid, fid = await _make_session_with_fix(test_db, test_user)
|
|
before = await test_db.scalar(
|
|
select(AISession.state_version).where(AISession.id == UUID(sid))
|
|
)
|
|
r = await client.patch(
|
|
f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
|
|
json={"ai_drafted_script": "echo hi"},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 200
|
|
after = await test_db.scalar(
|
|
select(AISession.state_version).where(AISession.id == UUID(sid))
|
|
)
|
|
assert after == (before or 0) + 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_script_rejects_terminal_fix(
|
|
client: AsyncClient, test_user, auth_headers, test_db
|
|
):
|
|
sid, fid = await _make_session_with_fix(test_db, test_user, status="applied_success")
|
|
r = await client.patch(
|
|
f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
|
|
json={"ai_drafted_script": "echo hi"},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_script_rejects_empty_body(
|
|
client: AsyncClient, test_user, auth_headers, test_db
|
|
):
|
|
sid, fid = await _make_session_with_fix(test_db, test_user)
|
|
r = await client.patch(
|
|
f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
|
|
json={"ai_drafted_script": ""},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 422 # pydantic min_length=1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_script_404_on_wrong_session(
|
|
client: AsyncClient, test_user, auth_headers, test_db
|
|
):
|
|
_, fid = await _make_session_with_fix(test_db, test_user)
|
|
wrong_sid = str(uuid4())
|
|
r = await client.patch(
|
|
f"/api/v1/ai-sessions/{wrong_sid}/suggested-fixes/{fid}/script",
|
|
json={"ai_drafted_script": "echo hi"},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 404
|