From 26ae0341535f172e9bcdf413619e2c144641b643 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 15:02:42 +0200 Subject: [PATCH] Add governed automation contracts --- docs/AUTOMATION_CONTRACTS.md | 49 ++++++++++ docs/POLICY_CONTRACTS.md | 15 +++ src/govoplan_core/celery_app.py | 48 ++++++++++ src/govoplan_core/core/access.py | 5 + src/govoplan_core/core/automation.py | 98 ++++++++++++++++++++ src/govoplan_core/core/dataflows.py | 44 +++++++++ src/govoplan_core/core/policy.py | 70 ++++++++++++++ tests/test_automation_contract.py | 75 +++++++++++++++ tests/test_dataflow_trigger_worker.py | 56 +++++++++++ tests/test_definition_governance_contract.py | 79 ++++++++++++++++ 10 files changed, 539 insertions(+) create mode 100644 docs/AUTOMATION_CONTRACTS.md create mode 100644 src/govoplan_core/core/automation.py create mode 100644 tests/test_automation_contract.py create mode 100644 tests/test_dataflow_trigger_worker.py create mode 100644 tests/test_definition_governance_contract.py diff --git a/docs/AUTOMATION_CONTRACTS.md b/docs/AUTOMATION_CONTRACTS.md new file mode 100644 index 0000000..3b77300 --- /dev/null +++ b/docs/AUTOMATION_CONTRACTS.md @@ -0,0 +1,49 @@ +# Automation Contracts + +Core defines provider-neutral automation contracts. It does not own domain +schedules, Workflow graphs, or Dataflow execution. + +## Invocation Envelope + +`AutomationInvocation` classifies a start as `manual`, `api`, `schedule`, +`event`, `workflow`, `dependency`, `retry`, or `backfill`. It carries opaque +trigger and delivery references, event identity, correlation and causation +IDs, scheduled time, requesting actor, and bounded metadata. Domain runs store +this envelope with their immutable definition revision. + +## Current Authorization + +An automated trigger must not persist a user session, bearer token, API key, +or a snapshot of all current permissions. It stores: + +- tenant, account, and membership IDs; +- an opaque authorization reference; +- the minimum scopes required by the pinned definition and output target. + +At delivery time the optional +`auth.automationPrincipalProvider` capability resolves current account, +membership, role, group, function, and delegation state. It intersects current +authorization with the stored grant. Missing, inactive, or reduced +authorization blocks the delivery before effects occur. + +## Definition Governance + +The optional `policy.definitionGovernance` capability evaluates `view`, +`edit`, `run`, `reuse`, `derive`, and `automate` for system, tenant, group, and +user definitions. A decision contains an ordered source path and effective +limits. Derived definitions pin their source revision and hash and retain +ancestor ceilings. Templates are reusable definitions and cannot run on their +own. + +Without Policy, domain modules use a conservative tenant-local fallback: +local definitions remain viewable/editable and active complete flows may run; +inheritance, reuse, derivation, and automation are unavailable. + +## Delivery Durability + +Domain trigger implementations persist idempotent deliveries before running. +The existing `PlatformEvent` bus is process-local and is not a durable +automation source. A transactional Core event/outbox bridge is still required +to guarantee capture across commits, restarts, and multiple workers. Until +that bridge exists, direct event ingestion must be authenticated, tenant +scoped, and limited to public or internal event envelopes. diff --git a/docs/POLICY_CONTRACTS.md b/docs/POLICY_CONTRACTS.md index 06ce589..baed745 100644 --- a/docs/POLICY_CONTRACTS.md +++ b/docs/POLICY_CONTRACTS.md @@ -13,6 +13,7 @@ consistent while each module still owns its domain rules. | RBAC/access policy | `govoplan-access` | access capabilities in `govoplan_core.core.access` | Permission decisions should use access capability contracts. Explain responses should adopt `PolicyDecision` when an API-level explanation is added. | | Governance defaults | `govoplan-admin` plus `govoplan-access` materializer | admin settings, governance template routes, access materialization capability | System governance can block tenant-local groups, roles, and API keys. | | Delegation and ownership policy | access/campaign/mail/files modules | capability checks and owner-scoped APIs | Source provenance should use this contract when policies become externally explainable. | +| Definition governance | `govoplan-policy` | capability `policy.definitionGovernance` | Resolves view, edit, run/start, reuse, derive, and automate for system, tenant, group, and user Dataflow/Workflow definitions. | ## Policy Decision @@ -111,6 +112,20 @@ matching compatibility DTOs only for its legacy admin surface. Admin overview responses remain module-local because the same counters are exposed from different menu contexts and are not yet a separately versioned platform API. +## Definition Governance + +Dataflow and Workflow submit a `DefinitionGovernanceRequest` using only stable +scope, principal, status, definition-kind, and limit fields. Policy returns a +standard `PolicyDecision`. System definitions may be inherited as read-only; +group and user definitions are visible only in matching contexts. Templates +may be viewed and derived but never run or automated. A derived definition +passes its pinned ancestor limits back through the request context, and Policy +applies those limits as ceilings rather than defaults that can be broadened. + +When the capability is absent, modules must not silently emulate cross-scope +inheritance. Their conservative fallback is limited to local tenant +definitions and disables reuse, derivation, and automation. + ## Frontend Contract Policy UIs must: diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index c3ca1cd..c6d663a 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -4,6 +4,10 @@ from celery import Celery from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CampaignDeliveryTaskProvider from govoplan_core.core.calendar import CAPABILITY_CALENDAR_OUTBOX, CalendarOutboxProvider +from govoplan_core.core.dataflows import ( + CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, + DataflowTriggerDispatcher, +) from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids from govoplan_core.core.modules import ModuleContext from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH, NotificationDispatchProvider @@ -29,6 +33,7 @@ celery.conf.update( "govoplan.notifications.deliver": {"queue": "notifications"}, "govoplan.notifications.deliver_pending": {"queue": "notifications"}, "govoplan.calendar.dispatch_outbox": {"queue": "calendar"}, + "govoplan.dataflow.dispatch_triggers": {"queue": "dataflow"}, }, worker_prefetch_multiplier=1, task_acks_late=True, @@ -39,6 +44,11 @@ celery.conf.update( "schedule": 60.0, "args": (None, 100), }, + "dataflow-triggers-every-minute": { + "task": "govoplan.dataflow.dispatch_triggers", + "schedule": 60.0, + "args": (100,), + }, }, ) @@ -86,6 +96,18 @@ def _calendar_outbox() -> CalendarOutboxProvider | None: return capability +def _dataflow_trigger_dispatcher() -> DataflowTriggerDispatcher | None: + registry = _platform_registry() + if not registry.has_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER): + return None + capability = registry.require_capability( + CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER + ) + if not isinstance(capability, DataflowTriggerDispatcher): + raise RuntimeError("Dataflow trigger dispatcher capability is invalid") + return capability + + @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0) def send_email(self, job_id: str): """Send one explicitly queued campaign job. @@ -149,3 +171,29 @@ def dispatch_calendar_outbox(self, tenant_id: str | None = None, limit: int = 50 result = dict(provider.dispatch_due(session, tenant_id=tenant_id, limit=limit)) session.commit() return result + + +@celery.task( + name="govoplan.dataflow.dispatch_triggers", + bind=True, + max_retries=0, +) +def dispatch_dataflow_triggers(self, limit: int = 100): + """Drain durable Dataflow trigger deliveries and due schedules.""" + + from govoplan_core.db.session import get_database + + with get_database().SessionLocal() as session: + provider = _dataflow_trigger_dispatcher() + if provider is None: + return { + "queued": 0, + "processed": 0, + "succeeded": 0, + "failed": 0, + "blocked": 0, + "skipped": 0, + } + result = dict(provider.dispatch_due(session, limit=limit)) + session.commit() + return result diff --git a/src/govoplan_core/core/access.py b/src/govoplan_core/core/access.py index 5d4fb62..2d57e16 100644 --- a/src/govoplan_core/core/access.py +++ b/src/govoplan_core/core/access.py @@ -28,6 +28,9 @@ CAPABILITY_AUDIT_RECORDER = "audit.recorder" CAPABILITY_AUDIT_RETENTION = "audit.retention" CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER = "auth.apiPrincipalProvider" +CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER = ( + "auth.automationPrincipalProvider" +) CAPABILITY_AUTH_PRINCIPAL_RESOLVER = "auth.principalResolver" CAPABILITY_AUTH_PERMISSION_EVALUATOR = "auth.permissionEvaluator" CAPABILITY_AUTH_TENANT_CONTEXT_SWITCHER = "auth.tenantContextSwitcher" @@ -48,6 +51,7 @@ ACCESS_CAPABILITY_NAMES = frozenset( CAPABILITY_AUDIT_RECORDER, CAPABILITY_AUDIT_RETENTION, CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER, + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_TENANT_CONTEXT_SWITCHER, @@ -57,6 +61,7 @@ ACCESS_CAPABILITY_NAMES = frozenset( AUTH_CAPABILITY_NAMES = frozenset( { CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER, + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_TENANT_CONTEXT_SWITCHER, diff --git a/src/govoplan_core/core/automation.py b/src/govoplan_core/core/automation.py new file mode 100644 index 0000000..b0392de --- /dev/null +++ b/src/govoplan_core/core/automation.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal, Protocol, runtime_checkable + +from govoplan_core.core.access import ( + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, +) + +AutomationInvocationKind = Literal[ + "manual", + "api", + "schedule", + "event", + "workflow", + "dependency", + "retry", + "backfill", +] + + +@dataclass(frozen=True, slots=True) +class AutomationInvocation: + kind: AutomationInvocationKind = "manual" + trigger_ref: str | None = None + delivery_ref: str | None = None + event_id: str | None = None + event_type: str | None = None + correlation_id: str | None = None + causation_id: str | None = None + scheduled_for: datetime | None = None + requested_by: str | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class AutomationPrincipalRequest: + tenant_id: str + account_id: str + membership_id: str + authorization_ref: str + grant_scopes: tuple[str, ...] + context: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class AutomationPrincipalResolution: + allowed: bool + principal: object | None = None + reason: str | None = None + granted_scopes: tuple[str, ...] = () + missing_scopes: tuple[str, ...] = () + provenance: Mapping[str, object] = field(default_factory=dict) + + +@runtime_checkable +class AutomationPrincipalProvider(Protocol): + def resolve_automation_principal( + self, + session: object, + *, + request: AutomationPrincipalRequest, + ) -> AutomationPrincipalResolution: + ... + + +def automation_principal_provider( + registry: object | None, +) -> AutomationPrincipalProvider | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability( + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + ) + ): + return None + capability = registry.capability( + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + ) + return ( + capability + if isinstance(capability, AutomationPrincipalProvider) + else None + ) + + +__all__ = [ + "CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER", + "AutomationInvocation", + "AutomationInvocationKind", + "AutomationPrincipalProvider", + "AutomationPrincipalRequest", + "AutomationPrincipalResolution", + "automation_principal_provider", +] diff --git a/src/govoplan_core/core/dataflows.py b/src/govoplan_core/core/dataflows.py index 4b3f992..2997dec 100644 --- a/src/govoplan_core/core/dataflows.py +++ b/src/govoplan_core/core/dataflows.py @@ -5,8 +5,12 @@ from dataclasses import dataclass, field from datetime import datetime from typing import Protocol, runtime_checkable +from govoplan_core.core.automation import AutomationInvocation +from govoplan_core.core.events import PlatformEvent + CAPABILITY_DATAFLOW_RUN_LIFECYCLE = "dataflow.runLifecycle" +CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER = "dataflow.triggerDispatcher" class DataflowRunError(ValueError): @@ -44,6 +48,9 @@ class DataflowRunRequest: idempotency_key: str row_limit: int = 500 publication: DataflowPublicationTarget | None = None + invocation: AutomationInvocation = field( + default_factory=AutomationInvocation + ) @dataclass(frozen=True, slots=True) @@ -59,6 +66,9 @@ class DataflowRunDescriptor: output_publication_ref: str | None = None output_datasource_ref: str | None = None output_materialization_ref: str | None = None + invocation_kind: str = "manual" + trigger_ref: str | None = None + delivery_ref: str | None = None error: str | None = None started_at: datetime | None = None finished_at: datetime | None = None @@ -96,6 +106,26 @@ class DataflowRunLifecycleProvider(Protocol): ... +@runtime_checkable +class DataflowTriggerDispatcher(Protocol): + def dispatch_due( + self, + session: object, + *, + now: datetime | None = None, + limit: int = 50, + ) -> Mapping[str, object]: + ... + + def ingest_event( + self, + session: object, + *, + event: PlatformEvent, + ) -> Mapping[str, object]: + ... + + def dataflow_run_lifecycle( registry: object | None, ) -> DataflowRunLifecycleProvider | None: @@ -103,6 +133,17 @@ def dataflow_run_lifecycle( return capability if isinstance(capability, DataflowRunLifecycleProvider) else None +def dataflow_trigger_dispatcher( + registry: object | None, +) -> DataflowTriggerDispatcher | None: + capability = _capability(registry, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER) + return ( + capability + if isinstance(capability, DataflowTriggerDispatcher) + else None + ) + + def _capability(registry: object | None, name: str) -> object | None: if ( registry is None @@ -116,6 +157,7 @@ def _capability(registry: object | None, name: str) -> object | None: __all__ = [ "CAPABILITY_DATAFLOW_RUN_LIFECYCLE", + "CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER", "DataflowPublicationTarget", "DataflowRunConflictError", "DataflowRunDescriptor", @@ -124,5 +166,7 @@ __all__ = [ "DataflowRunNotFoundError", "DataflowRunRequest", "DataflowRunUnavailableError", + "DataflowTriggerDispatcher", "dataflow_run_lifecycle", + "dataflow_trigger_dispatcher", ] diff --git a/src/govoplan_core/core/policy.py b/src/govoplan_core/core/policy.py index a5512b2..798c4b0 100644 --- a/src/govoplan_core/core/policy.py +++ b/src/govoplan_core/core/policy.py @@ -4,11 +4,24 @@ from dataclasses import dataclass, field from typing import Any, Iterable, Literal, Mapping, Protocol, cast, runtime_checkable from urllib.parse import quote, unquote +from govoplan_core.core.access import PrincipalRef + PolicyScopeType = Literal["system", "tenant", "user", "group", "campaign"] SchedulingParticipantVisibility = Literal["aggregates_only", "names_and_statuses"] +DefinitionScopeType = Literal["system", "tenant", "group", "user"] +DefinitionKind = Literal["flow", "template"] +DefinitionGovernanceAction = Literal[ + "view", + "edit", + "run", + "reuse", + "derive", + "automate", +] CAPABILITY_POLICY_PRIVACY_RETENTION = "policy.privacyRetention" CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY = "policy.schedulingParticipantPrivacy" +CAPABILITY_POLICY_DEFINITION_GOVERNANCE = "policy.definitionGovernance" POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ("system", "tenant", "user", "group", "campaign") @@ -128,6 +141,63 @@ class PolicyDecision: } +@dataclass(frozen=True, slots=True) +class DefinitionScopeRef: + scope_type: DefinitionScopeType + scope_id: str | None = None + + @property + def path(self) -> str: + return policy_source_path(self.scope_type, self.scope_id) + + +@dataclass(frozen=True, slots=True) +class DefinitionGovernanceRequest: + module_id: str + definition_ref: str + tenant_id: str + definition_scope: DefinitionScopeRef + target_scope: DefinitionScopeRef + definition_kind: DefinitionKind + action: DefinitionGovernanceAction + actor: PrincipalRef + status: str = "draft" + inherit_to_lower_scopes: bool = False + allow_run: bool = True + allow_reuse: bool = False + allow_automation: bool = False + context: Mapping[str, Any] = field(default_factory=dict) + + +@runtime_checkable +class DefinitionGovernancePolicy(Protocol): + def resolve_definition_action( + self, + *, + request: DefinitionGovernanceRequest, + ) -> PolicyDecision: + ... + + +def definition_governance_policy( + registry: object | None, +) -> DefinitionGovernancePolicy | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability( + CAPABILITY_POLICY_DEFINITION_GOVERNANCE + ) + ): + return None + capability = registry.capability(CAPABILITY_POLICY_DEFINITION_GOVERNANCE) + return ( + capability + if isinstance(capability, DefinitionGovernancePolicy) + else None + ) + + @dataclass(frozen=True, slots=True) class SchedulingParticipantPrivacyRequest: """Context for resolving what one Scheduling participant may see. diff --git a/tests/test_automation_contract.py b/tests/test_automation_contract.py new file mode 100644 index 0000000..15ae627 --- /dev/null +++ b/tests/test_automation_contract.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.automation import ( + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, + AutomationInvocation, + AutomationPrincipalProvider, + AutomationPrincipalRequest, + AutomationPrincipalResolution, + automation_principal_provider, +) +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.registry import PlatformRegistry + + +class _Provider: + def resolve_automation_principal(self, session, *, request): + del session + return AutomationPrincipalResolution( + allowed=True, + principal={"account_id": request.account_id}, + granted_scopes=request.grant_scopes, + ) + + +class AutomationContractTests(unittest.TestCase): + def test_invocation_records_stable_trigger_provenance(self) -> None: + invocation = AutomationInvocation( + kind="event", + trigger_ref="dataflow-trigger:1", + event_id="event-1", + event_type="files.uploaded", + ) + + self.assertEqual("event", invocation.kind) + self.assertEqual("event-1", invocation.event_id) + + def test_principal_provider_is_runtime_resolved(self) -> None: + provider = _Provider() + self.assertIsInstance(provider, AutomationPrincipalProvider) + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="automation_contract_test", + name="Automation contract test", + version="test", + capability_factories={ + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER: ( + lambda context: provider + ), + }, + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + + self.assertIs(provider, automation_principal_provider(registry)) + result = provider.resolve_automation_principal( + object(), + request=AutomationPrincipalRequest( + tenant_id="tenant-1", + account_id="account-1", + membership_id="membership-1", + authorization_ref="trigger:1", + grant_scopes=("dataflow:pipeline:run",), + ), + ) + self.assertTrue(result.allowed) + self.assertEqual(("dataflow:pipeline:run",), result.granted_scopes) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_dataflow_trigger_worker.py b/tests/test_dataflow_trigger_worker.py new file mode 100644 index 0000000..fb1e69f --- /dev/null +++ b/tests/test_dataflow_trigger_worker.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from govoplan_core.celery_app import celery, dispatch_dataflow_triggers + + +class DataflowTriggerWorkerTests(unittest.TestCase): + def test_worker_commits_dispatch_outcome(self) -> None: + session = MagicMock() + database = MagicMock() + database.SessionLocal.return_value.__enter__.return_value = session + provider = MagicMock() + provider.dispatch_due.return_value = { + "queued": 1, + "processed": 1, + "succeeded": 1, + "failed": 0, + "blocked": 0, + "skipped": 0, + } + + with ( + patch( + "govoplan_core.celery_app._dataflow_trigger_dispatcher", + return_value=provider, + ), + patch( + "govoplan_core.db.session.get_database", + return_value=database, + ), + ): + result = dispatch_dataflow_triggers.run(25) + + provider.dispatch_due.assert_called_once_with(session, limit=25) + session.commit.assert_called_once_with() + self.assertEqual(result["succeeded"], 1) + + def test_worker_route_and_periodic_dispatch_are_registered(self) -> None: + self.assertEqual( + celery.conf.task_routes["govoplan.dataflow.dispatch_triggers"], + {"queue": "dataflow"}, + ) + schedule = celery.conf.beat_schedule[ + "dataflow-triggers-every-minute" + ] + self.assertEqual( + schedule["task"], + "govoplan.dataflow.dispatch_triggers", + ) + self.assertEqual(schedule["schedule"], 60.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_definition_governance_contract.py b/tests/test_definition_governance_contract.py new file mode 100644 index 0000000..29bb7d1 --- /dev/null +++ b/tests/test_definition_governance_contract.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_DEFINITION_GOVERNANCE, + DefinitionGovernancePolicy, + DefinitionGovernanceRequest, + DefinitionScopeRef, + PolicyDecision, + definition_governance_policy, +) +from govoplan_core.core.registry import PlatformRegistry + + +class _Policy: + def resolve_definition_action(self, *, request): + return PolicyDecision( + allowed=request.definition_kind != "template" + or request.action != "run", + ) + + +class DefinitionGovernanceContractTests(unittest.TestCase): + def test_scope_paths_reuse_policy_provenance_format(self) -> None: + self.assertEqual("system", DefinitionScopeRef("system").path) + self.assertEqual( + "tenant:tenant-1", + DefinitionScopeRef("tenant", "tenant-1").path, + ) + + def test_policy_is_runtime_resolved(self) -> None: + provider = _Policy() + self.assertIsInstance(provider, DefinitionGovernancePolicy) + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="definition_governance_test", + name="Definition governance test", + version="test", + capability_factories={ + CAPABILITY_POLICY_DEFINITION_GOVERNANCE: ( + lambda context: provider + ), + }, + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + + resolved = definition_governance_policy(registry) + self.assertIs(provider, resolved) + decision = resolved.resolve_definition_action( + request=DefinitionGovernanceRequest( + module_id="dataflow", + definition_ref="pipeline:1", + tenant_id="tenant-1", + definition_scope=DefinitionScopeRef( + "tenant", + "tenant-1", + ), + target_scope=DefinitionScopeRef("tenant", "tenant-1"), + definition_kind="template", + action="run", + actor=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + ), + ) + ) + self.assertFalse(decision.allowed) + + +if __name__ == "__main__": + unittest.main()