"""관리자 감정 관측 조회의 권한·페이지·영속 경계 검증.""" 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 _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_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)