from __future__ import annotations import asyncio from contextlib import asynccontextmanager from copy import deepcopy from datetime import datetime, timezone from pathlib import Path import unittest from unittest.mock import AsyncMock, patch from uuid import UUID from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from pydantic import ValidationError from . import db from .deps import Principal, Role, get_current_principal from .main import app as main_app from .routes import protocols from .services import live_coach, protocol_registry, rag NOW = datetime(2026, 8, 27, 1, 0, tzinfo=timezone.utc) PROTOCOL_ID = "71000000-0000-4000-8000-000000000001" ADMIN_ID = "71000000-0000-4000-8000-000000000099" def protocol_row(status: str = "draft", **overrides: object) -> dict[str, object]: activated_at = NOW if status in {"active", "retired"} else None retired_at = NOW if status == "retired" else None row: dict[str, object] = { "protocol_id": PROTOCOL_ID, "source_id": f"protocol:{PROTOCOL_ID}", "title": "위기 개입 기본 프로토콜", "source_ref": "https://example.edu/protocols/crisis-v1", "version": 1, "license_class": "B", "external_llm_ok": False, "content": "위험도를 먼저 확인한다.\n\n안전 계획을 함께 수립한다.", "content_hash": "a" * 64, "status": status, "registered_by": ADMIN_ID, "registered_at": NOW, "activated_at": activated_at, "retired_at": retired_at, } row.update(overrides) return row class FakeProtocolConnection: def __init__(self, row: dict[str, object] | None = None) -> None: self.row = row self.rows: list[dict[str, object]] = [row] if row else [] self.fetchrow_calls: list[tuple[str, tuple[object, ...]]] = [] self.fetch_calls: list[tuple[str, tuple[object, ...]]] = [] self.execute_calls: list[tuple[str, tuple[object, ...]]] = [] self.source_registered = False self.documents_active = row is not None and row.get("status") == "active" self.fail_on_retire_status = False self.transaction_entries = 0 self.transaction_commits = 0 self.transaction_rollbacks = 0 self.transaction_lock = asyncio.Lock() class _Transaction: def __init__(self, conn: "FakeProtocolConnection") -> None: self.conn = conn self.snapshot: tuple[ dict[str, object] | None, list[dict[str, object]], bool, bool, ] | None = None async def __aenter__(self) -> "FakeProtocolConnection._Transaction": await self.conn.transaction_lock.acquire() self.conn.transaction_entries += 1 self.snapshot = ( deepcopy(self.conn.row), deepcopy(self.conn.rows), self.conn.source_registered, self.conn.documents_active, ) return self async def __aexit__(self, exc_type, exc, traceback) -> bool: try: if exc_type is None: self.conn.transaction_commits += 1 else: assert self.snapshot is not None ( self.conn.row, self.conn.rows, self.conn.source_registered, self.conn.documents_active, ) = self.snapshot self.conn.transaction_rollbacks += 1 finally: self.conn.transaction_lock.release() return False def transaction(self) -> "FakeProtocolConnection._Transaction": return self._Transaction(self) async def fetchrow(self, query: str, *args: object) -> dict[str, object] | None: self.fetchrow_calls.append((query, args)) if "INSERT INTO kb.protocol_registration" in query: self.row = protocol_row( protocol_id=args[0], source_id=args[1], title=args[2], source_ref=args[3], version=args[4], license_class=args[5], external_llm_ok=args[6], content=args[7], content_hash=args[8], registered_by=args[9], ) return self.row if "SET status = 'active'" in query and self.row is not None: self.row = {**self.row, "status": "active", "activated_at": NOW} return self.row if "SET status = 'retired'" in query and self.row is not None: if self.fail_on_retire_status: raise RuntimeError("retire status write failed") self.row = {**self.row, "status": "retired", "retired_at": NOW} return self.row return self.row async def fetch(self, query: str, *args: object) -> list[dict[str, object]]: self.fetch_calls.append((query, args)) return self.rows async def execute(self, query: str, *args: object) -> str: self.execute_calls.append((query, args)) if "INSERT INTO kb.source" in query: self.source_registered = True if "UPDATE kb.document SET is_active = FALSE" in query: self.documents_active = False return "UPDATE 1" class ProtocolRegistryUnitTests(unittest.IsolatedAsyncioTestCase): async def test_create_stores_normalized_draft_and_server_hash_only(self) -> None: conn = FakeProtocolConnection() record = await protocol_registry.create_protocol( conn, title=" 위기 개입 기본 프로토콜 ", source=" https://example.edu/protocols/crisis-v1 ", version=3, license_class="B", external_llm_ok=False, content="첫 문단\r\n\r\n둘째 문단\r\n", registered_by=ADMIN_ID, ) self.assertEqual(record.status, "draft") self.assertEqual(record.title, "위기 개입 기본 프로토콜") self.assertEqual(record.content, "첫 문단\n\n둘째 문단") self.assertEqual(record.content_hash, protocol_registry.content_hash(record.content)) self.assertTrue(record.source_id.startswith("protocol:")) self.assertEqual(conn.execute_calls, [], "draft creation must not touch kb.source/chunks") async def test_activate_reuses_source_document_chunk_and_preserves_license(self) -> None: conn = FakeProtocolConnection(protocol_row()) indexed = rag.IndexResult( doc_id=31, chunks_indexed=2, skipped_unchanged=False, embedded=True, degraded=False, ) fake_index = AsyncMock(return_value=indexed) with patch.object(protocol_registry.rag, "index_document", fake_index): record, result = await protocol_registry.activate_protocol( conn, protocol_id=PROTOCOL_ID, ) self.assertEqual(record.status, "active") self.assertEqual(result, indexed) source_call = next(call for call in conn.execute_calls if "INSERT INTO kb.source" in call[0]) self.assertEqual(source_call[1][2], "B") self.assertIs(source_call[1][4], False) index_request = fake_index.await_args.args[1] self.assertEqual(index_request.source_id, f"protocol:{PROTOCOL_ID}") self.assertEqual(index_request.content_hash, "a" * 64) self.assertTrue(index_request.chunks) self.assertEqual(index_request.chunks[0]["meta"]["license_class"], "B") self.assertIs(index_request.chunks[0]["meta"]["external_llm_ok"], False) async def test_activation_failure_never_marks_draft_active(self) -> None: conn = FakeProtocolConnection(protocol_row()) fake_index = AsyncMock(side_effect=rag.NotConfigured("vector unavailable")) with ( patch.object(protocol_registry.rag, "index_document", fake_index), self.assertRaises(protocol_registry.ProtocolStoreUnavailable), ): await protocol_registry.activate_protocol(conn, protocol_id=PROTOCOL_ID) self.assertEqual(conn.row["status"], "draft") # type: ignore[index] self.assertFalse( any("SET status = 'active'" in query for query, _ in conn.fetchrow_calls) ) async def test_retire_deactivates_documents_before_terminal_status(self) -> None: conn = FakeProtocolConnection(protocol_row("active")) record = await protocol_registry.retire_protocol(conn, protocol_id=PROTOCOL_ID) self.assertEqual(record.status, "retired") self.assertIn("UPDATE kb.document SET is_active = FALSE", conn.execute_calls[0][0]) self.assertEqual(conn.execute_calls[0][1], (f"protocol:{PROTOCOL_ID}",)) async def test_retired_protocol_cannot_be_activated_again(self) -> None: conn = FakeProtocolConnection(protocol_row("retired")) with self.assertRaises(protocol_registry.ProtocolTransitionConflict): await protocol_registry.activate_protocol(conn, protocol_id=PROTOCOL_ID) async def test_license_c_or_d_cannot_enable_external_llm(self) -> None: for license_class in ("C", "D"): with self.subTest(license_class=license_class): with self.assertRaises(protocol_registry.ProtocolPolicyViolation): protocol_registry.validate_license_policy(license_class, True) class ProtocolReadinessTests(unittest.IsolatedAsyncioTestCase): @staticmethod def _acquire(conn: FakeProtocolConnection): @asynccontextmanager async def fake_acquire(**_kwargs: object): yield conn return fake_acquire async def test_complete_owner_migration_requires_no_runtime_ddl(self) -> None: conn = FakeProtocolConnection( { "protocol_table": True, "source_license_constraint": True, "protocol_license_constraint": True, "protocol_lifecycle_constraint": True, "protocol_chunk_write_policy": True, "protocol_status_index": True, } ) with patch.object(db, "acquire", self._acquire(conn)): await protocol_registry.ensure_protocol_tables() self.assertEqual(conn.execute_calls, []) self.assertEqual(len(conn.fetchrow_calls), 1) readiness_sql = conn.fetchrow_calls[0][0] self.assertIn("to_regclass('kb.protocol_registration')", readiness_sql) self.assertNotIn("CREATE SCHEMA", readiness_sql) self.assertNotIn("CREATE TABLE", readiness_sql) async def test_missing_contract_fails_closed_without_runtime_ddl(self) -> None: conn = FakeProtocolConnection( { "protocol_table": True, "source_license_constraint": True, "protocol_license_constraint": False, "protocol_lifecycle_constraint": True, "protocol_chunk_write_policy": False, "protocol_status_index": False, } ) with ( patch.object(db, "acquire", self._acquire(conn)), self.assertRaises(protocol_registry.ProtocolStoreUnavailable) as raised, ): await protocol_registry.ensure_protocol_tables() self.assertIn("17_improvement_workbook_contracts.sql", str(raised.exception)) self.assertIn("protocol_license_constraint", str(raised.exception)) self.assertIn("protocol_chunk_write_policy", str(raised.exception)) self.assertIn("protocol_status_index", str(raised.exception)) self.assertEqual(conn.execute_calls, []) class ProtocolLifecycleTransactionTests(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.principal = Principal(user_id=ADMIN_ID, role=Role.ADMIN) @staticmethod def _acquire(conn: FakeProtocolConnection): @asynccontextmanager async def fake_acquire(**_kwargs: object): yield conn return fake_acquire async def test_activate_rolls_back_source_when_indexing_fails(self) -> None: conn = FakeProtocolConnection(protocol_row()) fake_index = AsyncMock(side_effect=rag.NotConfigured("vector unavailable")) with ( patch.object(protocols, "acquire", self._acquire(conn)), patch.object(protocol_registry.rag, "index_document", fake_index), self.assertRaises(HTTPException) as raised, ): await protocols.activate_admin_protocol( UUID(PROTOCOL_ID), self.principal, ) self.assertEqual(raised.exception.status_code, 503) self.assertEqual(conn.transaction_entries, 1) self.assertEqual(conn.transaction_commits, 0) self.assertEqual(conn.transaction_rollbacks, 1) self.assertFalse(conn.source_registered) self.assertEqual(conn.row["status"], "draft") # type: ignore[index] async def test_retire_rolls_back_document_deactivation_when_status_write_fails( self, ) -> None: conn = FakeProtocolConnection(protocol_row("active")) conn.fail_on_retire_status = True with ( patch.object(protocols, "acquire", self._acquire(conn)), self.assertRaises(HTTPException) as raised, ): await protocols.retire_admin_protocol( UUID(PROTOCOL_ID), self.principal, ) self.assertEqual(raised.exception.status_code, 503) self.assertEqual(conn.transaction_entries, 1) self.assertEqual(conn.transaction_commits, 0) self.assertEqual(conn.transaction_rollbacks, 1) self.assertTrue(conn.documents_active) self.assertEqual(conn.row["status"], "active") # type: ignore[index] async def test_concurrent_activation_serializes_and_indexes_once(self) -> None: conn = FakeProtocolConnection(protocol_row()) indexed = rag.IndexResult( doc_id=31, chunks_indexed=2, skipped_unchanged=False, embedded=True, degraded=False, ) fake_index = AsyncMock(return_value=indexed) with ( patch.object(protocols, "acquire", self._acquire(conn)), patch.object(protocol_registry.rag, "index_document", fake_index), ): results = await asyncio.gather( protocols.activate_admin_protocol(UUID(PROTOCOL_ID), self.principal), protocols.activate_admin_protocol(UUID(PROTOCOL_ID), self.principal), return_exceptions=True, ) successes = [item for item in results if not isinstance(item, Exception)] conflicts = [ item for item in results if isinstance(item, HTTPException) and item.status_code == 409 ] self.assertEqual(len(successes), 1) self.assertEqual(len(conflicts), 1) self.assertEqual(fake_index.await_count, 1) self.assertEqual(conn.transaction_entries, 2) self.assertEqual(conn.transaction_commits, 1) self.assertEqual(conn.transaction_rollbacks, 1) self.assertEqual(conn.row["status"], "active") # type: ignore[index] class IndexSourcePolicyTests(unittest.IsolatedAsyncioTestCase): async def test_unregistered_source_is_rejected(self) -> None: conn = FakeProtocolConnection(None) with self.assertRaises(rag.IndexPolicyViolation): await rag.validate_index_source(conn, "unknown") # type: ignore[arg-type] async def test_copyright_sensitive_external_source_is_rejected(self) -> None: conn = FakeProtocolConnection( {"license_class": "C", "external_llm_ok": True, "protocol_status": None} ) with self.assertRaises(rag.IndexPolicyViolation): await rag.validate_index_source(conn, "unsafe") # type: ignore[arg-type] async def test_draft_protocol_source_is_rejected(self) -> None: conn = FakeProtocolConnection( {"license_class": "B", "external_llm_ok": False, "protocol_status": "draft"} ) with self.assertRaises(rag.IndexPolicyViolation): await rag.validate_index_source(conn, "protocol:draft") # type: ignore[arg-type] async def test_active_or_legacy_registered_source_is_allowed(self) -> None: for protocol_status in ("active", None): with self.subTest(protocol_status=protocol_status): conn = FakeProtocolConnection( { "license_class": "A", "external_llm_ok": True, "protocol_status": protocol_status, } ) await rag.validate_index_source(conn, "registered") # type: ignore[arg-type] class ProtocolContractTests(unittest.TestCase): def test_schema_and_search_are_fail_closed(self) -> None: self.assertIn("status TEXT NOT NULL DEFAULT 'draft'", protocol_registry.PROTOCOL_SCHEMA_SQL) self.assertIn("ck_protocol_external_license", protocol_registry.PROTOCOL_SCHEMA_SQL) self.assertIn("p_kb_chunk_admin_write", protocol_registry.PROTOCOL_SCHEMA_SQL) self.assertIn("app.current_role_name() = 'admin'", protocol_registry.PROTOCOL_SCHEMA_SQL) self.assertIn("p_kb_chunk_admin_write", protocol_registry.PROTOCOL_READINESS_SQL) self.assertEqual(rag._HYBRID_SQL.count("JOIN kb.document d"), 2) self.assertEqual(rag._HYBRID_SQL.count("LEFT JOIN kb.protocol_registration pr"), 2) self.assertEqual(rag._HYBRID_SQL.count("pr.status = 'active'"), 2) self.assertIn("JOIN kb.source src ON src.source_id = c.source_id", rag._HYBRID_SQL) self.assertIn("src.license_class, src.external_llm_ok", rag._HYBRID_SQL) def test_registered_chunks_are_evaluator_only_and_sensitivity_two(self) -> None: record = protocol_registry._record(protocol_row("active")) chunks = protocol_registry.build_index_chunks(record, limit=32) self.assertGreater(len(chunks), 0) self.assertTrue(all(chunk["visible_to"] == ["evaluator"] for chunk in chunks)) self.assertTrue(all(chunk["sensitivity"] == 2 for chunk in chunks)) self.assertTrue( all(chunk["meta"]["license_class"] == "B" for chunk in chunks) ) self.assertTrue( all(chunk["meta"]["external_llm_ok"] is False for chunk in chunks) ) def test_external_prompt_grounding_allows_only_explicit_a_or_b(self) -> None: grounding = [ live_coach.LiveCoachGrounding( source_id="allowed-a", title="A", license_class="A", external_llm_ok=True, summary="allowed", ), live_coach.LiveCoachGrounding( source_id="allowed-b", title="B", license_class="B", external_llm_ok=True, summary="allowed", ), live_coach.LiveCoachGrounding( source_id="denied-flag", title="A false", license_class="A", external_llm_ok=False, summary="denied", ), live_coach.LiveCoachGrounding( source_id="denied-c", title="C", license_class="C", external_llm_ok=True, summary="denied", ), live_coach.LiveCoachGrounding( source_id="denied-missing", title="missing", summary="denied", ), ] safe = live_coach._external_safe_grounding(grounding) self.assertEqual([item.source_id for item in safe], ["allowed-a", "allowed-b"]) def test_create_contract_rejects_license_bypass(self) -> None: with self.assertRaises(ValidationError): protocols.AdminProtocolCreate( title="민감 프로토콜", source="licensed://restricted", version=1, license="C", external_llm_ok=True, content="저작권 민감 원문", ) def test_owner_migration_closes_existing_db_contracts(self) -> None: migration = ( Path(__file__).resolve().parents[3] / "infra" / "db" / "init" / "17_improvement_workbook_contracts.sql" ).read_text(encoding="utf-8") self.assertIn( "ADD COLUMN IF NOT EXISTS learner_feedback_enabled BOOLEAN NOT NULL DEFAULT TRUE", migration, ) self.assertEqual(migration.count("learner_feedback_enabled"), 2) self.assertIn("CREATE TABLE IF NOT EXISTS kb.protocol_registration", migration) self.assertIn("CREATE INDEX IF NOT EXISTS idx_protocol_registration_status", migration) self.assertLess( migration.index("UPDATE kb.source"), migration.index("VALIDATE CONSTRAINT ck_kb_source_external_license"), ) self.assertLess( migration.index("UPDATE kb.protocol_registration"), migration.index("VALIDATE CONSTRAINT ck_protocol_external_license"), ) def test_openapi_exposes_typed_admin_lifecycle(self) -> None: schema = main_app.openapi() paths = schema["paths"] self.assertIn("/admin/protocols", paths) self.assertIn("/admin/protocols/{protocol_id}/activate", paths) self.assertIn("/admin/protocols/{protocol_id}/retire", paths) create_schema = schema["components"]["schemas"]["AdminProtocolCreate"] self.assertIn("license", create_schema["required"]) self.assertIn("content", create_schema["required"]) response_schema = schema["components"]["schemas"]["AdminProtocolResponse"] self.assertIn("content_hash", response_schema["required"]) self.assertIn("registered_at", response_schema["required"]) def test_learner_is_forbidden_before_protocol_store_access(self) -> None: contract_app = FastAPI() contract_app.include_router(protocols.router) contract_app.dependency_overrides[get_current_principal] = lambda: Principal( user_id="learner-1", role=Role.LEARNER, ) with TestClient(contract_app) as client: response = client.get("/admin/protocols") self.assertEqual(response.status_code, 403) if __name__ == "__main__": unittest.main()