govoplan-policy/tests/test_definition_policy_overrides.py

from __future__ import annotations

import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.policy import DefinitionGovernanceRequest, DefinitionScopeRef
from govoplan_policy.backend.db.models import PolicyOverride
from govoplan_policy.backend.definition_governance import (
    DEFINITION_POLICY_FIELDS,
    DefinitionGovernancePolicyProvider,
)
from govoplan_policy.backend.definition_policy_service import (
    DefinitionPolicyError,
    definition_policy_response_payload,
    definition_policy_state,
    save_definition_policy,
)


class DefinitionPolicyOverrideTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        PolicyOverride.__table__.create(self.engine)
        self.session_factory = sessionmaker(
            bind=self.engine,
            expire_on_commit=False,
        )
        self.session: Session = self.session_factory()
        self.provider = DefinitionGovernancePolicyProvider()
        self.actor = PrincipalRef(
            account_id="account-1",
            membership_id="user-1",
            tenant_id="tenant-1",
            group_ids=frozenset({"group-1"}),
        )

    def tearDown(self) -> None:
        self.session.close()
        self.engine.dispose()

    def _save(
        self,
        *,
        scope_type: str,
        policy: object,
        scope_id: str | None = None,
        module_id: str = "dataflow",
    ):
        return save_definition_policy(
            self.session,
            module_id=module_id,
            tenant_id="tenant-1",
            scope_type=scope_type,
            scope_id=scope_id,
            policy=policy,
            actor_id="account-1",
        )

    def _resolve(self, *, action: str, definition_scope: DefinitionScopeRef):
        return self.provider.resolve_definition_action(
            self.session,
            request=DefinitionGovernanceRequest(
                module_id="dataflow",
                definition_ref="pipeline:1",
                tenant_id="tenant-1",
                definition_scope=definition_scope,
                target_scope=DefinitionScopeRef("tenant", "tenant-1"),
                definition_kind="flow",
                action=action,  # type: ignore[arg-type]
                actor=self.actor,
                status="active",
                inherit_to_lower_scopes=True,
                allow_run=True,
                allow_reuse=True,
                allow_automation=True,
            ),
        )

    def test_persists_and_revises_an_explicit_policy(self) -> None:
        first = self._save(scope_type="tenant", policy={"allow_run": False})
        second = self._save(
            scope_type="tenant",
            policy={"allow_run": False, "allow_reuse": False},
        )

        self.assertEqual(first.row.id, second.row.id)
        self.assertEqual(2, second.row.revision)
        self.assertEqual(
            {"allow_run": False, "allow_reuse": False},
            second.local_policy,
        )

    def test_system_and_tenant_limits_apply_to_group_and_user_targets(self) -> None:
        self._save(
            scope_type="system",
            module_id="*",
            policy={"allow_automation": False},
        )
        self._save(
            scope_type="tenant",
            policy={"allow_reuse": False},
        )

        group = definition_policy_state(
            self.session,
            module_id="dataflow",
            tenant_id="tenant-1",
            scope_type="group",
            scope_id="group-1",
        )
        user = definition_policy_state(
            self.session,
            module_id="dataflow",
            tenant_id="tenant-1",
            scope_type="user",
            scope_id="user-1",
        )

        for state in (group, user):
            self.assertFalse(state.effective.limits["allow_automation"])
            self.assertFalse(state.effective.limits["allow_reuse"])
            self.assertEqual(
                ["system", "tenant:tenant-1"],
                [step.path for step in state.effective.source_path],
            )

    def test_lower_scopes_cannot_broaden_parent_restrictions(self) -> None:
        self._save(scope_type="tenant", policy={"allow_run": False})

        with self.assertRaisesRegex(
            DefinitionPolicyError,
            "cannot broaden parent restrictions: allow_run",
        ):
            self._save(
                scope_type="group",
                scope_id="group-1",
                policy={"allow_run": True},
            )

    def test_explicit_policy_restricts_provider_actions(self) -> None:
        self._save(
            scope_type="tenant",
            policy={"allow_edit": False, "allow_reuse": False},
        )

        edit = self._resolve(
            action="edit",
            definition_scope=DefinitionScopeRef("tenant", "tenant-1"),
        )
        reuse = self._resolve(
            action="reuse",
            definition_scope=DefinitionScopeRef("group", "group-1"),
        )

        self.assertFalse(edit.allowed)
        self.assertFalse(reuse.allowed)
        self.assertEqual(
            "Editing is disabled by explicit Policy restrictions.",
            edit.reason,
        )
        self.assertEqual("tenant:tenant-1", edit.source_path[0].path)

    def test_malformed_persisted_policy_fails_closed_and_is_redacted(self) -> None:
        row = PolicyOverride(
            policy_family="definition",
            target_key="dataflow",
            tenant_id="tenant-1",
            scope_type="tenant",
            scope_id="tenant-1",
            scope_key="tenant:tenant-1",
            policy={"allow_run": "definitely", "secret": "do-not-echo"},
        )
        self.session.add(row)
        self.session.flush()

        decision = self._resolve(
            action="run",
            definition_scope=DefinitionScopeRef("tenant", "tenant-1"),
        )
        state = definition_policy_state(
            self.session,
            module_id="dataflow",
            tenant_id="tenant-1",
            scope_type="tenant",
        )
        payload = definition_policy_response_payload(state)

        self.assertFalse(decision.allowed)
        self.assertEqual(
            {field: False for field in DEFINITION_POLICY_FIELDS},
            decision.details["effective_limits"],
        )
        self.assertEqual(
            [
                {
                    "code": "definition_policy.invalid",
                    "scope": "tenant:tenant-1",
                    "target_key": "dataflow",
                }
            ],
            decision.details["policy_diagnostics"],
        )
        self.assertEqual(
            {"configuration_status": "invalid_fail_closed"},
            payload["policy"],
        )
        self.assertNotIn("secret", repr(decision.to_dict()))
        self.assertNotIn("do-not-echo", repr(payload))


if __name__ == "__main__":
    unittest.main()