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 `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

View File

@@ -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

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.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,
*,

View File

@@ -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}

View File

@@ -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)

View File

@@ -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("<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:
payload = audit_operation_context(
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.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(

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 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<string, unknown> | 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 <PolicyPathHelp lines={retentionPolicyPathLines(field, items)} />;
}
function PolicyPathHelp({ lines }: {lines: 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[] {
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<strin
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> {
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 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";