1228 lines
51 KiB
Python
1228 lines
51 KiB
Python
"""Regression tests for persona approval and faculty review boundaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from . import persona_generation_contract, persona_repository, session_persistence
|
|
from .deps import Principal, Role
|
|
from .engine_client import GenerateResponse
|
|
from .persona_repository import PersonaDraftRecord, 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(),
|
|
consent_at=1.0 if role == Role.LEARNER else None,
|
|
profile_completed_at=1.0 if role == Role.LEARNER else None,
|
|
)
|
|
|
|
|
|
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),
|
|
"triggers": dict(card.triggers),
|
|
"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.fetchval_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:
|
|
if "archived AS" in query:
|
|
persona_id = str(args[0])
|
|
target = next(
|
|
(row for row in self.rows if row["persona_id"] == persona_id and row["status"] != "archived"),
|
|
None,
|
|
)
|
|
if target is None:
|
|
return None
|
|
code = str(target["code"]).upper()
|
|
returned: dict[str, Any] | None = None
|
|
for row in self.rows:
|
|
if str(row["code"]).upper() == code and row["status"] != "archived":
|
|
row["status"] = "archived"
|
|
row["approved_by"] = None
|
|
row["approved_at"] = None
|
|
if row["persona_id"] == persona_id:
|
|
returned = row
|
|
return returned
|
|
if "display_name = $4" in query:
|
|
persona_id = str(args[0])
|
|
next_status = str(args[2])
|
|
for row in self.rows:
|
|
if row["persona_id"] != persona_id or row["status"] not in {"draft", "review"}:
|
|
continue
|
|
row.update(
|
|
{
|
|
"code": str(args[1]).upper(),
|
|
"status": next_status,
|
|
"display_name": args[3],
|
|
"difficulty": args[4],
|
|
"theory_target": list(args[5]),
|
|
"demographics": dict(args[6]),
|
|
"presenting": dict(args[7]),
|
|
"history": dict(args[8]),
|
|
"big5": dict(args[9]),
|
|
"resistance": dict(args[10]),
|
|
"speech_style": dict(args[11]),
|
|
"affect_baseline": dict(args[12]),
|
|
"ccd": dict(args[13]),
|
|
"dsm5_dimensional": dict(args[14]),
|
|
"source_provenance": args[15],
|
|
"is_synthetic": bool(args[16]),
|
|
"triggers": dict(args[17]),
|
|
"approved_by": None,
|
|
"approved_at": None,
|
|
}
|
|
)
|
|
return row
|
|
return None
|
|
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
|
|
if "INSERT INTO app.persona_card" in query:
|
|
row = {
|
|
"persona_id": str(args[0]),
|
|
"code": str(args[1]).upper(),
|
|
"version": int(args[2]),
|
|
"status": str(args[3]),
|
|
"display_name": args[4],
|
|
"difficulty": args[5],
|
|
"theory_target": list(args[6]),
|
|
"demographics": dict(args[7]),
|
|
"presenting": dict(args[8]),
|
|
"history": dict(args[9]),
|
|
"big5": dict(args[10]),
|
|
"resistance": dict(args[11]),
|
|
"speech_style": dict(args[12]),
|
|
"affect_baseline": dict(args[13]),
|
|
"ccd": dict(args[14]),
|
|
"dsm5_dimensional": dict(args[15]),
|
|
"source_provenance": args[16],
|
|
"is_synthetic": bool(args[17]),
|
|
"triggers": dict(args[18]),
|
|
"created_by": args[19],
|
|
"approved_by": None,
|
|
"created_at": "2026-01-04T00:00:00",
|
|
"approved_at": None,
|
|
}
|
|
self.rows.append(row)
|
|
return row
|
|
if "upper(code) = upper($1)" in query and "status IN ('draft', 'review')" in query:
|
|
code = str(args[0]).upper()
|
|
matches = [
|
|
row
|
|
for row in self.rows
|
|
if str(row["code"]).upper() == code and row["status"] in {"draft", "review"}
|
|
]
|
|
matches.sort(key=lambda row: int(row["version"]), reverse=True)
|
|
return matches[0] if matches else None
|
|
if "WHERE persona_id = $1::uuid" in query:
|
|
persona_id = str(args[0])
|
|
for row in self.rows:
|
|
if row["persona_id"] == persona_id and "status = 'approved'" in query:
|
|
if row["status"] == "approved":
|
|
return row
|
|
return None
|
|
if row["persona_id"] == persona_id and row["status"] in {"draft", "review"}:
|
|
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 fetchval(self, query: str, *args: Any) -> int:
|
|
self.fetchval_calls.append((query, args))
|
|
code = str(args[0]).upper()
|
|
versions = [int(row["version"]) for row in self.rows if str(row["code"]).upper() == code]
|
|
return (max(versions) if versions else 0) + 1
|
|
|
|
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)
|
|
|
|
|
|
def _draft_payload(
|
|
card: persona_service.PersonaCard,
|
|
*,
|
|
submit_for_review: bool = False,
|
|
) -> personas.PersonaDraftPayload:
|
|
return personas.PersonaDraftPayload(
|
|
code=card.code,
|
|
display_name=card.display_name,
|
|
difficulty=card.difficulty, # type: ignore[arg-type]
|
|
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),
|
|
triggers=dict(card.triggers),
|
|
source_provenance=card.source_provenance,
|
|
is_synthetic=card.is_synthetic,
|
|
submit_for_review=submit_for_review,
|
|
)
|
|
|
|
|
|
class PersonaApprovalBoundaryTest(unittest.IsolatedAsyncioTestCase):
|
|
def test_persona_seed_runner_reports_repo_manifest_without_db(self) -> None:
|
|
script_path = Path(__file__).resolve().parents[3] / "scripts" / "materialize-persona-seeds.py"
|
|
spec = importlib.util.spec_from_file_location("materialize_persona_seeds", script_path)
|
|
self.assertIsNotNone(spec)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
rows = module._manifest_rows()
|
|
|
|
self.assertEqual([row["code"] for row in rows], ["P1", "P2", "P3", "P4", "P5", "P6", "P7"])
|
|
self.assertEqual({row["version"] for row in rows}, {persona_repository.SEED_VERSION})
|
|
self.assertEqual(rows[0]["persona_id"], str(persona_repository.seed_persona_id("P1")))
|
|
self.assertIn("source_provenance", rows[0])
|
|
|
|
async def test_persona_seed_runner_apply_initializes_and_closes_pool(self) -> None:
|
|
script_path = Path(__file__).resolve().parents[3] / "scripts" / "materialize-persona-seeds.py"
|
|
spec = importlib.util.spec_from_file_location("materialize_persona_seeds_apply", script_path)
|
|
self.assertIsNotNone(spec)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
init_pool = AsyncMock()
|
|
materialize = AsyncMock(return_value=7)
|
|
close_pool = AsyncMock()
|
|
with (
|
|
patch.object(module, "init_pool", init_pool),
|
|
patch.object(module, "materialize_seed_personas", materialize),
|
|
patch.object(module, "close_pool", close_pool),
|
|
):
|
|
exit_code = await module._main_async(module.parse_args(["--apply"]))
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
init_pool.assert_awaited_once()
|
|
materialize.assert_awaited_once()
|
|
close_pool.assert_awaited_once()
|
|
|
|
async def test_repository_loads_repo_persona_files_for_p4_to_p7(self) -> None:
|
|
cards = persona_repository.load_file_personas()
|
|
|
|
self.assertEqual([card.code for card in cards], ["P4", "P5", "P6", "P7"])
|
|
self.assertEqual(cards[0].display_name, "하늘(가명) · 고2 · 학업/시험 불안")
|
|
self.assertEqual(cards[0].difficulty, "easy")
|
|
self.assertIn("cbt", cards[0].theory_target)
|
|
|
|
async def test_seed_fallback_includes_repo_persona_files(self) -> None:
|
|
entry = persona_repository.seed_fallback_persona("p4")
|
|
|
|
self.assertIsNotNone(entry)
|
|
assert entry is not None
|
|
self.assertEqual(entry.card.code, "P4")
|
|
self.assertEqual(entry.source, "seed_fallback")
|
|
self.assertTrue(entry.degraded)
|
|
|
|
async def test_materialize_seed_personas_upserts_code_and_repo_file_cards(self) -> None:
|
|
conn = _PersonaCardConn([])
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
count = await persona_repository.materialize_seed_personas()
|
|
|
|
self.assertEqual(count, 7)
|
|
persona_card_calls = [
|
|
(query, args)
|
|
for query, args in conn.execute_calls
|
|
if "INSERT INTO app.persona_card" in query
|
|
]
|
|
voice_map_calls = [
|
|
(query, args)
|
|
for query, args in conn.execute_calls
|
|
if "INSERT INTO app.persona_voice_map" in query
|
|
]
|
|
inserted_codes = [args[1] for _, args in persona_card_calls]
|
|
self.assertEqual(inserted_codes, ["P1", "P2", "P3", "P4", "P5", "P6", "P7"])
|
|
self.assertEqual(len(voice_map_calls), 7)
|
|
self.assertEqual(voice_map_calls[0][1][2], "coral")
|
|
self.assertEqual(voice_map_calls[0][1][3]["preset"], "soft-young-fem")
|
|
|
|
async def test_materialize_seed_personas_does_not_overwrite_existing_rows(self) -> None:
|
|
conn = _PersonaCardConn([])
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
await persona_repository.materialize_seed_personas()
|
|
|
|
query, _ = conn.execute_calls[0]
|
|
self.assertIn("ON CONFLICT (code, version) DO NOTHING", query)
|
|
self.assertNotIn("DO UPDATE SET", query)
|
|
|
|
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()
|
|
|
|
async def test_session_joined_card_hydrates_triggers_for_runtime_prompt(self) -> None:
|
|
row = _card_row(
|
|
persona_service.P2,
|
|
persona_id="00000000-0000-0000-0000-000000000107",
|
|
status="approved",
|
|
)
|
|
row["triggers"] = {
|
|
"sore_spots": ["무능하다는 낙인"],
|
|
"forbidden": ["비웃음"],
|
|
"reaction": "말수가 급격히 줄고 방어한다.",
|
|
}
|
|
joined_row = {f"card_{key}": value for key, value in row.items()}
|
|
|
|
card = session_persistence._card_from_joined_session_row(joined_row)
|
|
|
|
self.assertIsNotNone(card)
|
|
assert card is not None
|
|
self.assertEqual(card.triggers["sore_spots"], ["무능하다는 낙인"])
|
|
self.assertEqual(card.triggers["forbidden"], ["비웃음"])
|
|
|
|
|
|
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_teacher_creates_persona_draft_version_and_audits(self) -> None:
|
|
author_id = "00000000-0000-0000-0000-000000000901"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id="00000000-0000-0000-0000-000000000501",
|
|
status="approved",
|
|
version=1,
|
|
),
|
|
]
|
|
)
|
|
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),
|
|
):
|
|
created = await persona_repository.create_persona_draft(
|
|
card=persona_service.P2,
|
|
author_id=author_id,
|
|
role="teacher",
|
|
submit_for_review=True,
|
|
)
|
|
|
|
self.assertEqual(created.code, "P2")
|
|
self.assertEqual(created.version, 2)
|
|
self.assertEqual(created.status, "review")
|
|
self.assertEqual(acquire_calls, [{"role": "teacher", "user_id": author_id}])
|
|
self.assertEqual(conn.fetchval_calls[0][1], ("P2",))
|
|
audit_query, audit_args = conn.execute_calls[0]
|
|
self.assertIn("INSERT INTO audit.audit_log", audit_query)
|
|
self.assertEqual(audit_args[1], "persona_draft_create")
|
|
self.assertEqual(audit_args[4]["next_status"], "review")
|
|
|
|
async def test_teacher_clones_approved_persona_into_new_draft_version(self) -> None:
|
|
author_id = "00000000-0000-0000-0000-000000000901"
|
|
persona_id = "00000000-0000-0000-0000-000000000501"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P2,
|
|
persona_id=persona_id,
|
|
status="approved",
|
|
version=1,
|
|
),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
record = await persona_repository.create_persona_revision_from_existing(
|
|
persona_id=persona_id,
|
|
author_id=author_id,
|
|
role="teacher",
|
|
)
|
|
|
|
self.assertIsNotNone(record)
|
|
assert record is not None
|
|
self.assertEqual(record.review.persona_id, persona_id)
|
|
self.assertEqual(record.review.code, "P2")
|
|
self.assertEqual(record.review.version, 2)
|
|
self.assertEqual(record.review.status, "draft")
|
|
self.assertEqual(record.card.display_name, persona_service.P2.display_name)
|
|
self.assertEqual(conn.fetchval_calls[0][1], ("P2",))
|
|
_, audit_args = conn.execute_calls[0]
|
|
self.assertEqual(audit_args[1], "persona_revision_create")
|
|
self.assertEqual(audit_args[4]["version"], 2)
|
|
|
|
async def test_clone_existing_approved_persona_reuses_open_draft(self) -> None:
|
|
author_id = "00000000-0000-0000-0000-000000000901"
|
|
persona_id = "00000000-0000-0000-0000-000000000501"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(persona_service.P2, persona_id=persona_id, status="approved", version=1),
|
|
_card_row(persona_service.P2, persona_id=persona_id, status="draft", version=2),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
record = await persona_repository.create_persona_revision_from_existing(
|
|
persona_id=persona_id,
|
|
author_id=author_id,
|
|
role="admin",
|
|
)
|
|
|
|
self.assertIsNotNone(record)
|
|
assert record is not None
|
|
self.assertEqual(record.review.version, 2)
|
|
self.assertEqual(record.review.status, "draft")
|
|
self.assertEqual(conn.fetchval_calls, [])
|
|
self.assertEqual(conn.execute_calls, [])
|
|
|
|
async def test_teacher_updates_persona_draft_and_submits_review(self) -> None:
|
|
author_id = "00000000-0000-0000-0000-000000000901"
|
|
persona_id = "00000000-0000-0000-0000-000000000502"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(
|
|
persona_service.P3,
|
|
persona_id=persona_id,
|
|
status="draft",
|
|
version=4,
|
|
),
|
|
]
|
|
)
|
|
edited_card = persona_service.P3
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
updated = await persona_repository.update_persona_draft(
|
|
persona_id=persona_id,
|
|
card=edited_card,
|
|
author_id=author_id,
|
|
role="admin",
|
|
submit_for_review=True,
|
|
)
|
|
|
|
self.assertIsNotNone(updated)
|
|
assert updated is not None
|
|
self.assertEqual(updated.status, "review")
|
|
self.assertEqual(updated.version, 4)
|
|
update_query, update_args = conn.fetchrow_calls[0]
|
|
self.assertIn("display_name = $4", update_query)
|
|
self.assertEqual(update_args[0], persona_id)
|
|
self.assertEqual(update_args[2], "review")
|
|
_, audit_args = conn.execute_calls[0]
|
|
self.assertEqual(audit_args[1], "persona_draft_update")
|
|
|
|
async def test_teacher_reads_persona_draft_detail_route(self) -> None:
|
|
record = PersonaDraftRecord(
|
|
review=PersonaReviewItem(
|
|
persona_id="00000000-0000-0000-0000-000000000503",
|
|
code="P3",
|
|
version=2,
|
|
status="draft",
|
|
display_name=persona_service.P3.display_name,
|
|
difficulty=persona_service.P3.difficulty,
|
|
theory_target=list(persona_service.P3.theory_target),
|
|
source_provenance=persona_service.P3.source_provenance,
|
|
is_synthetic=persona_service.P3.is_synthetic,
|
|
created_at="2026-01-04T00:00:00",
|
|
approved_at=None,
|
|
),
|
|
card=persona_service.P3,
|
|
)
|
|
|
|
with patch.object(
|
|
personas,
|
|
"get_persona_draft_record",
|
|
AsyncMock(return_value=record),
|
|
) as get_draft:
|
|
response = await personas.get_persona_draft_route(
|
|
"00000000-0000-0000-0000-000000000503",
|
|
_principal(Role.TEACHER),
|
|
)
|
|
|
|
self.assertEqual(response.code, "P3")
|
|
self.assertEqual(response.presenting, persona_service.P3.presenting)
|
|
get_draft.assert_awaited_once_with(
|
|
persona_id="00000000-0000-0000-0000-000000000503",
|
|
role="teacher",
|
|
)
|
|
|
|
async def test_teacher_create_draft_route_calls_repository(self) -> None:
|
|
created = PersonaReviewItem(
|
|
persona_id="00000000-0000-0000-0000-000000000504",
|
|
code="P2",
|
|
version=2,
|
|
status="review",
|
|
display_name=persona_service.P2.display_name,
|
|
difficulty=persona_service.P2.difficulty,
|
|
theory_target=list(persona_service.P2.theory_target),
|
|
source_provenance=persona_service.P2.source_provenance,
|
|
is_synthetic=persona_service.P2.is_synthetic,
|
|
created_at="2026-01-04T00:00:00",
|
|
approved_at=None,
|
|
)
|
|
|
|
with patch.object(
|
|
personas,
|
|
"create_persona_draft",
|
|
AsyncMock(return_value=created),
|
|
) as create_draft:
|
|
response = await personas.create_persona_draft_route(
|
|
_draft_payload(persona_service.P2, submit_for_review=True),
|
|
_principal(Role.ADMIN),
|
|
)
|
|
|
|
self.assertEqual(response.status, "review")
|
|
args = create_draft.await_args.kwargs
|
|
self.assertEqual(args["role"], "admin")
|
|
self.assertEqual(args["author_id"], "00000000-0000-0000-0000-000000000901")
|
|
self.assertTrue(args["submit_for_review"])
|
|
self.assertEqual(args["card"].code, "P2")
|
|
|
|
async def test_teacher_generates_persona_draft_from_masked_source_notes(self) -> None:
|
|
captured: list[Any] = []
|
|
evidence = [
|
|
personas.PersonaGenerationEvidence(
|
|
chunk_id=44,
|
|
source_id="persona_authoring_test",
|
|
score=0.88,
|
|
kb_kind="supervisor_pattern",
|
|
heading_path="상담 기록",
|
|
excerpt="내담자 연락처 [PHONE]. 최근 이별 뒤 관계 불안을 호소함.",
|
|
)
|
|
]
|
|
source_ref = personas.PersonaSourceDocumentResponse(
|
|
source_id="persona_authoring_test",
|
|
doc_id=9,
|
|
doc_uri="persona-authoring/test/inline.txt",
|
|
title="상담 기록",
|
|
source_kind="client_record",
|
|
kb_kind="diagnostic",
|
|
license_class="B",
|
|
external_llm_ok=True,
|
|
content_hash="hash",
|
|
chunk_count=1,
|
|
chunks_indexed=1,
|
|
embedded=True,
|
|
pii_entities_masked=["PHONE"],
|
|
)
|
|
|
|
async def fake_generate(req: Any) -> GenerateResponse:
|
|
captured.append(req)
|
|
return GenerateResponse(
|
|
text="",
|
|
model="test-model",
|
|
provider="test",
|
|
structured={
|
|
"draft": {
|
|
"code": "P8",
|
|
"display_name": "자료 기반 내담자",
|
|
"difficulty": "moderate",
|
|
"theory_target": ["humanistic"],
|
|
"demographics": {"age_band": "20s"},
|
|
"presenting": {"complaint": "관계 갈등"},
|
|
"history": {"precipitant": "최근 이별"},
|
|
"big5": {"O": 0.5, "C": 0.5, "E": 0.4, "A": 0.5, "N": 0.7},
|
|
"resistance": {
|
|
"base_resistance": 0.45,
|
|
"unlock_rate": 0.12,
|
|
"decay_floor": 0.05,
|
|
},
|
|
"speech_style": {"register": "polite"},
|
|
"affect_baseline": {
|
|
"negative_affect": 0.55,
|
|
"hopelessness": 0.2,
|
|
"anxiety": 0.5,
|
|
"suicide_ideation_stage": 1,
|
|
},
|
|
"ccd": {"core_belief": "나는 버려질 수 있다"},
|
|
"dsm5_dimensional": {"anxiety": 0.5},
|
|
"triggers": {
|
|
"sore_spots": ["버림받음"],
|
|
"forbidden": ["그냥 잊으라는 조언"],
|
|
"reaction": "말수가 줄어든다.",
|
|
},
|
|
"source_provenance": "masked client_record",
|
|
"is_synthetic": True,
|
|
},
|
|
"source_summary": "관계 상실 후 불안이 높아진 사례",
|
|
"warnings": ["임상팀 검수 필요"],
|
|
},
|
|
)
|
|
|
|
with (
|
|
patch.object(personas, "_register_persona_source_document", AsyncMock(return_value=source_ref)),
|
|
patch.object(personas, "_retrieve_persona_generation_evidence", AsyncMock(return_value=evidence)),
|
|
patch.object(personas.engine_client, "generate", fake_generate),
|
|
):
|
|
response = await personas.generate_persona_draft_route(
|
|
personas.PersonaDraftGenerateRequest(
|
|
source_text="내담자 휴대폰 010-1234-5678. 최근 이별 뒤 관계 불안을 호소함.",
|
|
source_kind="client_record",
|
|
code_hint="P8",
|
|
),
|
|
_principal(Role.TEACHER),
|
|
)
|
|
|
|
self.assertEqual(response.draft.code, "P8")
|
|
self.assertEqual(response.draft.triggers["sore_spots"], ["버림받음"])
|
|
self.assertEqual(response.source_summary, "관계 상실 후 불안이 높아진 사례")
|
|
self.assertIn("PHONE", response.pii_entities_masked)
|
|
self.assertEqual(response.source_references[0].source_id, "persona_authoring_test")
|
|
self.assertEqual(response.evidence_chunks[0].chunk_id, 44)
|
|
self.assertIn("RAG sources=persona_authoring_test", response.draft.source_provenance)
|
|
self.assertIn("chunks=44", response.draft.source_provenance)
|
|
self.assertIn("prompt=persona-draft-rag@2026-06-28.1#", response.draft.source_provenance)
|
|
self.assertEqual(len(captured), 1)
|
|
self.assertEqual(captured[0].metadata["prompt_bundle"]["id"], "persona-draft-rag")
|
|
self.assertEqual(captured[0].metadata["prompt_bundle"]["version"], "2026-06-28.1")
|
|
self.assertRegex(captured[0].metadata["prompt_bundle"]["hash"], r"^[0-9a-f]{12}$")
|
|
self.assertEqual(captured[0].structured_schema, persona_generation_contract.persona_generation_schema())
|
|
sent_text = captured[0].messages[-1].content
|
|
self.assertNotIn("010-1234-5678", sent_text)
|
|
self.assertIn("[PHONE]", sent_text)
|
|
self.assertIn("RAG 근거 청크", sent_text)
|
|
|
|
async def test_teacher_registers_persona_source_as_evaluator_only_kb_document(self) -> None:
|
|
conn = _PersonaCardConn([])
|
|
captured_index: list[Any] = []
|
|
|
|
async def fake_index_document(_conn: Any, req: Any) -> Any:
|
|
captured_index.append(req)
|
|
return personas.rag.IndexResult(
|
|
doc_id=42,
|
|
chunks_indexed=len(req.chunks),
|
|
skipped_unchanged=False,
|
|
embedded=False,
|
|
degraded=True,
|
|
)
|
|
|
|
with (
|
|
patch.object(personas, "acquire", lambda **_: _Acquire(conn)),
|
|
patch.object(personas.rag, "index_document", fake_index_document),
|
|
):
|
|
response = await personas.create_persona_source_route(
|
|
personas.PersonaSourceDocumentRequest(
|
|
filename="case-note.txt",
|
|
source_kind="client_record",
|
|
text=(
|
|
"내담자 전화번호 010-1234-5678. 관계 단절 이후 불안을 호소함.\n\n"
|
|
"상담 장면에서는 조언을 들으면 침묵이 늘어남."
|
|
),
|
|
),
|
|
_principal(Role.TEACHER),
|
|
)
|
|
|
|
self.assertEqual(response.doc_id, 42)
|
|
self.assertEqual(response.source_kind, "client_record")
|
|
self.assertEqual(response.kb_kind, "diagnostic")
|
|
self.assertEqual(response.license_class, "B")
|
|
self.assertTrue(response.external_llm_ok)
|
|
self.assertIn("PHONE", response.pii_entities_masked)
|
|
self.assertEqual(len(captured_index), 1)
|
|
index_req = captured_index[0]
|
|
self.assertTrue(index_req.source_id.startswith("persona_authoring_"))
|
|
self.assertNotIn("010-1234-5678", index_req.chunks[0]["chunk_text"])
|
|
self.assertIn("[PHONE]", index_req.chunks[0]["chunk_text"])
|
|
self.assertEqual(index_req.chunks[0]["visible_to"], ["evaluator"])
|
|
self.assertEqual(index_req.chunks[0]["sensitivity"], 2)
|
|
self.assertEqual(index_req.chunks[0]["meta"]["source_kind"], "client_record")
|
|
self.assertTrue(index_req.chunks[0]["meta"]["raw_source_not_indexed"])
|
|
self.assertEqual(index_req.chunks[0]["meta"]["raw_source_storage"], "hash_only")
|
|
self.assertTrue(index_req.chunks[0]["meta"]["raw_source_id"].endswith("_raw"))
|
|
self.assertRegex(index_req.chunks[0]["meta"]["raw_source_content_hash"], r"^[0-9a-f]{64}$")
|
|
self.assertNotIn("raw_source_not_stored", index_req.chunks[0]["meta"])
|
|
source_query, source_args = conn.execute_calls[0]
|
|
self.assertIn("INSERT INTO kb.source", source_query)
|
|
self.assertEqual(source_args[2], "diagnostic")
|
|
raw_query, raw_args = conn.execute_calls[1]
|
|
self.assertIn("INSERT INTO kb.raw_source_artifact", raw_query)
|
|
self.assertEqual(raw_args[0], index_req.chunks[0]["meta"]["raw_source_id"])
|
|
self.assertEqual(raw_args[1], index_req.source_id)
|
|
self.assertEqual(raw_args[2], "00000000-0000-0000-0000-000000000901")
|
|
self.assertEqual(raw_args[5], "B")
|
|
raw_summary = json.loads(raw_args[6])
|
|
self.assertEqual(raw_summary["derived_source_id"], index_req.source_id)
|
|
self.assertTrue(raw_summary["raw_text_not_indexed"])
|
|
self.assertEqual(raw_summary["storage"], "hash_only")
|
|
|
|
async def test_learner_cannot_create_persona_draft_route(self) -> None:
|
|
with patch.object(
|
|
personas,
|
|
"create_persona_draft",
|
|
AsyncMock(side_effect=AssertionError("learner must not reach draft repository")),
|
|
) as create_draft:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await personas.create_persona_draft_route(
|
|
_draft_payload(persona_service.P1),
|
|
_principal(Role.LEARNER),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
create_draft.assert_not_awaited()
|
|
|
|
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_archive_persona_family_archives_all_versions_and_audits(self) -> None:
|
|
archiver_id = "00000000-0000-0000-0000-000000000901"
|
|
persona_id = "00000000-0000-0000-0000-000000000501"
|
|
conn = _PersonaCardConn(
|
|
[
|
|
_card_row(persona_service.P2, persona_id=persona_id, status="approved", version=1),
|
|
_card_row(persona_service.P2, persona_id=persona_id, status="approved", version=2),
|
|
_card_row(
|
|
persona_service.P3,
|
|
persona_id="00000000-0000-0000-0000-000000000503",
|
|
status="approved",
|
|
version=1,
|
|
),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch.object(persona_repository, "get_pool", return_value=object()),
|
|
patch.object(persona_repository, "acquire", lambda **_: _Acquire(conn)),
|
|
):
|
|
archived = await persona_repository.archive_persona_family(
|
|
persona_id=persona_id,
|
|
archiver_id=archiver_id,
|
|
role="teacher",
|
|
)
|
|
catalog = await persona_repository.list_approved_personas()
|
|
|
|
self.assertIsNotNone(archived)
|
|
assert archived is not None
|
|
self.assertEqual(archived.status, "archived")
|
|
self.assertTrue(
|
|
all(row["status"] == "archived" for row in conn.rows if str(row["code"]).upper() == "P2")
|
|
)
|
|
self.assertEqual([entry.card.code for entry in catalog], ["P3"])
|
|
_, audit_args = conn.execute_calls[0]
|
|
self.assertEqual(audit_args[1], "persona_archive")
|
|
self.assertEqual(audit_args[4]["scope"], "code_family")
|
|
|
|
async def test_archive_route_blocks_learner_before_repository_access(self) -> None:
|
|
with patch.object(
|
|
personas,
|
|
"archive_persona_family",
|
|
AsyncMock(side_effect=AssertionError("learner must not archive personas")),
|
|
) as archive:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await personas.archive_persona_route(
|
|
"00000000-0000-0000-0000-000000000501",
|
|
_principal(Role.LEARNER),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
archive.assert_not_awaited()
|
|
|
|
async def test_teacher_revision_route_returns_editable_draft_detail(self) -> None:
|
|
record = PersonaDraftRecord(
|
|
review=PersonaReviewItem(
|
|
persona_id="00000000-0000-0000-0000-000000000503",
|
|
code="P3",
|
|
version=2,
|
|
status="draft",
|
|
display_name=persona_service.P3.display_name,
|
|
difficulty=persona_service.P3.difficulty,
|
|
theory_target=list(persona_service.P3.theory_target),
|
|
source_provenance=persona_service.P3.source_provenance,
|
|
is_synthetic=persona_service.P3.is_synthetic,
|
|
created_at="2026-01-04T00:00:00",
|
|
approved_at=None,
|
|
),
|
|
card=persona_service.P3,
|
|
)
|
|
|
|
with patch.object(
|
|
personas,
|
|
"create_persona_revision_from_existing",
|
|
AsyncMock(return_value=record),
|
|
) as create_revision:
|
|
response = await personas.create_persona_revision_route(
|
|
"00000000-0000-0000-0000-000000000503",
|
|
personas.PersonaRevisionRequest(),
|
|
_principal(Role.TEACHER),
|
|
)
|
|
|
|
self.assertEqual(response.status, "draft")
|
|
self.assertEqual(response.presenting, persona_service.P3.presenting)
|
|
create_revision.assert_awaited_once_with(
|
|
persona_id="00000000-0000-0000-0000-000000000503",
|
|
author_id="00000000-0000-0000-0000-000000000901",
|
|
role="teacher",
|
|
submit_for_review=False,
|
|
)
|
|
|
|
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()
|