From c61abe154cf6a5563979cb4244d6e73d3cff3bf3 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 10 Jul 2026 18:40:38 +0200 Subject: [PATCH] Add module event producer coverage --- docs/EVENTS_AND_AUDIT.md | 17 +++- src/govoplan_core/audit/logging.py | 60 +++++++++++++- src/govoplan_core/core/change_sequence.py | 34 ++++++++ src/govoplan_core/core/events.py | 21 +++++ tests/test_change_sequence.py | 41 ++++++++++ tests/test_core_events.py | 78 +++++++++++++++++++ tests/test_module_system.py | 48 ++++++++++++ webui/src/components/PolicyPathHelp.tsx | 32 ++++++++ .../privacy/RetentionPolicyManagement.tsx | 25 +----- webui/src/index.ts | 2 + 10 files changed, 332 insertions(+), 26 deletions(-) create mode 100644 webui/src/components/PolicyPathHelp.tsx diff --git a/docs/EVENTS_AND_AUDIT.md b/docs/EVENTS_AND_AUDIT.md index 7ebdb80..13d9101 100644 --- a/docs/EVENTS_AND_AUDIT.md +++ b/docs/EVENTS_AND_AUDIT.md @@ -14,6 +14,12 @@ dispatch**: - Use `govoplan_core.core.events.PlatformEvent` for domain and platform events. - Use `EventBus` as the in-process dispatch contract for same-process module reactions that are safe to run inline. +- Use the shared `audit_event` / `audit_from_principal` helper for audited + module actions. The helper persists the audit row and immediately publishes a + governed `PlatformEvent` whose `type` is the audit action. +- Use `record_change` for module delta feeds. It persists the change-sequence + row and immediately publishes a generic module change event such as + `mail.profile.updated`. - Persist durable integration/workflow events through a database outbox before acknowledging the state change that produced them. - Drain the outbox through a small dispatcher process. The dispatcher may call @@ -75,7 +81,9 @@ otherwise it generates a new ID. Responses include `X-Correlation-ID`. Audit logging reads the current event context and stores trace IDs in `details._trace`. Callers can also pass explicit `correlation_id` and -`causation_id` to `audit_event` or `audit_from_principal`. +`causation_id` to `audit_event` or `audit_from_principal`. The same trace is +copied into the emitted `PlatformEvent`, so audit rows and event subscribers can +be correlated without route-specific glue code. Admin and lifecycle code should use the compact operational detail shape documented in `govoplan-audit/docs/AUDIT_TRACE_CONTEXT.md`. The core @@ -99,8 +107,11 @@ additional detail values. - the compatibility `audit_event` helper while routes are still migrating - retention orchestration that calls module capabilities -Feature modules should record audit facts through a small audit API or future -`audit.sink` capability. They should not import audit storage internals. +Feature modules should record audit facts through the shared helper or a future +`audit.sink` capability. They should not import audit storage internals. Module +state changes that only need delta-feed visibility should go through +`record_change`; route-level business actions should still record a semantic +audit action. ## Initial Domain Event Inventory diff --git a/src/govoplan_core/audit/logging.py b/src/govoplan_core/audit/logging.py index 1142492..5a90cff 100644 --- a/src/govoplan_core/audit/logging.py +++ b/src/govoplan_core/audit/logging.py @@ -7,7 +7,15 @@ from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal from govoplan_core.core.change_sequence import record_change -from govoplan_core.core.events import current_event_trace, normalize_trace_id +from govoplan_core.core.events import ( + EventActorRef, + EventObjectRef, + EventTenantRef, + PlatformEvent, + current_event_trace, + normalize_trace_id, + publish_platform_event, +) from govoplan_core.privacy.retention import sanitize_audit_details_for_policy from govoplan_audit.backend.db.models import AuditLog @@ -34,6 +42,27 @@ SENSITIVE_DETAIL_KEYS = { "build_token", } +_ACTION_MODULE_PREFIXES = { + "api_key": "access", + "configuration_change": "access", + "configuration_package": "access", + "group": "access", + "profile": "access", + "role": "access", + "system_access": "access", + "system_account": "access", + "system_memberships": "access", + "system_role": "access", + "user": "access", + "tenant": "tenancy", + "privacy_retention": "policy", + "retention_policy": "policy", + "files": "files", + "mail": "mail", + "campaign": "campaigns", + "report": "campaigns", +} + def _optional_text(value: object | None) -> str | None: if value is None: @@ -135,6 +164,34 @@ def _restore_trace_after_policy(details: dict[str, Any], trace: dict[str, str]) return restored +def _module_id_for_audit_action(action: str) -> str: + prefix = action.split(".", 1)[0].strip().casefold() + return _ACTION_MODULE_PREFIXES.get(prefix, prefix or AUDIT_MODULE_ID) + + +def _publish_audit_platform_event(item: AuditLog) -> None: + trace = _compact_trace(item.details.get("_trace") if isinstance(item.details, Mapping) else None) + publish_platform_event( + PlatformEvent( + type=item.action, + module_id=_module_id_for_audit_action(item.action), + payload={ + "audit_log_id": item.id, + "scope": item.scope, + "details": dict(item.details or {}), + }, + correlation_id=trace.get("correlation_id"), + causation_id=trace.get("causation_id"), + actor=EventActorRef(type="user", id=item.user_id) if item.user_id else ( + EventActorRef(type="api_key", id=item.api_key_id) if item.api_key_id else None + ), + tenant=EventTenantRef(id=item.tenant_id) if item.tenant_id else None, + resource=EventObjectRef(type=item.object_type, id=item.object_id) if item.object_type else None, + classification="internal", + ) + ) + + def audit_event( session: Session, *, @@ -194,6 +251,7 @@ def audit_event( actor_id=user_id or api_key_id, payload={"scope": scope, "action": action, "object_type": object_type, "object_id": object_id}, ) + _publish_audit_platform_event(item) if commit: session.commit() return item diff --git a/src/govoplan_core/core/change_sequence.py b/src/govoplan_core/core/change_sequence.py index e499b37..fadd0d1 100644 --- a/src/govoplan_core/core/change_sequence.py +++ b/src/govoplan_core/core/change_sequence.py @@ -6,6 +6,7 @@ from typing import Any, Iterable, Sequence from sqlalchemy import BigInteger, DateTime, Index, Integer, JSON, String, UniqueConstraint, func from sqlalchemy.orm import Mapped, Session, mapped_column +from govoplan_core.core.events import EventActorRef, EventObjectRef, EventTenantRef, PlatformEvent, publish_platform_event from govoplan_core.db.base import Base, utcnow WATERMARK_PREFIX = "seq:" @@ -95,9 +96,42 @@ def record_change( payload=payload or {}, ) session.add(entry) + _publish_change_event(entry) return entry +def _publish_change_event(entry: ChangeSequenceEntry) -> None: + event_type = _change_event_type(entry.module_id, entry.resource_type, entry.operation) + publish_platform_event( + PlatformEvent( + type=event_type, + module_id=entry.module_id, + payload={ + "collection": entry.collection, + "operation": entry.operation, + "payload": dict(entry.payload or {}), + }, + actor=EventActorRef(type=entry.actor_type, id=entry.actor_id) if entry.actor_type else None, + tenant=EventTenantRef(id=entry.tenant_id) if entry.tenant_id else None, + resource=EventObjectRef(type=entry.resource_type, id=entry.resource_id), + classification="internal", + ) + ) + + +def _change_event_type(module_id: str, resource_type: str, operation: str) -> str: + resource = _event_segment(resource_type) + module = _event_segment(module_id) + if resource.startswith(f"{module}."): + resource = resource[len(module) + 1:] + return ".".join(item for item in (module, resource, _event_segment(operation)) if item) + + +def _event_segment(value: str) -> str: + compact = "".join(character if character.isalnum() else "." for character in value.casefold()) + return ".".join(part for part in compact.split(".") if part) + + def max_sequence_id( session: Session, *, diff --git a/src/govoplan_core/core/events.py b/src/govoplan_core/core/events.py index cbfbf22..412f5f7 100644 --- a/src/govoplan_core/core/events.py +++ b/src/govoplan_core/core/events.py @@ -168,5 +168,26 @@ class EventBus: handler(traced) +_default_event_bus = EventBus() +_current_event_bus: ContextVar[EventBus | None] = ContextVar("govoplan_event_bus", default=None) + + +def platform_event_bus() -> EventBus: + return _current_event_bus.get() or _default_event_bus + + +@contextmanager +def event_bus_context(bus: EventBus): + token = _current_event_bus.set(bus) + try: + yield bus + finally: + _current_event_bus.reset(token) + + +def publish_platform_event(event: PlatformEvent) -> None: + platform_event_bus().publish(event) + + def _compact_dict(value: Mapping[str, Any]) -> dict[str, Any]: return {key: item for key, item in value.items() if item is not None} diff --git a/tests/test_change_sequence.py b/tests/test_change_sequence.py index 69f323e..34d9f83 100644 --- a/tests/test_change_sequence.py +++ b/tests/test_change_sequence.py @@ -18,6 +18,7 @@ from govoplan_core.core.change_sequence import ( sequence_entries_since, sequence_watermark_is_expired, ) +from govoplan_core.core.events import EventActorRef, EventBus, EventObjectRef, EventTenantRef, PlatformEvent, event_bus_context from govoplan_core.db.base import Base @@ -64,6 +65,46 @@ class ChangeSequenceTests(unittest.TestCase): finally: engine.dispose() + def test_record_change_publishes_module_change_event(self) -> None: + engine = create_engine("sqlite:///:memory:") + SessionLocal = sessionmaker(bind=engine) + try: + Base.metadata.create_all(bind=engine, tables=[ChangeSequenceEntry.__table__, ChangeSequenceRetentionFloor.__table__]) + seen: list[PlatformEvent] = [] + bus = EventBus() + bus.subscribe("*", seen.append) + cases = ( + ("files", "files.assets", "file", "file-1", "created", "files.file.created"), + ("mail", "mail.profiles", "mail_profile", "profile-1", "updated", "mail.profile.updated"), + ("campaigns", "campaigns.jobs", "campaign_job", "job-1", "updated", "campaigns.campaign.job.updated"), + ) + with SessionLocal() as session: + with event_bus_context(bus): + for module_id, collection, resource_type, resource_id, operation, _event_type in cases: + record_change( + session, + tenant_id="tenant-1", + module_id=module_id, + collection=collection, + resource_type=resource_type, + resource_id=resource_id, + operation=operation, + actor_type="user", + actor_id="user-1", + payload={"scope_type": "tenant"}, + ) + + self.assertEqual([case[-1] for case in cases], [event.type for event in seen]) + event = seen[1] + self.assertEqual("mail", event.module_id) + self.assertEqual(EventActorRef(type="user", id="user-1"), event.actor) + self.assertEqual(EventTenantRef(id="tenant-1"), event.tenant) + self.assertEqual(EventObjectRef(type="mail_profile", id="profile-1"), event.resource) + self.assertEqual("mail.profiles", event.payload["collection"]) + self.assertEqual({"scope_type": "tenant"}, event.payload["payload"]) + finally: + engine.dispose() + def test_pruning_records_retention_floor_for_stale_watermarks(self) -> None: engine = create_engine("sqlite:///:memory:") SessionLocal = sessionmaker(bind=engine) diff --git a/tests/test_core_events.py b/tests/test_core_events.py index efd4589..e687c4c 100644 --- a/tests/test_core_events.py +++ b/tests/test_core_events.py @@ -16,6 +16,7 @@ from govoplan_core.core.events import ( EventTenantRef, PlatformEvent, current_event_trace, + event_bus_context, event_context, normalize_trace_id, ) @@ -149,6 +150,83 @@ class CoreEventTests(unittest.TestCase): finally: shutil.rmtree(root, ignore_errors=True) + def test_audit_event_publishes_governed_platform_event(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-audit-event-")) + 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) + seen: list[PlatformEvent] = [] + bus = EventBus() + bus.subscribe("*", seen.append) + with database.session() as session: + with event_bus_context(bus), event_context(correlation_id="corr-1"): + item = audit_event( + session, + tenant_id="tenant-1", + user_id="user-1", + scope="tenant", + action="user.updated", + object_type="user", + object_id="user-2", + details={"password": "secret", "field": "display_name"}, + ) + + action_events = [event for event in seen if event.type == "user.updated"] + self.assertEqual(1, len(action_events)) + event = action_events[0] + self.assertEqual("access", event.module_id) + self.assertEqual("corr-1", event.correlation_id) + self.assertEqual(EventActorRef(type="user", id="user-1"), event.actor) + self.assertEqual(EventTenantRef(id="tenant-1"), event.tenant) + self.assertEqual(EventObjectRef(type="user", id="user-2"), event.resource) + self.assertEqual(item.id, event.payload["audit_log_id"]) + self.assertEqual("", event.payload["details"]["password"]) + finally: + shutil.rmtree(root, ignore_errors=True) + + def test_audit_events_map_existing_module_action_prefixes(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-audit-module-events-")) + 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) + seen: list[PlatformEvent] = [] + bus = EventBus() + bus.subscribe("*", seen.append) + cases = { + "user.created": "access", + "tenant.created": "tenancy", + "privacy_retention.policy_updated": "policy", + "files.connector.imported": "files", + "campaign.created": "campaigns", + "report.email_sent": "campaigns", + } + with database.session() as session: + with event_bus_context(bus): + for action in cases: + audit_event( + session, + tenant_id="tenant-1", + user_id="user-1", + action=action, + object_type="demo", + object_id=action, + ) + + by_type = {event.type: event.module_id for event in seen if event.type in cases} + self.assertEqual(cases, by_type) + 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", diff --git a/tests/test_module_system.py b/tests/test_module_system.py index daccb1a..5a273e4 100644 --- a/tests/test_module_system.py +++ b/tests/test_module_system.py @@ -100,6 +100,7 @@ from govoplan_core.core.modules import MigrationSpec, ModuleManifest from govoplan_core.core.registry import PlatformRegistry, RegistryError from govoplan_core.admin.models import SystemSettings from govoplan_core.db.base import Base +from govoplan_core.db.bootstrap import bootstrap_dev_data from govoplan_core.db.session import configure_database as configure_database_handle, get_database, reset_database from govoplan_core.security.module_permissions import scopes_grant_compatible from govoplan_core.security.permissions import scope_grants @@ -107,6 +108,7 @@ from govoplan_core.server.app import create_app from govoplan_core.server.config import GovoplanServerConfig from govoplan_core.server.registry import available_module_manifests, build_platform_registry from govoplan_core.server.route_validation import RouteCollisionError +from govoplan_core.tenancy.scope import create_scope_tables from govoplan_files.backend.storage.backends import LocalFilesystemStorageBackend, StorageBackendError _MANIFEST_PROJECT_MODULES = { @@ -530,6 +532,52 @@ finally: self.assertEqual(files_enabled, registry.integration_enabled("campaigns", "files")) self.assertEqual(mail_enabled, registry.integration_enabled("campaigns", "mail")) + def test_tenant_lifecycle_with_product_modules_installed_and_absent(self) -> None: + cases = ( + ("tenancy_only", ()), + ("files_only", ("files",)), + ("mail_only", ("mail",)), + ("campaign_only", ("campaigns",)), + ("full_product", ("files", "mail", "campaigns")), + ) + for name, product_modules in cases: + with self.subTest(name=name): + app, _settings_obj = self._app_for_modules(("tenancy", *product_modules)) + database = get_database() + create_scope_tables(database.engine) + Base.metadata.create_all(bind=database.engine) + with database.session() as session: + bootstrap_dev_data(session, user_password="test-admin") + + with TestClient(app) as client: + login = client.post( + "/api/v1/auth/login", + json={"email": "admin@example.local", "password": "test-admin"}, + ) + self.assertEqual(200, login.status_code, login.text) + headers = {"Authorization": f"Bearer {login.json()['access_token']}"} + + created = client.post( + "/api/v1/admin/tenants", + headers=headers, + json={ + "slug": f"lifecycle-{name}", + "name": f"Lifecycle {name}", + "settings": {}, + }, + ) + self.assertEqual(201, created.status_code, created.text) + tenant_id = created.json()["id"] + + updated = client.patch( + f"/api/v1/admin/tenants/{tenant_id}", + headers=headers, + json={"name": f"Lifecycle {name} Updated", "is_active": False}, + ) + self.assertEqual(200, updated.status_code, updated.text) + self.assertEqual(f"Lifecycle {name} Updated", updated.json()["name"]) + self.assertFalse(updated.json()["is_active"]) + def test_registry_rejects_missing_required_capability(self) -> None: registry = PlatformRegistry() registry.register(ModuleManifest( diff --git a/webui/src/components/PolicyPathHelp.tsx b/webui/src/components/PolicyPathHelp.tsx new file mode 100644 index 0000000..d7fb98e --- /dev/null +++ b/webui/src/components/PolicyPathHelp.tsx @@ -0,0 +1,32 @@ +import type { ReactNode } from "react"; +import type { PolicySourcePathItem } from "./PolicySourcePath"; + +export type PolicyPathHelpProps = { + lines: ReactNode[]; + className?: string; + lineClassName?: string; +}; + +export type NormalizedPolicySourcePathItem = { + label: string; + policy?: unknown; +}; + +export default function PolicyPathHelp({ lines, className = "", lineClassName = "" }: PolicyPathHelpProps) { + return ( + + {lines.map((line, index) => + + {line} + + )} + ); + +} + +export function normalizePolicySourcePathItems(items: PolicySourcePathItem[]): NormalizedPolicySourcePathItem[] { + return items.map((item) => { + if (typeof item === "string") return { label: item }; + return { label: item.label, policy: item.policy }; + }).filter((item) => item.label.trim()); +} diff --git a/webui/src/features/privacy/RetentionPolicyManagement.tsx b/webui/src/features/privacy/RetentionPolicyManagement.tsx index c8452a8..07eaa7c 100644 --- a/webui/src/features/privacy/RetentionPolicyManagement.tsx +++ b/webui/src/features/privacy/RetentionPolicyManagement.tsx @@ -15,6 +15,7 @@ import Button from "../../components/Button"; import Card from "../../components/Card"; import DismissibleAlert from "../../components/DismissibleAlert"; import LoadingFrame from "../../components/LoadingFrame"; +import PolicyPathHelp, { normalizePolicySourcePathItems, type NormalizedPolicySourcePathItem } from "../../components/PolicyPathHelp"; import { PolicyRow, PolicySection, PolicyTable } from "../../components/PolicyTable"; import type { PolicySourcePathItem } from "../../components/PolicySourcePath"; import FormField from "../../components/FormField"; @@ -382,25 +383,12 @@ function policyDraftKey(policy: PrivacyRetentionPolicyPatch, isSystem: boolean): return JSON.stringify(isSystem ? normalizeFullPolicy(policy) : normalizePatch(policy)); } -type PolicySourceItem = { - label: string; - policy?: Record | null; -}; - function retentionPolicyPathHelp(field: FieldDefinition, sources: PolicySourcePathItem[], scopeType: PrivacyRetentionPolicyScope): ReactNode { - const items = normalizePolicySourceItems(sources.length > 0 ? sources : retentionPolicySourcePath(scopeType)); + const items = normalizePolicySourcePathItems(sources.length > 0 ? sources : retentionPolicySourcePath(scopeType)); return ; } -function PolicyPathHelp({ lines }: {lines: string[];}) { - return ( - - {lines.map((line, index) => {line})} - ); - -} - -function retentionPolicyPathLines(field: FieldDefinition, items: PolicySourceItem[]): string[] { +function retentionPolicyPathLines(field: FieldDefinition, items: NormalizedPolicySourcePathItem[]): string[] { if (items.length === 0) return ["i18n:govoplan-core.system_default.1f06f3ed"]; const lines: string[] = []; for (const [index, item] of items.entries()) { @@ -428,13 +416,6 @@ function retentionSourceValue(field: FieldDefinition, sourcePolicy: Record { - if (typeof item === "string") return { label: item }; - return { label: item.label, policy: item.policy }; - }).filter((item) => item.label.trim()); -} - function asRecord(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; } diff --git a/webui/src/index.ts b/webui/src/index.ts index 11adacd..03588f5 100644 --- a/webui/src/index.ts +++ b/webui/src/index.ts @@ -54,6 +54,8 @@ export { default as PageTitle } from "./components/PageTitle"; export { default as PasswordField } from "./components/PasswordField"; export type { PasswordFieldProps } from "./components/PasswordField"; export { PolicyRow, PolicySection, PolicyTable } from "./components/PolicyTable"; +export { default as PolicyPathHelp, normalizePolicySourcePathItems } from "./components/PolicyPathHelp"; +export type { NormalizedPolicySourcePathItem, PolicyPathHelpProps } from "./components/PolicyPathHelp"; export { default as PolicySourcePath } from "./components/PolicySourcePath"; export type { PolicySourcePathItem, PolicySourcePathProps } from "./components/PolicySourcePath"; export { default as PolicyLockedHint } from "./components/PolicyLockedHint";