544 lines
22 KiB
Python
544 lines
22 KiB
Python
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()
|