govoplan-workflow/tests/test_governance.py

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.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_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()