758cd61621
Service layer (production code): - branch_manager: set account_id on SessionBranch (root + fork) and ForkPoint from session.account_id; load session in create_fork for this purpose - handoff_manager: set account_id on SessionHandoff from session.account_id - ai_suggestions endpoint: set account_id on AISuggestion from current_user - steps endpoint (/feedback): set account_id on StepRating from current_user - ratings endpoint: set account_id on StepRating from current_user Test infrastructure: - conftest.py: seed PLATFORM_ACCOUNT_ID (00000000-...-0001) account after Base.metadata.create_all so global categories and gallery items have a valid FK - test_rls_isolation: add _ensure_rls_schema fixture that runs 'alembic upgrade head' before module tests — previous function-scoped test_db fixtures drop the schema, leaving the RLS tests with no tables - test_branding: create Account before User in helper functions - test_admin_gallery: set account_id=PLATFORM_ACCOUNT_ID on Tree/ScriptTemplate - test_public_templates: set account_id=PLATFORM_ACCOUNT_ID on Tree, ScriptTemplate, TreeCategory - test_resolution_outputs: set account_id=session.account_id on SessionResolutionOutput - test_analytics_phase5: set account_id on PsaPostLog - test_draft_trees: replace account_id=None with PLATFORM_ACCOUNT_ID in migration default test (NOT NULL now enforced) - test_maintenance_schedules: set account_id on other_tree - test_save_session_as_tree: set account_id on all 5 Session() constructors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""Integration tests for ResolutionOutputGenerator."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
from httpx import AsyncClient
|
|
|
|
from app.models.ai_session import AISession
|
|
from app.models.session_resolution_output import SessionResolutionOutput
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("app.services.resolution_output_generator._call_ai")
|
|
async def test_generate_all_creates_three_outputs(
|
|
mock_call_ai, client: AsyncClient, test_user, auth_headers, test_db
|
|
):
|
|
"""generate_all creates PSA notes, KB article, and client summary."""
|
|
mock_call_ai.return_value = ("Generated content here", 100, 50)
|
|
|
|
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="resolved",
|
|
confidence_tier="guided",
|
|
conversation_messages=[
|
|
{"role": "user", "content": "DNS not working"},
|
|
{"role": "assistant", "content": "Fixed by flushing DNS cache"},
|
|
],
|
|
resolution_summary="Flushed DNS cache",
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
from app.services.resolution_output_generator import ResolutionOutputGenerator
|
|
gen = ResolutionOutputGenerator(test_db)
|
|
outputs = await gen.generate_all(session.id)
|
|
|
|
assert len(outputs) == 3
|
|
types = {o.output_type for o in outputs}
|
|
assert types == {"psa_ticket_notes", "knowledge_base", "client_summary"}
|
|
assert all(o.status == "draft" for o in outputs)
|
|
assert mock_call_ai.call_count == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_output(client: AsyncClient, test_user, auth_headers, test_db):
|
|
"""Editing an output stores edited_content."""
|
|
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="resolved",
|
|
confidence_tier="guided",
|
|
conversation_messages=[],
|
|
resolution_summary="Fixed it",
|
|
)
|
|
test_db.add(session)
|
|
await test_db.flush()
|
|
|
|
output = SessionResolutionOutput(
|
|
session_id=session.id,
|
|
account_id=session.account_id,
|
|
output_type="psa_ticket_notes",
|
|
generated_content="Original notes",
|
|
status="draft",
|
|
generated_by_model="claude-sonnet-4-6",
|
|
)
|
|
test_db.add(output)
|
|
await test_db.flush()
|
|
|
|
from app.services.resolution_output_generator import ResolutionOutputGenerator
|
|
gen = ResolutionOutputGenerator(test_db)
|
|
edited = await gen.edit_output(output.id, "My edited notes")
|
|
|
|
assert edited.edited_content == "My edited notes"
|