from __future__ import annotations from types import SimpleNamespace import unittest from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from govoplan_core.admin.models import SystemSettings from govoplan_core.auth import ApiPrincipal, get_api_principal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.core.configuration_control import ( configuration_control_snapshot, create_configuration_change_request, ) from govoplan_core.core.configuration_safety import ( classify_configuration_field, plan_configuration_change, ) from govoplan_core.db.session import get_session from govoplan_policy.backend.api.v1.routes import router from govoplan_policy.backend.db.models import PolicyOverride class CampaignArchiveEncryptionApiTests(unittest.TestCase): """Exercise the real HTTP route, safety catalog, persistence, and history.""" def setUp(self) -> None: self.engine = create_engine( "sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) self.addCleanup(self.engine.dispose) for table in (SystemSettings.__table__, PolicyOverride.__table__, ChangeSequenceEntry.__table__): table.create(self.engine) self.principal = self._principal("admin:policies:read", "admin:policies:write", "system:settings:write") app = FastAPI() app.include_router(router, prefix="/api/v1") def session_dependency(): with Session(self.engine) as session: yield session app.dependency_overrides[get_session] = session_dependency app.dependency_overrides[get_api_principal] = lambda: self.principal self.client = TestClient(app) self.addCleanup(self.client.close) @staticmethod def _principal(*scopes: str) -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id="admin-account", membership_id="admin-user", tenant_id="tenant-1", scopes=frozenset(scopes), ), account=SimpleNamespace(id="admin-account"), user=SimpleNamespace(id="admin-user"), ) def test_system_legacy_opt_in_passes_real_catalog_and_retains_history(self) -> None: path = "/api/v1/admin/campaign-archive-encryption/policies/system" field = classify_configuration_field("campaign_archive_encryption_policy") self.assertIsNotNone(field) self.assertEqual("policy", field.owner_module) self.assertTrue(field.validation_required) self.assertTrue(field.rollback_history_required) self.assertEqual({}, self.client.get(path).json()["policy"]) policy = { "allowed_password_encryption_methods": ["aes", "zip_standard"], "allowed_password_delivery_channels": ["phone", "letter"], } response = self.client.put(path, json={"policy": policy}) self.assertEqual(200, response.status_code, response.text) self.assertEqual(policy["allowed_password_encryption_methods"], response.json()["effective_policy"]["allowed_password_encryption_methods"]) loaded = self.client.get(path) self.assertEqual(200, loaded.status_code) self.assertEqual(policy, loaded.json()["policy"]) with Session(self.engine) as session: history = configuration_control_snapshot(session)["history"] self.assertEqual(1, len(history)) self.assertEqual("campaign_archive_encryption_policy", history[0]["key"]) self.assertEqual("campaign_archive_encryption_policy.updated", history[0]["audit_event"]) self.assertEqual({}, history[0]["before"]) self.assertEqual(policy, history[0]["after"]) self.assertTrue(history[0]["plan"]["allowed"]) self.assertEqual([], history[0]["plan"]["blockers"]) audit_changes = session.query(ChangeSequenceEntry).filter( ChangeSequenceEntry.module_id == "audit" ).all() self.assertEqual(1, len(audit_changes)) self.assertEqual("campaign_archive_encryption_policy.updated", audit_changes[0].payload["action"]) narrowed_policy = {"allowed_password_encryption_methods": ["aes"]} narrowed = self.client.put(path, json={"policy": narrowed_policy}) self.assertEqual(200, narrowed.status_code, narrowed.text) with Session(self.engine) as session: history = configuration_control_snapshot(session)["history"] self.assertEqual(2, len(history)) self.assertEqual(policy, history[0]["before"]) self.assertEqual(policy, history[0]["rollback_value"]) self.assertEqual(narrowed_policy, history[0]["after"]) def test_read_only_actor_cannot_change_system_policy(self) -> None: self.principal = self._principal("admin:policies:read") policy = {"allowed_password_encryption_methods": ["aes", "zip_standard"]} response = self.client.put( "/api/v1/admin/campaign-archive-encryption/policies/system", json={"policy": policy} ) self.assertEqual(403, response.status_code) plan = plan_configuration_change("campaign_archive_encryption_policy", actor_scopes=tuple(self.principal.scopes), value=policy) self.assertFalse(plan.allowed) self.assertEqual(("system:settings:write", "admin:policies:write"), plan.missing_scopes) with Session(self.engine) as session: self.assertEqual(0, session.query(PolicyOverride).count()) def test_tenant_policy_writer_cannot_loosen_global_system_ceiling(self) -> None: self.principal = self._principal("admin:policies:read", "admin:policies:write") policy = {"allowed_password_encryption_methods": ["aes", "zip_standard"]} response = self.client.put( "/api/v1/admin/campaign-archive-encryption/policies/system", json={"policy": policy} ) self.assertIn(response.status_code, (403, 409)) self.assertIn("system:settings:write", response.text) with Session(self.engine) as session: self.assertEqual(0, session.query(PolicyOverride).count()) self.assertEqual(0, session.query(ChangeSequenceEntry).count()) narrowed = self.client.put( "/api/v1/admin/campaign-archive-encryption/policies/tenant", json={"policy": {"allowed_password_encryption_methods": ["aes"]}}, ) self.assertEqual(200, narrowed.status_code, narrowed.text) def test_invalid_method_and_child_ceiling_still_fail_closed(self) -> None: path = "/api/v1/admin/campaign-archive-encryption/policies" invalid = self.client.put(f"{path}/system", json={"policy": {"allowed_password_encryption_methods": ["plaintext"]}}) self.assertEqual(422, invalid.status_code) child = self.client.put(f"{path}/tenant", json={"policy": {"allowed_password_encryption_methods": ["aes", "zip_standard"]}}) self.assertEqual(422, child.status_code) self.assertIn("parent", child.text) with Session(self.engine) as session: self.assertEqual(0, session.query(PolicyOverride).count()) def test_configuration_preview_preserves_only_known_non_secret_enum_lists(self) -> None: unsafe = { "allowed_password_encryption_methods": ["aes", "literal-secret"], "allowed_password_delivery_channels": {"password": "nested-secret"}, "password": "actual-secret", "arbitrary_field": ["unknown-secret"], } with Session(self.engine) as session: request = create_configuration_change_request( session, key="campaign_archive_encryption_policy", value=unsafe, actor_user_id="admin-user", actor_scopes=tuple(self.principal.scopes), dry_run=False, target={"scope_type": "system"}, ) self.assertEqual({key: "" for key in unsafe}, request["value_preview"]) self.assertNotIn("literal-secret", str(configuration_control_snapshot(session))) self.assertNotIn("actual-secret", str(configuration_control_snapshot(session))) for malformed in ("scalar-secret", ["list-secret"], None, 7): with self.subTest(malformed=type(malformed).__name__): malformed_request = create_configuration_change_request( session, key="campaign_archive_encryption_policy", value=malformed, actor_user_id="admin-user", actor_scopes=tuple(self.principal.scopes), dry_run=False, target={"scope_type": "system"}, ) self.assertEqual("", malformed_request["value_preview"]) snapshot = str(configuration_control_snapshot(session)) self.assertNotIn("scalar-secret", snapshot) self.assertNotIn("list-secret", snapshot) invalid = self.client.put( "/api/v1/admin/campaign-archive-encryption/policies/system", json={"policy": unsafe}, ) self.assertEqual(422, invalid.status_code) if __name__ == "__main__": unittest.main()