752 lines
29 KiB
Python
752 lines
29 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 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(),
|
|
)
|
|
|
|
|
|
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.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 "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]),
|
|
"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]),
|
|
"created_by": args[18],
|
|
"approved_by": None,
|
|
"created_at": "2026-01-04T00:00:00",
|
|
"approved_at": None,
|
|
}
|
|
self.rows.append(row)
|
|
return row
|
|
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 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),
|
|
source_provenance=card.source_provenance,
|
|
is_synthetic=card.is_synthetic,
|
|
submit_for_review=submit_for_review,
|
|
)
|
|
|
|
|
|
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_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_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_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_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()
|