vignette/apps/api/app/test_admin_affect.py

"""관리자 감정 관측 조회의 권한·페이지·영속 경계 검증."""

from __future__ import annotations

import unittest
from datetime import datetime, timezone
from unittest.mock import AsyncMock, patch
from uuid import UUID

from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient

from .contracts.admin_affect import (
    AdminAffectRuntimeResponse,
    AdminAffectSessionListResponse,
)
from .deps import Principal, Role, get_current_principal
from .routes import admin as admin_routes
from .services import admin_affect


SESSION_ID = UUID("00000000-0000-0000-0000-000000000101")
TURN_ID = UUID("00000000-0000-0000-0000-000000000201")
ADMIN_ID = "00000000-0000-0000-0000-000000000901"
OBSERVED_AT = datetime(2026, 9, 23, 1, 2, 3, tzinfo=timezone.utc)


class _Acquire:
    def __init__(self, conn: object) -> None:
        self.conn = conn

    async def __aenter__(self) -> object:
        return self.conn

    async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
        return None


def _trace(*, seq: int) -> dict[str, object]:
    dimensions = []
    for key in (
        "anxiety",
        "sadness",
        "anger",
        "shame",
        "guilt",
        "loneliness",
        "relief",
        "hope",
        "trust",
    ):
        dimensions.append(
            {
                "key": key,
                "before": 0.4,
                "target": 0.6,
                "after": 0.47,
                "confidence": 0.8,
                "probabilities": [0.05, 0.1, 0.2, 0.35, 0.3],
                "decision": "accepted",
            }
        )
    return {
        "schema_version": 1,
        "provider": "openrouter",
        "model": "~typesafe/jev-latest",
        "latency_ms": 91,
        "input_tokens": 12,
        "output_tokens": 18,
        "cost_usd": None,
        "turn_seq": seq,
        "policy": {
            "version": "jev-affect-v1",
            "min_confidence": 0.65,
            "accepted_alpha": 0.35,
            "accepted_cap": 0.15,
            "tentative_alpha": 0.15,
            "tentative_cap": 0.075,
            "tentative_confidence_floor": 0.35,
            "adjacent_probability_threshold": 0.8,
        },
        "context": {
            "stage": "탐색",
            "resistance": 0.4,
            "effective_openness": 0.6,
            "rapport_credit": 0.2,
        },
        "dimensions": dimensions,
    }


def _trace_v2(*, seq: int) -> dict[str, object]:
    dimensions = []
    reaction = []
    for key in (
        "anxiety",
        "sadness",
        "anger",
        "shame",
        "guilt",
        "loneliness",
        "relief",
        "hope",
        "trust",
    ):
        dimensions.append(
            {
                "key": key,
                "before": 0.4,
                "target": 0.6,
                "after": 0.47,
                "confidence": 0.8,
                "probabilities": [0.05, 0.1, 0.2, 0.35, 0.3],
                "decision": "accepted",
            }
        )
        reaction.append({"key": key, "value": 0.6, "included": True})
    return {
        "schema_version": 2,
        "provider": "openrouter",
        "model": "~typesafe/jev-latest",
        "latency_ms": 91,
        "input_tokens": 12,
        "output_tokens": 18,
        "cost_usd": None,
        "turn_seq": seq,
        "policy": {
            "version": "jev-affect-v2",
            "min_confidence": 0.65,
            "tentative_confidence_floor": 0.35,
            "adjacent_probability_threshold": 0.8,
            "worsening_accepted_alpha": 0.35,
            "worsening_accepted_cap": 0.15,
            "worsening_tentative_alpha": 0.15,
            "worsening_tentative_cap": 0.075,
            "recovery_accepted_alpha": 0.20,
            "recovery_accepted_cap": 0.08,
            "recovery_tentative_alpha": 0.08,
            "recovery_tentative_cap": 0.04,
        },
        "context": {
            "stage": "탐색",
            "resistance": 0.4,
            "effective_openness": 0.6,
            "rapport_credit": 0.2,
        },
        "dimensions": dimensions,
        "appraisal": [
            {
                "key": "a_judged",
                "kind": "noul",
                "probability": 0.8,
                "confidence": 0.7,
                "decision": "true",
            },
            {
                "key": "a_coping",
                "kind": "choice",
                "choice": "manageable",
                "probabilities": {
                    "nothing_asked": 0.1,
                    "manageable": 0.7,
                    "stretch": 0.1,
                    "overwhelming": 0.1,
                },
                "confidence": 0.6,
                "decision": "manageable",
            },
        ],
        "reaction": reaction,
        "expression": {
            "behavior": {
                "choice": "disclose_more",
                "probabilities": {"disclose_more": 0.7, "hold_core": 0.3},
                "confidence": 0.6,
                "decision": "disclose_more",
            },
            "gated_behavior": "disclose_more",
            "gate_reason": None,
            "stance": "engage",
            "display": {
                "choice": "as_felt",
                "probabilities": {"as_felt": 0.7, "masked": 0.3},
                "confidence": 0.6,
                "decision": "as_felt",
            },
            "disclose_ready": {"probability": 0.7, "confidence": 0.6, "decision": "true"},
            "hidden_gap": False,
        },
        "sore_spot_count": 1,
    }


def _admin_principal() -> Principal:
    return Principal(user_id=ADMIN_ID, role=Role.ADMIN)


class AdminAffectStoreTest(unittest.IsolatedAsyncioTestCase):
    async def test_list_uses_admin_scope_and_excludes_learner_identity(self) -> None:
        case = self

        class Conn:
            async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
                case.assertIn("FROM app.sessions", query)
                case.assertNotIn("learner", query.lower())
                return {"total": 1}

            async def fetch(self, query: str, *args: object) -> list[dict[str, object]]:
                case.assertIn("app.client_affect_trace", query)
                case.assertNotIn("learner_id", query)
                case.assertNotIn("app.app_user", query)
                case.assertNotIn(" text", query.lower())
                case.assertEqual(args, (30, 0))
                return [
                    {
                        "session_id": SESSION_ID,
                        "persona_code": "P4",
                        "started_at": OBSERVED_AT,
                        "ended": False,
                        "trace_count": 2,
                    }
                ]

        runtime = AdminAffectRuntimeResponse(
            enabled=True,
            provider="openrouter",
            model="~typesafe/jev-latest",
            configured=True,
        )
        with (
            patch.object(admin_affect, "acquire", return_value=_Acquire(Conn())) as acquire,
            patch.object(admin_affect, "runtime_snapshot", return_value=runtime),
        ):
            response = await admin_affect.list_sessions(
                user_id=ADMIN_ID,
                limit=30,
                offset=0,
            )

        acquire.assert_called_once_with(role="admin", user_id=ADMIN_ID)
        self.assertEqual(response.total, 1)
        self.assertEqual(response.sessions[0].session_id, str(SESSION_ID))
        self.assertEqual(response.sessions[0].trace_count, 2)

    async def test_list_preserves_total_beyond_the_last_page(self) -> None:
        case = self

        class Conn:
            async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
                return {"total": 4}

            async def fetch(self, query: str, *args: object) -> list[dict[str, object]]:
                case.assertEqual(args, (30, 30))
                return []

        with patch.object(admin_affect, "acquire", return_value=_Acquire(Conn())):
            response = await admin_affect.list_sessions(
                user_id=ADMIN_ID,
                limit=30,
                offset=30,
            )

        self.assertEqual(response.total, 4)
        self.assertEqual(response.sessions, [])

    async def test_detail_returns_actual_snapshot_and_ascending_page(self) -> None:
        case = self

        class Conn:
            async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
                if "FROM app.sessions AS s" in query:
                    case.assertNotIn("learner", query.lower())
                    case.assertNotIn("text", query.lower())
                    return {
                        "session_id": SESSION_ID,
                        "persona_code": "P4",
                        "affect_state": {
                            "emotion_anxiety": 0.6,
                            "emotion_hope": 0.2,
                            "emotion_anger": 1.5,
                        },
                    }
                case.assertIn("count(*)", query)
                return {"total_traces": 4}

            async def fetch(self, query: str, *args: object) -> list[dict[str, object]]:
                case.assertIn("turn.seq < $2", query)
                case.assertIn("ORDER BY turn.seq DESC", query)
                case.assertNotIn("turn.text", query)
                case.assertEqual(args, (SESSION_ID, 9, 3))
                return [
                    {"turn_id": TURN_ID, "seq": 8, "created_at": OBSERVED_AT, "trace": _trace(seq=8)},
                    {"turn_id": TURN_ID, "seq": 7, "created_at": OBSERVED_AT, "trace": _trace(seq=7)},
                    {"turn_id": TURN_ID, "seq": 6, "created_at": OBSERVED_AT, "trace": _trace(seq=6)},
                ]

        with patch.object(admin_affect, "acquire", return_value=_Acquire(Conn())) as acquire:
            response = await admin_affect.get_session_detail(
                user_id=ADMIN_ID,
                session_id=SESSION_ID,
                limit=2,
                before_seq=9,
            )

        acquire.assert_called_once_with(role="admin", user_id=ADMIN_ID)
        self.assertEqual([record.seq for record in response.traces], [7, 8])
        self.assertTrue(response.has_more)
        self.assertEqual(response.total_traces, 4)
        self.assertEqual(response.current_emotions["anxiety"], 0.6)
        self.assertEqual(response.current_emotions["hope"], 0.2)
        self.assertIsNone(response.current_emotions["sadness"])
        self.assertIsNone(response.current_emotions["anger"])

    async def test_detail_parses_mixed_v1_and_v2_trace_rows(self) -> None:
        case = self

        class Conn:
            async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
                if "FROM app.sessions AS s" in query:
                    return {
                        "session_id": SESSION_ID,
                        "persona_code": "P4",
                        "affect_state": {"emotion_anxiety": 0.5},
                    }
                return {"total_traces": 2}

            async def fetch(self, query: str, *args: object) -> list[dict[str, object]]:
                return [
                    {"turn_id": TURN_ID, "seq": 5, "created_at": OBSERVED_AT, "trace": _trace_v2(seq=5)},
                    {"turn_id": TURN_ID, "seq": 4, "created_at": OBSERVED_AT, "trace": _trace(seq=4)},
                ]

        with patch.object(admin_affect, "acquire", return_value=_Acquire(Conn())):
            response = await admin_affect.get_session_detail(
                user_id=ADMIN_ID,
                session_id=SESSION_ID,
                limit=100,
                before_seq=None,
            )

        case.assertEqual([record.trace.schema_version for record in response.traces], [1, 2])
        v2_record = response.traces[1]
        case.assertEqual(v2_record.trace.sore_spot_count, 1)
        case.assertEqual(v2_record.trace.expression.gated_behavior, "disclose_more")

    async def test_detail_rejects_trace_with_unsupported_schema_version(self) -> None:
        class Conn:
            async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
                if "FROM app.sessions AS s" in query:
                    return {"session_id": SESSION_ID, "persona_code": "P4", "affect_state": {}}
                return {"total_traces": 1}

            async def fetch(self, query: str, *args: object) -> list[dict[str, object]]:
                broken = _trace(seq=1)
                broken["schema_version"] = 3
                return [{"turn_id": TURN_ID, "seq": 1, "created_at": OBSERVED_AT, "trace": broken}]

        with patch.object(admin_affect, "acquire", return_value=_Acquire(Conn())):
            with self.assertRaises(admin_affect.AdminAffectPersistenceError):
                await admin_affect.get_session_detail(
                    user_id=ADMIN_ID,
                    session_id=SESSION_ID,
                    limit=100,
                    before_seq=None,
                )

    async def test_detail_keeps_trace_free_legacy_session_observable(self) -> None:
        class Conn:
            async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
                if "FROM app.sessions AS s" in query:
                    return {
                        "session_id": SESSION_ID,
                        "persona_code": "P4",
                        "affect_state": {},
                    }
                return {"total_traces": 0}

            async def fetch(self, query: str, *args: object) -> list[dict[str, object]]:
                return []

        with patch.object(admin_affect, "acquire", return_value=_Acquire(Conn())):
            response = await admin_affect.get_session_detail(
                user_id=ADMIN_ID,
                session_id=SESSION_ID,
                limit=100,
                before_seq=None,
            )

        self.assertEqual(response.traces, [])
        self.assertEqual(response.total_traces, 0)
        self.assertFalse(response.has_more)
        self.assertTrue(all(value is None for value in response.current_emotions.values()))

    async def test_detail_maps_missing_session_and_database_failure_separately(self) -> None:
        class MissingConn:
            async def fetchrow(self, query: str, *args: object) -> None:
                return None

        with patch.object(admin_affect, "acquire", return_value=_Acquire(MissingConn())):
            with self.assertRaises(admin_affect.AdminAffectSessionNotFoundError):
                await admin_affect.get_session_detail(
                    user_id=ADMIN_ID,
                    session_id=SESSION_ID,
                    limit=100,
                    before_seq=None,
                )

        with patch.object(admin_affect, "acquire", side_effect=RuntimeError("database down")):
            with self.assertRaises(admin_affect.AdminAffectPersistenceError):
                await admin_affect.list_sessions(user_id=ADMIN_ID, limit=30, offset=0)


class AdminAffectHttpBoundaryTest(unittest.TestCase):
    def _client(self, principal: Principal) -> TestClient:
        app = FastAPI()
        app.include_router(admin_routes.router)
        app.dependency_overrides[get_current_principal] = lambda: principal
        return TestClient(app)

    def test_non_admin_is_rejected_before_service_access(self) -> None:
        principal = Principal(user_id="learner-1", role=Role.LEARNER)
        with patch.object(admin_affect, "list_sessions", AsyncMock()) as list_sessions:
            response = self._client(principal).get("/admin/affect/sessions")

        self.assertEqual(response.status_code, 403)
        list_sessions.assert_not_awaited()

    def test_invalid_uuid_and_pagination_are_validation_errors(self) -> None:
        client = self._client(_admin_principal())

        self.assertEqual(client.get("/admin/affect/sessions?limit=0").status_code, 422)
        self.assertEqual(
            client.get("/admin/affect/sessions/not-a-uuid").status_code,
            422,
        )
        self.assertEqual(
            client.get(f"/admin/affect/sessions/{SESSION_ID}?before_seq=0").status_code,
            422,
        )

    def test_route_maps_persistence_error_to_503(self) -> None:
        with patch.object(
            admin_affect,
            "list_sessions",
            AsyncMock(side_effect=admin_affect.AdminAffectPersistenceError("database down")),
        ):
            response = self._client(_admin_principal()).get("/admin/affect/sessions")

        self.assertEqual(response.status_code, 503)

    def test_route_maps_missing_session_to_404(self) -> None:
        with patch.object(
            admin_affect,
            "get_session_detail",
            AsyncMock(side_effect=admin_affect.AdminAffectSessionNotFoundError("missing")),
        ):
            response = self._client(_admin_principal()).get(
                f"/admin/affect/sessions/{SESSION_ID}"
            )

        self.assertEqual(response.status_code, 404)

    def test_route_serializes_list_response(self) -> None:
        result = AdminAffectSessionListResponse(
            runtime=AdminAffectRuntimeResponse(
                enabled=True,
                provider="openrouter",
                model="~typesafe/jev-latest",
                configured=True,
            ),
            sessions=[],
            total=0,
            limit=30,
            offset=0,
        )
        with patch.object(admin_affect, "list_sessions", AsyncMock(return_value=result)) as list_sessions:
            response = self._client(_admin_principal()).get("/admin/affect/sessions")

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["runtime"]["provider"], "openrouter")
        self.assertEqual(response.json()["sessions"], [])
        list_sessions.assert_awaited_once_with(user_id=ADMIN_ID, limit=30, offset=0)