085460b5e0
SSOT 대시보드:
- 한신대 기술분석 PDF(19쪽) 정합성 분석 + 이번 세션 발견 섹션 추가
- 섹션 폴드아웃(접기)·상단 목차(드릴다운)·모두 펼치기/접기 — 내용 보존, 레이아웃만 정리
페르소나 반응 강화('저항·반응 조절' 핵심 차별):
- PersonaCard.triggers(역린) 필드 + CCD 핵심상처 파생 역린 블록
- L0에 무례·모욕·조롱 시 현실적 동맹 균열 반응 지침
버그·성능 수정(라이브/E2E로 포착):
- 게이트웨이 페르소나 격리: --append-system-prompt를 --system-prompt(교체)로 + --exclude-dynamic-system-prompt-sections (내담자 캐릭터 붕괴·개발맥락 누출 차단)
- RAG: 임베더 동기 로드(약 7-13초)를 _warm_rag_caches 백그라운드 warm으로(세션 생성 블로킹 회귀 수정)
- voice TTS RMS 데드힌트 제거, init_state OpennessParams 파라미터객체화
- 한국어 PII(날짜·금액·주소) 마스킹 보강
- 레이아웃 시각 게이트: 폼 컨트롤 값 스크롤 오탐 제외(7/7)
검증: 백엔드 84/84, E2E 42(데스크톱 27·모바일 11·아바타 4), 시각 게이트 7/7
498 lines
19 KiB
Python
498 lines
19 KiB
Python
"""Regression tests for persona approval and faculty review boundaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from . import persona_repository
|
|
from .deps import Principal, Role
|
|
from .persona_repository import PersonaReviewItem
|
|
from .routes import personas, sessions
|
|
from .services import persona as persona_service
|
|
|
|
|
|
def _principal(role: Role = Role.LEARNER) -> Principal:
|
|
return Principal(
|
|
user_id="00000000-0000-0000-0000-000000000901",
|
|
role=role,
|
|
cohort_ids=["cohort-a"] if role == Role.TEACHER else [],
|
|
email=f"{role.value}@example.test",
|
|
display_name=role.value.title(),
|
|
)
|
|
|
|
|
|
def _card_row(
|
|
card: persona_service.PersonaCard,
|
|
*,
|
|
persona_id: str,
|
|
status: str,
|
|
version: int = 1,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"persona_id": persona_id,
|
|
"code": card.code,
|
|
"version": version,
|
|
"status": status,
|
|
"display_name": card.display_name,
|
|
"difficulty": card.difficulty,
|
|
"theory_target": list(card.theory_target),
|
|
"demographics": dict(card.demographics),
|
|
"presenting": dict(card.presenting),
|
|
"history": dict(card.history),
|
|
"big5": dict(card.big5),
|
|
"resistance": dict(card.resistance),
|
|
"speech_style": dict(card.speech_style),
|
|
"affect_baseline": dict(card.affect_baseline),
|
|
"ccd": dict(card.ccd),
|
|
"dsm5_dimensional": dict(card.dsm5_dimensional),
|
|
"source_provenance": card.source_provenance,
|
|
"is_synthetic": card.is_synthetic,
|
|
"created_at": "2026-01-01T00:00:00",
|
|
"approved_at": "2026-01-02T00:00:00" if status == "approved" else None,
|
|
}
|
|
|
|
|
|
class _Acquire:
|
|
def __init__(self, conn: "_PersonaCardConn") -> None:
|
|
self.conn = conn
|
|
|
|
async def __aenter__(self) -> "_PersonaCardConn":
|
|
return self.conn
|
|
|
|
async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
|
|
return None
|
|
|
|
|
|
class _PersonaCardConn:
|
|
def __init__(self, rows: list[dict[str, Any]]) -> None:
|
|
self.rows = rows
|
|
self.fetch_calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
self.fetchrow_calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
self.execute_calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
|
|
async def fetch(self, query: str, *args: Any) -> list[dict[str, Any]]:
|
|
self.fetch_calls.append((query, args))
|
|
return self._filter_rows(query, args)
|
|
|
|
async def fetchrow(self, query: str, *args: Any) -> dict[str, Any] | None:
|
|
self.fetchrow_calls.append((query, args))
|
|
if "UPDATE app.persona_card" in query:
|
|
persona_id = str(args[0])
|
|
next_status = str(args[1])
|
|
approved_by = args[2]
|
|
for row in self.rows:
|
|
if row["persona_id"] != persona_id or row["status"] not in {"draft", "review"}:
|
|
continue
|
|
row["status"] = next_status
|
|
row["approved_by"] = approved_by
|
|
row["approved_at"] = "2026-01-03T00:00:00" if next_status == "approved" else None
|
|
return row
|
|
return None
|
|
rows = self._filter_rows(query, args)
|
|
code = str(args[0]).upper() if args else ""
|
|
matches = [row for row in rows if str(row["code"]).upper() == code]
|
|
matches.sort(key=lambda row: int(row["version"]), reverse=True)
|
|
return matches[0] if matches else None
|
|
|
|
async def execute(self, query: str, *args: Any) -> str:
|
|
self.execute_calls.append((query, args))
|
|
return "INSERT 0 1"
|
|
|
|
def _filter_rows(self, query: str, args: tuple[Any, ...]) -> list[dict[str, Any]]:
|
|
if "WHERE status = 'approved'" in query:
|
|
return [row for row in self.rows if row["status"] == "approved"]
|
|
if "status = ANY($1::text[])" in query:
|
|
statuses = {str(status) for status in args[0]}
|
|
return [row for row in self.rows if row["status"] in statuses]
|
|
return list(self.rows)
|
|
|
|
|
|
class PersonaApprovalBoundaryTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_catalog_repository_lists_only_approved_personas(self) -> None:
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P1,
|
|
persona_id="00000000-0000-0000-0000-000000000001",
|
|
status="approved",
|
|
),
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id="00000000-0000-0000-0000-000000000002",
|
|
status="draft",
|
|
),
|
|
_card_row(
|
|
persona_service.P3,
|
|
persona_id="00000000-0000-0000-0000-000000000003",
|
|
status="review",
|
|
),
|
|
]
|
|
)
|
|
acquire_calls: list[dict[str, Any]] = []
|
|
|
|
def fake_acquire(**kwargs: Any) -> _Acquire:
|
|
acquire_calls.append(kwargs)
|
|
return _Acquire(conn)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", fake_acquire),
|
|
):
|
|
result = await persona_repository.list_approved_personas()
|
|
|
|
self.assertEqual([entry.card.code for entry in result], ["P1"])
|
|
self.assertEqual(acquire_calls, [{"ai_context": True}])
|
|
self.assertIn("WHERE status = 'approved'", conn.fetch_calls[0][0])
|
|
|
|
async def test_start_lookup_ignores_draft_or_review_persona_versions(self) -> None:
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id="00000000-0000-0000-0000-000000000102",
|
|
status="draft",
|
|
version=2,
|
|
),
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id="00000000-0000-0000-0000-000000000101",
|
|
status="review",
|
|
version=1,
|
|
),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
result = await persona_repository.get_approved_persona("p2")
|
|
|
|
self.assertIsNone(result)
|
|
query, args = conn.fetchrow_calls[0]
|
|
self.assertIn("WHERE status = 'approved'", query)
|
|
self.assertEqual(args, ("P2",))
|
|
|
|
async def test_session_start_rejects_persona_without_approved_catalog_entry(self) -> None:
|
|
principal = _principal(Role.LEARNER)
|
|
|
|
with (
|
|
patch.object(sessions, "get_catalog_persona", AsyncMock(return_value=None)) as get_persona,
|
|
patch.object(
|
|
sessions.session_persistence,
|
|
"create_session",
|
|
AsyncMock(side_effect=AssertionError("draft persona must not start a session")),
|
|
) as create_session,
|
|
):
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await sessions.start_session(
|
|
sessions.SessionStartRequest(persona_code="P2"),
|
|
principal,
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 404)
|
|
self.assertIn("unknown persona P2", caught.exception.detail)
|
|
get_persona.assert_awaited_once_with("P2")
|
|
create_session.assert_not_awaited()
|
|
|
|
|
|
class PersonaReviewQueueTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_review_queue_repository_fetches_draft_and_review_for_teacher(self) -> None:
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P1,
|
|
persona_id="00000000-0000-0000-0000-000000000201",
|
|
status="approved",
|
|
),
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id="00000000-0000-0000-0000-000000000202",
|
|
status="draft",
|
|
),
|
|
_card_row(
|
|
persona_service.P3,
|
|
persona_id="00000000-0000-0000-0000-000000000203",
|
|
status="review",
|
|
),
|
|
]
|
|
)
|
|
acquire_calls: list[dict[str, Any]] = []
|
|
|
|
def fake_acquire(**kwargs: Any) -> _Acquire:
|
|
acquire_calls.append(kwargs)
|
|
return _Acquire(conn)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", fake_acquire),
|
|
):
|
|
queue = await persona_repository.list_persona_review_queue(role="teacher")
|
|
|
|
self.assertEqual([item.code for item in queue], ["P2", "P3"])
|
|
self.assertEqual([item.status for item in queue], ["draft", "review"])
|
|
self.assertEqual(acquire_calls, [{"role": "teacher"}])
|
|
query, args = conn.fetch_calls[0]
|
|
self.assertIn("status = ANY($1::text[])", query)
|
|
self.assertEqual(args, (["draft", "review"],))
|
|
|
|
async def test_review_queue_repository_rejects_learner_role(self) -> None:
|
|
with patch.object(
|
|
persona_repository,
|
|
"get_pool",
|
|
side_effect=AssertionError("learner must be rejected before DB access"),
|
|
):
|
|
with self.assertRaises(ValueError):
|
|
await persona_repository.list_persona_review_queue(role="learner")
|
|
|
|
async def test_learner_cannot_call_review_route(self) -> None:
|
|
with patch.object(
|
|
personas,
|
|
"list_persona_review_queue",
|
|
AsyncMock(side_effect=AssertionError("learner must not reach review repository")),
|
|
) as review_queue:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await personas.list_persona_reviews(_principal(Role.LEARNER))
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
review_queue.assert_not_awaited()
|
|
|
|
async def test_teacher_review_route_returns_state_metadata(self) -> None:
|
|
review_items = [
|
|
PersonaReviewItem(
|
|
persona_id="00000000-0000-0000-0000-000000000302",
|
|
code="P2",
|
|
version=2,
|
|
status="draft",
|
|
display_name="Draft Persona",
|
|
difficulty="moderate",
|
|
theory_target=["humanistic"],
|
|
source_provenance="faculty import",
|
|
is_synthetic=True,
|
|
created_at="2026-01-01T00:00:00",
|
|
approved_at=None,
|
|
),
|
|
PersonaReviewItem(
|
|
persona_id="00000000-0000-0000-0000-000000000303",
|
|
code="P3",
|
|
version=1,
|
|
status="review",
|
|
display_name="Review Persona",
|
|
difficulty="hard",
|
|
theory_target=["cbt"],
|
|
source_provenance="faculty import",
|
|
is_synthetic=True,
|
|
created_at="2026-01-02T00:00:00",
|
|
approved_at=None,
|
|
),
|
|
]
|
|
|
|
with patch.object(
|
|
personas,
|
|
"list_persona_review_queue",
|
|
AsyncMock(return_value=review_items),
|
|
) as review_queue:
|
|
response = await personas.list_persona_reviews(_principal(Role.TEACHER))
|
|
|
|
review_queue.assert_awaited_once_with(role="teacher")
|
|
self.assertEqual([item.code for item in response], ["P2", "P3"])
|
|
self.assertEqual([item.status for item in response], ["draft", "review"])
|
|
self.assertEqual(response[0].version, 2)
|
|
self.assertIsNone(response[0].approved_at)
|
|
|
|
async def test_admin_review_route_uses_admin_db_role(self) -> None:
|
|
with patch.object(
|
|
personas,
|
|
"list_persona_review_queue",
|
|
AsyncMock(return_value=[]),
|
|
) as review_queue:
|
|
response = await personas.list_persona_reviews(_principal(Role.ADMIN))
|
|
|
|
self.assertEqual(response, [])
|
|
review_queue.assert_awaited_once_with(role="admin")
|
|
|
|
async def test_teacher_approves_review_persona_and_audits_decision(self) -> None:
|
|
reviewer_id = "00000000-0000-0000-0000-000000000901"
|
|
persona_id = "00000000-0000-0000-0000-000000000401"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id=persona_id,
|
|
status="review",
|
|
version=2,
|
|
),
|
|
]
|
|
)
|
|
acquire_calls: list[dict[str, Any]] = []
|
|
|
|
def fake_acquire(**kwargs: Any) -> _Acquire:
|
|
acquire_calls.append(kwargs)
|
|
return _Acquire(conn)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", fake_acquire),
|
|
):
|
|
updated = await persona_repository.update_persona_review_status(
|
|
persona_id=persona_id,
|
|
action="approve",
|
|
reviewer_id=reviewer_id,
|
|
role="teacher",
|
|
)
|
|
|
|
self.assertIsNotNone(updated)
|
|
assert updated is not None
|
|
self.assertEqual(updated.status, "approved")
|
|
self.assertEqual(updated.approved_at, "2026-01-03T00:00:00")
|
|
self.assertEqual(acquire_calls, [{"role": "teacher", "user_id": reviewer_id}])
|
|
update_query, update_args = conn.fetchrow_calls[0]
|
|
self.assertIn("UPDATE app.persona_card", update_query)
|
|
self.assertIn("status IN ('draft', 'review')", update_query)
|
|
self.assertEqual(update_args, (persona_id, "approved", reviewer_id))
|
|
audit_query, audit_args = conn.execute_calls[0]
|
|
self.assertIn("INSERT INTO audit.audit_log", audit_query)
|
|
self.assertEqual(audit_args[1], "persona_approve")
|
|
self.assertEqual(audit_args[2], "persona_card")
|
|
self.assertEqual(audit_args[3], persona_id)
|
|
self.assertEqual(audit_args[4]["next_status"], "approved")
|
|
|
|
async def test_reject_review_persona_returns_it_to_draft_and_audits(self) -> None:
|
|
reviewer_id = "00000000-0000-0000-0000-000000000901"
|
|
persona_id = "00000000-0000-0000-0000-000000000402"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P3,
|
|
persona_id=persona_id,
|
|
status="review",
|
|
version=1,
|
|
),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
updated = await persona_repository.update_persona_review_status(
|
|
persona_id=persona_id,
|
|
action="reject",
|
|
reviewer_id=reviewer_id,
|
|
role="admin",
|
|
)
|
|
|
|
self.assertIsNotNone(updated)
|
|
assert updated is not None
|
|
self.assertEqual(updated.status, "draft")
|
|
self.assertIsNone(updated.approved_at)
|
|
_, update_args = conn.fetchrow_calls[0]
|
|
self.assertEqual(update_args, (persona_id, "draft", None))
|
|
_, audit_args = conn.execute_calls[0]
|
|
self.assertEqual(audit_args[1], "persona_reject")
|
|
self.assertEqual(audit_args[4]["next_status"], "draft")
|
|
|
|
async def test_review_update_ignores_already_approved_persona(self) -> None:
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P1,
|
|
persona_id="00000000-0000-0000-0000-000000000403",
|
|
status="approved",
|
|
),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
updated = await persona_repository.update_persona_review_status(
|
|
persona_id="00000000-0000-0000-0000-000000000403",
|
|
action="approve",
|
|
reviewer_id="00000000-0000-0000-0000-000000000901",
|
|
role="teacher",
|
|
)
|
|
|
|
self.assertIsNone(updated)
|
|
self.assertEqual(conn.execute_calls, [])
|
|
|
|
async def test_learner_cannot_call_review_decision_route(self) -> None:
|
|
with patch.object(
|
|
personas,
|
|
"update_persona_review_status",
|
|
AsyncMock(side_effect=AssertionError("learner must not reach review update")),
|
|
) as update_review:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await personas.decide_persona_review(
|
|
"00000000-0000-0000-0000-000000000404",
|
|
personas.PersonaReviewDecisionRequest(action="approve"),
|
|
_principal(Role.LEARNER),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
update_review.assert_not_awaited()
|
|
|
|
async def test_review_decision_route_returns_404_for_non_pending_persona(self) -> None:
|
|
with patch.object(
|
|
personas,
|
|
"update_persona_review_status",
|
|
AsyncMock(return_value=None),
|
|
) as update_review:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await personas.decide_persona_review(
|
|
"00000000-0000-0000-0000-000000000405",
|
|
personas.PersonaReviewDecisionRequest(action="approve"),
|
|
_principal(Role.TEACHER),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 404)
|
|
update_review.assert_awaited_once_with(
|
|
persona_id="00000000-0000-0000-0000-000000000405",
|
|
action="approve",
|
|
reviewer_id="00000000-0000-0000-0000-000000000901",
|
|
role="teacher",
|
|
)
|
|
|
|
async def test_review_decision_route_returns_updated_summary(self) -> None:
|
|
updated_item = PersonaReviewItem(
|
|
persona_id="00000000-0000-0000-0000-000000000406",
|
|
code="P2",
|
|
version=3,
|
|
status="approved",
|
|
display_name="Approved Persona",
|
|
difficulty="moderate",
|
|
theory_target=["humanistic"],
|
|
source_provenance="faculty import",
|
|
is_synthetic=True,
|
|
created_at="2026-01-01T00:00:00",
|
|
approved_at="2026-01-03T00:00:00",
|
|
)
|
|
|
|
with patch.object(
|
|
personas,
|
|
"update_persona_review_status",
|
|
AsyncMock(return_value=updated_item),
|
|
) as update_review:
|
|
response = await personas.decide_persona_review(
|
|
"00000000-0000-0000-0000-000000000406",
|
|
personas.PersonaReviewDecisionRequest(action="approve"),
|
|
_principal(Role.ADMIN),
|
|
)
|
|
|
|
self.assertEqual(response.status, "approved")
|
|
self.assertEqual(response.approved_at, "2026-01-03T00:00:00")
|
|
update_review.assert_awaited_once_with(
|
|
persona_id="00000000-0000-0000-0000-000000000406",
|
|
action="approve",
|
|
reviewer_id="00000000-0000-0000-0000-000000000901",
|
|
role="admin",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|