271 lines
8.5 KiB
Python
271 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.policy import PolicyDecision
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_workflow.backend.db.models import (
|
|
WorkflowDefinition,
|
|
WorkflowDefinitionRevision,
|
|
)
|
|
from govoplan_workflow.backend.governance import normalize_definition_scope
|
|
from govoplan_workflow.backend.schemas import (
|
|
WorkflowDefinitionCreateRequest,
|
|
WorkflowDefinitionDeriveRequest,
|
|
WorkflowDefinitionUpdateRequest,
|
|
)
|
|
from govoplan_workflow.backend.service import (
|
|
WorkflowConflictError,
|
|
activate_definition,
|
|
create_definition,
|
|
definition_response,
|
|
derive_definition,
|
|
update_definition,
|
|
)
|
|
from test_service import sample_graph
|
|
|
|
|
|
POLICY_CAPABILITY = "policy.definitionGovernance"
|
|
|
|
|
|
def principal() -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(
|
|
{
|
|
"workflow:definition:read",
|
|
"workflow:definition:write",
|
|
"workflow:instance:start",
|
|
}
|
|
),
|
|
),
|
|
account=object(),
|
|
user=object(),
|
|
)
|
|
|
|
|
|
class DefinitionPolicy:
|
|
def resolve_definition_action(self, *, request):
|
|
local = request.definition_scope.scope_type == "tenant"
|
|
inherited = (
|
|
request.definition_scope.scope_type == "system"
|
|
and request.inherit_to_lower_scopes
|
|
)
|
|
allowed = local or inherited
|
|
if request.action == "edit":
|
|
allowed = local
|
|
elif request.action in {"reuse", "derive"}:
|
|
allowed = allowed and request.allow_reuse
|
|
elif request.action == "run":
|
|
allowed = (
|
|
allowed
|
|
and request.definition_kind == "flow"
|
|
and request.status == "active"
|
|
and request.allow_run
|
|
)
|
|
elif request.action == "automate":
|
|
allowed = (
|
|
allowed
|
|
and request.definition_kind == "flow"
|
|
and request.allow_automation
|
|
)
|
|
return PolicyDecision(
|
|
allowed=allowed,
|
|
reason=None if allowed else "Definition action denied.",
|
|
)
|
|
|
|
|
|
class Registry:
|
|
def has_capability(self, name: str) -> bool:
|
|
return name == POLICY_CAPABILITY
|
|
|
|
def capability(self, name: str):
|
|
if not self.has_capability(name):
|
|
raise KeyError(name)
|
|
return DefinitionPolicy()
|
|
|
|
|
|
class WorkflowGovernanceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[
|
|
WorkflowDefinition.__table__,
|
|
WorkflowDefinitionRevision.__table__,
|
|
],
|
|
)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.session: Session = self.Session()
|
|
self.principal = principal()
|
|
self.registry = Registry()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
Base.metadata.drop_all(
|
|
self.engine,
|
|
tables=[
|
|
WorkflowDefinitionRevision.__table__,
|
|
WorkflowDefinition.__table__,
|
|
],
|
|
)
|
|
self.engine.dispose()
|
|
|
|
def test_derivation_pins_template_revision_and_provenance(self) -> None:
|
|
template = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="system-admin",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
key="permit-review",
|
|
name="Permit review template",
|
|
graph=sample_graph(),
|
|
scope_type="system",
|
|
definition_kind="template",
|
|
inherit_to_lower_scopes=True,
|
|
allow_reuse=True,
|
|
),
|
|
)
|
|
derived = derive_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.principal,
|
|
registry=self.registry,
|
|
source_definition_id=template.id,
|
|
payload=WorkflowDefinitionDeriveRequest(
|
|
name="Tenant permit review",
|
|
allow_start=True,
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
response = definition_response(
|
|
self.session,
|
|
derived,
|
|
principal=self.principal,
|
|
registry=self.registry,
|
|
)
|
|
self.assertEqual(template.id, derived.derived_from_definition_id)
|
|
self.assertEqual(1, derived.derived_from_revision)
|
|
self.assertEqual(
|
|
template.revisions[0].content_hash,
|
|
derived.derived_from_hash,
|
|
)
|
|
self.assertEqual(
|
|
"system",
|
|
response.governance.derivation_provenance["source_scope"][
|
|
"scope_type"
|
|
],
|
|
)
|
|
self.assertFalse(response.governance.automation_runtime_available)
|
|
|
|
def test_user_scope_normalizes_membership_to_account_id(self) -> None:
|
|
tenant_id, scope_type, scope_id, scope_key = normalize_definition_scope(
|
|
self.principal,
|
|
scope_type="user",
|
|
scope_id="membership-1",
|
|
administrative=False,
|
|
)
|
|
|
|
self.assertEqual("tenant-1", tenant_id)
|
|
self.assertEqual("user", scope_type)
|
|
self.assertEqual("account-1", scope_id)
|
|
self.assertEqual("user:account-1", scope_key)
|
|
|
|
def test_template_cannot_be_activated(self) -> None:
|
|
template = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Reusable review",
|
|
graph=sample_graph(),
|
|
definition_kind="template",
|
|
allow_reuse=True,
|
|
),
|
|
)
|
|
|
|
with self.assertRaises(WorkflowConflictError):
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=template.id,
|
|
actor_id="account-1",
|
|
)
|
|
|
|
def test_derived_limits_cannot_be_broadened_transitively(self) -> None:
|
|
template = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Restricted review",
|
|
graph=sample_graph(),
|
|
definition_kind="template",
|
|
allow_reuse=True,
|
|
allow_automation=False,
|
|
inherit_to_lower_scopes=False,
|
|
),
|
|
)
|
|
derived = derive_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.principal,
|
|
registry=self.registry,
|
|
source_definition_id=template.id,
|
|
payload=WorkflowDefinitionDeriveRequest(
|
|
name="Tenant review",
|
|
allow_reuse=True,
|
|
allow_automation=True,
|
|
inherit_to_lower_scopes=True,
|
|
),
|
|
)
|
|
update_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=derived.id,
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionUpdateRequest(
|
|
name=derived.name,
|
|
graph=sample_graph(),
|
|
expected_revision=1,
|
|
allow_reuse=True,
|
|
allow_automation=True,
|
|
inherit_to_lower_scopes=True,
|
|
),
|
|
)
|
|
grandchild = derive_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.principal,
|
|
registry=self.registry,
|
|
source_definition_id=derived.id,
|
|
payload=WorkflowDefinitionDeriveRequest(
|
|
name="User review",
|
|
scope_type="user",
|
|
scope_id="membership-1",
|
|
allow_automation=True,
|
|
inherit_to_lower_scopes=True,
|
|
),
|
|
)
|
|
|
|
self.assertFalse(derived.allow_automation)
|
|
self.assertFalse(derived.inherit_to_lower_scopes)
|
|
self.assertFalse(grandchild.allow_automation)
|
|
self.assertFalse(grandchild.inherit_to_lower_scopes)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|