Add module event producer coverage
Some checks failed
Dependency Audit / dependency-audit (push) Has been cancelled
Module Matrix / module-matrix (push) Has been cancelled

This commit is contained in:
2026-07-10 18:40:38 +02:00
parent 83784ea19d
commit c61abe154c
10 changed files with 332 additions and 26 deletions

View File

@@ -14,6 +14,12 @@ dispatch**:
- Use `govoplan_core.core.events.PlatformEvent` for domain and platform events. - Use `govoplan_core.core.events.PlatformEvent` for domain and platform events.
- Use `EventBus` as the in-process dispatch contract for same-process module - Use `EventBus` as the in-process dispatch contract for same-process module
reactions that are safe to run inline. 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 - Persist durable integration/workflow events through a database outbox before
acknowledging the state change that produced them. acknowledging the state change that produced them.
- Drain the outbox through a small dispatcher process. The dispatcher may call - 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 Audit logging reads the current event context and stores trace IDs in
`details._trace`. Callers can also pass explicit `correlation_id` and `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 Admin and lifecycle code should use the compact operational detail shape
documented in `govoplan-audit/docs/AUDIT_TRACE_CONTEXT.md`. The core 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 - the compatibility `audit_event` helper while routes are still migrating
- retention orchestration that calls module capabilities - retention orchestration that calls module capabilities
Feature modules should record audit facts through a small audit API or future Feature modules should record audit facts through the shared helper or a future
`audit.sink` capability. They should not import audit storage internals. `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 ## Initial Domain Event Inventory

View File

@@ -7,7 +7,15 @@ from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.change_sequence import record_change 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_core.privacy.retention import sanitize_audit_details_for_policy
from govoplan_audit.backend.db.models import AuditLog from govoplan_audit.backend.db.models import AuditLog
@@ -34,6 +42,27 @@ SENSITIVE_DETAIL_KEYS = {
"build_token", "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: def _optional_text(value: object | None) -> str | None:
if value is None: if value is None:
@@ -135,6 +164,34 @@ def _restore_trace_after_policy(details: dict[str, Any], trace: dict[str, str])
return restored 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( def audit_event(
session: Session, session: Session,
*, *,
@@ -194,6 +251,7 @@ def audit_event(
actor_id=user_id or api_key_id, actor_id=user_id or api_key_id,
payload={"scope": scope, "action": action, "object_type": object_type, "object_id": object_id}, payload={"scope": scope, "action": action, "object_type": object_type, "object_id": object_id},
) )
_publish_audit_platform_event(item)
if commit: if commit:
session.commit() session.commit()
return item return item

View File

@@ -6,6 +6,7 @@ from typing import Any, Iterable, Sequence
from sqlalchemy import BigInteger, DateTime, Index, Integer, JSON, String, UniqueConstraint, func from sqlalchemy import BigInteger, DateTime, Index, Integer, JSON, String, UniqueConstraint, func
from sqlalchemy.orm import Mapped, Session, mapped_column 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 from govoplan_core.db.base import Base, utcnow
WATERMARK_PREFIX = "seq:" WATERMARK_PREFIX = "seq:"
@@ -95,9 +96,42 @@ def record_change(
payload=payload or {}, payload=payload or {},
) )
session.add(entry) session.add(entry)
_publish_change_event(entry)
return 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( def max_sequence_id(
session: Session, session: Session,
*, *,

View File

@@ -168,5 +168,26 @@ class EventBus:
handler(traced) 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]: def _compact_dict(value: Mapping[str, Any]) -> dict[str, Any]:
return {key: item for key, item in value.items() if item is not None} return {key: item for key, item in value.items() if item is not None}

View File

@@ -18,6 +18,7 @@ from govoplan_core.core.change_sequence import (
sequence_entries_since, sequence_entries_since,
sequence_watermark_is_expired, 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 from govoplan_core.db.base import Base
@@ -64,6 +65,46 @@ class ChangeSequenceTests(unittest.TestCase):
finally: finally:
engine.dispose() 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: def test_pruning_records_retention_floor_for_stale_watermarks(self) -> None:
engine = create_engine("sqlite:///:memory:") engine = create_engine("sqlite:///:memory:")
SessionLocal = sessionmaker(bind=engine) SessionLocal = sessionmaker(bind=engine)

View File

@@ -16,6 +16,7 @@ from govoplan_core.core.events import (
EventTenantRef, EventTenantRef,
PlatformEvent, PlatformEvent,
current_event_trace, current_event_trace,
event_bus_context,
event_context, event_context,
normalize_trace_id, normalize_trace_id,
) )
@@ -149,6 +150,83 @@ class CoreEventTests(unittest.TestCase):
finally: finally:
shutil.rmtree(root, ignore_errors=True) 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("<redacted>", 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: def test_audit_operation_context_keeps_lifecycle_ids_and_sanitizes_details(self) -> None:
payload = audit_operation_context( payload = audit_operation_context(
module_id="mail", module_id="mail",

View File

@@ -100,6 +100,7 @@ from govoplan_core.core.modules import MigrationSpec, ModuleManifest
from govoplan_core.core.registry import PlatformRegistry, RegistryError from govoplan_core.core.registry import PlatformRegistry, RegistryError
from govoplan_core.admin.models import SystemSettings from govoplan_core.admin.models import SystemSettings
from govoplan_core.db.base import Base 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.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.module_permissions import scopes_grant_compatible
from govoplan_core.security.permissions import scope_grants 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.config import GovoplanServerConfig
from govoplan_core.server.registry import available_module_manifests, build_platform_registry from govoplan_core.server.registry import available_module_manifests, build_platform_registry
from govoplan_core.server.route_validation import RouteCollisionError 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 from govoplan_files.backend.storage.backends import LocalFilesystemStorageBackend, StorageBackendError
_MANIFEST_PROJECT_MODULES = { _MANIFEST_PROJECT_MODULES = {
@@ -530,6 +532,52 @@ finally:
self.assertEqual(files_enabled, registry.integration_enabled("campaigns", "files")) self.assertEqual(files_enabled, registry.integration_enabled("campaigns", "files"))
self.assertEqual(mail_enabled, registry.integration_enabled("campaigns", "mail")) 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: def test_registry_rejects_missing_required_capability(self) -> None:
registry = PlatformRegistry() registry = PlatformRegistry()
registry.register(ModuleManifest( registry.register(ModuleManifest(

View File

@@ -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 (
<span className={["policy-path-help", className].filter(Boolean).join(" ")}>
{lines.map((line, index) =>
<span className={["policy-path-help-line", lineClassName].filter(Boolean).join(" ")} key={`${String(line)}-${index}`}>
{line}
</span>
)}
</span>);
}
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());
}

View File

@@ -15,6 +15,7 @@ import Button from "../../components/Button";
import Card from "../../components/Card"; import Card from "../../components/Card";
import DismissibleAlert from "../../components/DismissibleAlert"; import DismissibleAlert from "../../components/DismissibleAlert";
import LoadingFrame from "../../components/LoadingFrame"; import LoadingFrame from "../../components/LoadingFrame";
import PolicyPathHelp, { normalizePolicySourcePathItems, type NormalizedPolicySourcePathItem } from "../../components/PolicyPathHelp";
import { PolicyRow, PolicySection, PolicyTable } from "../../components/PolicyTable"; import { PolicyRow, PolicySection, PolicyTable } from "../../components/PolicyTable";
import type { PolicySourcePathItem } from "../../components/PolicySourcePath"; import type { PolicySourcePathItem } from "../../components/PolicySourcePath";
import FormField from "../../components/FormField"; import FormField from "../../components/FormField";
@@ -382,25 +383,12 @@ function policyDraftKey(policy: PrivacyRetentionPolicyPatch, isSystem: boolean):
return JSON.stringify(isSystem ? normalizeFullPolicy(policy) : normalizePatch(policy)); return JSON.stringify(isSystem ? normalizeFullPolicy(policy) : normalizePatch(policy));
} }
type PolicySourceItem = {
label: string;
policy?: Record<string, unknown> | null;
};
function retentionPolicyPathHelp(field: FieldDefinition, sources: PolicySourcePathItem[], scopeType: PrivacyRetentionPolicyScope): ReactNode { 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 <PolicyPathHelp lines={retentionPolicyPathLines(field, items)} />; return <PolicyPathHelp lines={retentionPolicyPathLines(field, items)} />;
} }
function PolicyPathHelp({ lines }: {lines: string[];}) { function retentionPolicyPathLines(field: FieldDefinition, items: NormalizedPolicySourcePathItem[]): string[] {
return (
<span className="policy-path-help">
{lines.map((line, index) => <span className="policy-path-help-line" key={`${line}-${index}`}>{line}</span>)}
</span>);
}
function retentionPolicyPathLines(field: FieldDefinition, items: PolicySourceItem[]): string[] {
if (items.length === 0) return ["i18n:govoplan-core.system_default.1f06f3ed"]; if (items.length === 0) return ["i18n:govoplan-core.system_default.1f06f3ed"];
const lines: string[] = []; const lines: string[] = [];
for (const [index, item] of items.entries()) { for (const [index, item] of items.entries()) {
@@ -428,13 +416,6 @@ function retentionSourceValue(field: FieldDefinition, sourcePolicy: Record<strin
return typeof value === "number" ? daysLabel(value) : value === null && isSystem ? daysLabel(null) : "i18n:govoplan-core.inherit.18f99833"; return typeof value === "number" ? daysLabel(value) : value === null && isSystem ? daysLabel(null) : "i18n:govoplan-core.inherit.18f99833";
} }
function normalizePolicySourceItems(items: PolicySourcePathItem[]): PolicySourceItem[] {
return items.map((item) => {
if (typeof item === "string") return { label: item };
return { label: item.label, policy: item.policy };
}).filter((item) => item.label.trim());
}
function asRecord(value: unknown): Record<string, unknown> { function asRecord(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {}; return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
} }

View File

@@ -54,6 +54,8 @@ export { default as PageTitle } from "./components/PageTitle";
export { default as PasswordField } from "./components/PasswordField"; export { default as PasswordField } from "./components/PasswordField";
export type { PasswordFieldProps } from "./components/PasswordField"; export type { PasswordFieldProps } from "./components/PasswordField";
export { PolicyRow, PolicySection, PolicyTable } from "./components/PolicyTable"; 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 { default as PolicySourcePath } from "./components/PolicySourcePath";
export type { PolicySourcePathItem, PolicySourcePathProps } from "./components/PolicySourcePath"; export type { PolicySourcePathItem, PolicySourcePathProps } from "./components/PolicySourcePath";
export { default as PolicyLockedHint } from "./components/PolicyLockedHint"; export { default as PolicyLockedHint } from "./components/PolicyLockedHint";