from __future__ import annotations import shutil import tempfile import unittest from pathlib import Path from fastapi import APIRouter from fastapi.testclient import TestClient from govoplan_core.audit.logging import audit_event, audit_operation_context from govoplan_core.core.events import ( EventActorRef, EventBus, EventObjectRef, EventTenantRef, PlatformEvent, current_event_trace, event_context, normalize_trace_id, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.base import Base from govoplan_core.server.fastapi import create_govoplan_app from tests.db_isolation import temporary_database class CoreEventTests(unittest.TestCase): def test_event_bus_adds_trace_ids_and_propagates_causation_to_nested_events(self) -> None: bus = EventBus() seen: list[PlatformEvent] = [] def parent_handler(event: PlatformEvent) -> None: seen.append(event) bus.publish(PlatformEvent(type="child.ready", module_id="demo")) bus.subscribe("parent.ready", parent_handler) bus.subscribe("child.ready", seen.append) bus.publish(PlatformEvent(type="parent.ready", module_id="demo", correlation_id="corr-1")) self.assertEqual(2, len(seen)) parent, child = seen self.assertEqual("corr-1", parent.correlation_id) self.assertIsNone(parent.causation_id) self.assertEqual("corr-1", child.correlation_id) self.assertEqual(parent.event_id, child.causation_id) def test_platform_event_serializes_governed_envelope_fields(self) -> None: event = PlatformEvent( type="case.permit.updated", module_id="cases", payload={"status": "approved"}, event_id="event-1", correlation_id="corr-1", causation_id="cause-1", actor=EventActorRef(type="user", id="user-1", label="Case worker"), tenant=EventTenantRef(id="tenant-1", slug="city", label="City Office"), subject=EventObjectRef(type="person", id="person-1", label="Applicant"), resource=EventObjectRef(type="permit", id="permit-1", label="Permit"), classification="confidential", ) payload = event.to_dict() self.assertEqual("case.permit.updated", payload["type"]) self.assertEqual("cases", payload["module_id"]) self.assertEqual({"status": "approved"}, payload["payload"]) self.assertEqual("event-1", payload["event_id"]) self.assertEqual("corr-1", payload["correlation_id"]) self.assertEqual("cause-1", payload["causation_id"]) self.assertEqual({"type": "user", "id": "user-1", "label": "Case worker"}, payload["actor"]) self.assertEqual({"id": "tenant-1", "slug": "city", "label": "City Office"}, payload["tenant"]) self.assertEqual({"type": "person", "id": "person-1", "label": "Applicant"}, payload["subject"]) self.assertEqual({"type": "permit", "id": "permit-1", "label": "Permit"}, payload["resource"]) self.assertEqual("confidential", payload["classification"]) def test_event_bus_preserves_envelope_fields_when_adding_trace(self) -> None: bus = EventBus() seen: list[PlatformEvent] = [] event = PlatformEvent( type="files.file.uploaded", module_id="files", actor=EventActorRef(type="api_key", id="key-1"), tenant=EventTenantRef(id="tenant-1"), subject=EventObjectRef(type="file", id="file-1"), resource=EventObjectRef(type="folder", id="folder-1"), classification="restricted", ) bus.subscribe("files.file.uploaded", seen.append) bus.publish(event) self.assertEqual(1, len(seen)) traced = seen[0] self.assertIsNotNone(traced.correlation_id) self.assertEqual(event.actor, traced.actor) self.assertEqual(event.tenant, traced.tenant) self.assertEqual(event.subject, traced.subject) self.assertEqual(event.resource, traced.resource) self.assertEqual("restricted", traced.classification) def test_request_correlation_middleware_sets_event_context_and_response_header(self) -> None: router = APIRouter() @router.get("/trace") def trace() -> dict[str, str | None]: active = current_event_trace() return {"correlation_id": active.correlation_id if active else None} app = create_govoplan_app( title="event test", version="test", registry=PlatformRegistry(), api_router=router, ) with TestClient(app) as client: response = client.get("/trace", headers={"X-Request-ID": "request-123"}) self.assertEqual(200, response.status_code, response.text) self.assertEqual("request-123", response.json()["correlation_id"]) self.assertEqual("request-123", response.headers["X-Correlation-ID"]) def test_audit_event_persists_trace_details_from_event_context(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-audit-trace-")) try: with temporary_database(f"sqlite:///{root / 'audit.db'}") as database: from govoplan_access.backend.db import models as access_models # noqa: F401 from govoplan_admin.backend.db import models as admin_models # noqa: F401 from govoplan_audit.backend.db import models as audit_models # noqa: F401 from govoplan_tenancy.backend.db import models as tenancy_models # noqa: F401 Base.metadata.create_all(bind=database.engine) with database.session() as session: with event_context(correlation_id="corr-1", causation_id="cause-1"): item = audit_event( session, tenant_id=None, scope="system", action="demo.changed", object_type="demo", object_id="demo-1", details={"password": "secret"}, ) self.assertEqual({"correlation_id": "corr-1", "causation_id": "cause-1"}, item.details["_trace"]) self.assertEqual("", item.details["password"]) finally: shutil.rmtree(root, ignore_errors=True) def test_audit_operation_context_keeps_lifecycle_ids_and_sanitizes_details(self) -> None: payload = audit_operation_context( module_id="mail", request_id="request-1", run_id="run-1", outcome="queued", trace={"correlation_id": " corr-1 ", "causation_id": "cause-1", "token": "secret"}, options={"migrate_database": True, "api_key": "secret"}, empty=None, ) self.assertEqual("mail", payload["module_id"]) self.assertEqual("request-1", payload["request_id"]) self.assertEqual("run-1", payload["run_id"]) self.assertEqual("queued", payload["outcome"]) self.assertEqual({"correlation_id": "corr-1", "causation_id": "cause-1"}, payload["_trace"]) self.assertEqual("", payload["options"]["api_key"]) self.assertNotIn("token", payload["_trace"]) self.assertNotIn("empty", payload) def test_trace_ids_are_normalized_for_headers_and_audit_input(self) -> None: self.assertEqual("abc-123", normalize_trace_id(" abc-123 ")) self.assertIsNone(normalize_trace_id("../bad")) self.assertIsNone(normalize_trace_id("x" * 129)) if __name__ == "__main__": unittest.main()