Add governed automation contracts

2026-07-28 15:02:42 +02:00
parent baa2143a26
commit 26ae034153
10 changed files with 539 additions and 0 deletions
--- a/tests/test_definition_governance_contract.py
+++ 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()