80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
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()
|