"""API + service tests for Phase 5 inline Script Generator integration. Covers: - TemplateExtractionService: well-formed, fallback on bad output, missing-key fallback. - /suggested-fixes/{fix_id}/decision side effects: * one_off returns rendered_script, no draft_templates row. * draft_template returns rendered_script + draft_template_id, draft persisted. * build_template returns redirect_path. * dismissed (Phase 3) still works. - 400 when ai_drafted_script is missing for a non-template fix. """ from __future__ import annotations from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient from sqlalchemy import select from app.api.endpoints.session_suggested_fixes import _clear_preview_cache_for_tests from app.models.ai_session import AISession from app.models.draft_template import DraftTemplate from app.models.session_suggested_fix import SessionSuggestedFix from app.services.template_extraction_service import ( _fallback, _parse_response, extract_parameters, ) @pytest.fixture(autouse=True) def _isolate_preview_cache(): _clear_preview_cache_for_tests() yield _clear_preview_cache_for_tests() async def _make_session_with_fix( test_db, user, *, with_template_id: bool = False, with_drafted_script: bool = True, ) -> tuple[AISession, SessionSuggestedFix]: session = AISession( user_id=user["user_data"]["id"], account_id=user["user_data"]["account_id"], session_type="chat", intake_type="free_text", intake_content={"text": "phase 5 test"}, status="active", confidence_tier="discovery", conversation_messages=[], ) test_db.add(session) await test_db.flush() fix = SessionSuggestedFix( session_id=session.id, account_id=session.account_id, title="Reset cached creds", description="Clearing the cached credential...", confidence_pct=85, ai_drafted_script=( 'cmdkey /delete:"outlook.office365.com"\n' 'Restart-Process -Name OUTLOOK' ) if with_drafted_script else None, ai_drafted_parameters={"target_user": "jsmith"} if with_drafted_script else None, ) test_db.add(fix) await test_db.commit() await test_db.refresh(session) await test_db.refresh(fix) return session, fix # ── TemplateExtractionService: parse + fallback ─────────────────────────── class TestParseResponse: def test_well_formed(self): raw = ( '{"parameters": [{"key":"host","label":"Host","type":"text",' '"inferred_from":"session fact"}],' '"templated_body":"Get-Service -ComputerName {{ host }}"}' ) result = _parse_response(raw) assert result is not None assert len(result["parameters"]) == 1 assert result["parameters"][0]["key"] == "host" assert result["templated_body"].endswith("{{ host }}") def test_strips_fences(self): raw = '```json\n{"parameters": [], "templated_body": "x"}\n```' result = _parse_response(raw) assert result is not None and result["parameters"] == [] def test_invalid_key_dropped(self): # Capital letters and dashes in key names violate snake_case — drop. raw = ( '{"parameters":[{"key":"BadKey-Name","type":"text"}],' '"templated_body":"x"}' ) result = _parse_response(raw) assert result is not None and result["parameters"] == [] def test_unknown_type_falls_back_to_text(self): raw = ( '{"parameters":[{"key":"x","type":"weird"}],"templated_body":"x"}' ) result = _parse_response(raw) assert result is not None and result["parameters"][0]["type"] == "text" def test_malformed_json_returns_none(self): assert _parse_response("not json") is None def test_non_dict_returns_none(self): assert _parse_response('["a","b"]') is None @pytest.mark.asyncio async def test_extract_parameters_round_trip_failure_uses_fallback(): """Templated_body referencing an undeclared placeholder triggers fallback.""" fake_provider = AsyncMock() fake_provider.generate_json = AsyncMock(return_value=( # Declares parameter `host` but the body references `port` too. '{"parameters":[{"key":"host","label":"Host","type":"text"}],' '"templated_body":"Get-Service -ComputerName {{ host }} -Port {{ port }}"}', 100, 50, )) with patch( "app.services.template_extraction_service.get_ai_provider", return_value=fake_provider, ): result = await extract_parameters( script_body="Get-Service -ComputerName srv01 -Port 8080", ) fb = _fallback("Get-Service -ComputerName srv01 -Port 8080") assert result == fb @pytest.mark.asyncio async def test_extract_parameters_llm_exception_uses_fallback(): fake_provider = AsyncMock() fake_provider.generate_json = AsyncMock(side_effect=RuntimeError("boom")) with patch( "app.services.template_extraction_service.get_ai_provider", return_value=fake_provider, ): result = await extract_parameters(script_body="echo hello") assert result == _fallback("echo hello") # ── Decision endpoint: one_off ───────────────────────────────────────────── @pytest.mark.asyncio async def test_one_off_returns_rendered_script_no_draft( client: AsyncClient, test_user, auth_headers, test_db ): session, fix = await _make_session_with_fix(test_db, test_user) r = await client.post( f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision", headers=auth_headers, json={"decision": "one_off"}, ) assert r.status_code == 200 body = r.json() assert body["user_decision"] == "one_off" assert body["rendered_script"] is not None assert "cmdkey" in body["rendered_script"] assert body["draft_template_id"] is None assert body["redirect_path"] is None # No draft_templates row should have been created. rows = ( await test_db.execute(select(DraftTemplate).where(DraftTemplate.source_session_id == session.id)) ).scalars().all() assert list(rows) == [] # ── Decision endpoint: draft_template ───────────────────────────────────── @pytest.mark.asyncio async def test_draft_template_creates_draft_with_extracted_params( client: AsyncClient, test_user, auth_headers, test_db ): session, fix = await _make_session_with_fix(test_db, test_user) fake_provider = AsyncMock() fake_provider.generate_json = AsyncMock(return_value=( '{"parameters":[{"key":"target_user","label":"Target User","type":"text",' '"inferred_from":"session fact"}],' '"templated_body":"cmdkey /delete:\\"outlook.office365.com\\"\\n' 'Restart-Process -Name OUTLOOK"}', 80, 60, )) with patch( "app.services.template_extraction_service.get_ai_provider", return_value=fake_provider, ): r = await client.post( f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision", headers=auth_headers, json={"decision": "draft_template"}, ) assert r.status_code == 200 body = r.json() assert body["user_decision"] == "draft_template" assert body["rendered_script"] is not None assert body["draft_template_id"] is not None assert body["redirect_path"] is None drafts = ( await test_db.execute(select(DraftTemplate).where(DraftTemplate.source_session_id == session.id)) ).scalars().all() drafts = list(drafts) assert len(drafts) == 1 draft = drafts[0] assert draft.status == "pending" assert draft.proposed_name == fix.title proposed = draft.proposed_parameters.get("parameters") if isinstance(draft.proposed_parameters, dict) else None assert isinstance(proposed, list) and len(proposed) == 1 assert proposed[0]["key"] == "target_user" # ── Decision endpoint: build_template ───────────────────────────────────── @pytest.mark.asyncio async def test_build_template_returns_redirect_path( client: AsyncClient, test_user, auth_headers, test_db ): session, fix = await _make_session_with_fix(test_db, test_user) r = await client.post( f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision", headers=auth_headers, json={"decision": "build_template"}, ) assert r.status_code == 200 body = r.json() assert body["redirect_path"] is not None assert str(session.id) in body["redirect_path"] assert str(fix.id) in body["redirect_path"] # ── Decision endpoint: 400 when no drafted script ───────────────────────── @pytest.mark.asyncio async def test_one_off_without_drafted_script_returns_400( client: AsyncClient, test_user, auth_headers, test_db ): """A template-matched fix takes the dedicated /scripts/generate path; trying to one_off it via this endpoint without an ai_drafted_script must surface a clear client-error, not silently render nothing.""" session, fix = await _make_session_with_fix(test_db, test_user, with_drafted_script=False) r = await client.post( f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision", headers=auth_headers, json={"decision": "one_off"}, ) assert r.status_code == 400 assert "ai_drafted_script" in r.json()["detail"] # ── Decision endpoint: edited script overrides ai_drafted_script ────────── @pytest.mark.asyncio async def test_edited_script_overrides_ai_drafted( client: AsyncClient, test_user, auth_headers, test_db ): session, fix = await _make_session_with_fix(test_db, test_user) r = await client.post( f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision", headers=auth_headers, json={ "decision": "one_off", "edited_script": "Get-Service -Name Dnscache", }, ) assert r.status_code == 200 assert r.json()["rendered_script"] == "Get-Service -Name Dnscache"