govoplan-policy/tests/test_view_governance.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 DefinitionScopeRef, ViewGovernanceRequest
from govoplan_policy.backend.db.models import PolicyOverride
from govoplan_policy.backend.view_governance import ViewGovernancePolicyProvider
from govoplan_policy.backend.view_policy_service import (
    ViewPolicyError,
    save_view_policy,
    view_policy_response_payload,
    view_policy_state,
)


class ViewGovernanceTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        PolicyOverride.__table__.create(self.engine)
        self.session: Session = sessionmaker(
            bind=self.engine,
            expire_on_commit=False,
        )()
        self.provider = ViewGovernancePolicyProvider()
        self.actor = PrincipalRef(
            account_id="account-1",
            membership_id="membership-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,
    ):
        return save_view_policy(
            self.session,
            tenant_id="tenant-1",
            scope_type=scope_type,
            scope_id=scope_id,
            policy=policy,
            actor_id="account-1",
        )

    def _resolve(
        self,
        *,
        action: str = "view",
        view_id: str | None = None,
        requested_surface_ids: tuple[str, ...] = (),
    ):
        return self.provider.resolve_view_action(
            self.session,
            request=ViewGovernanceRequest(
                tenant_id="tenant-1",
                action=action,  # type: ignore[arg-type]
                actor=self.actor,
                target_scope=DefinitionScopeRef("user", "account-1"),
                view_id=view_id,
                candidate_view_ids=("view-1", "view-2", "view-3"),
                candidate_surface_ids=("surface.a", "surface.b", "surface.c"),
                requested_surface_ids=requested_surface_ids,
            ),
        )

    def test_hierarchy_intersects_view_and_surface_ceilings(self) -> None:
        self._save(
            "system",
            {
                "allowed_view_ids": ["view-1", "view-2"],
                "visible_surface_ids": ["surface.a", "surface.b", "surface.c"],
            },
        )
        self._save(
            "tenant",
            {
                "allowed_view_ids": ["view-2"],
                "visible_surface_ids": ["surface.a", "surface.b"],
            },
        )

        decision = self._resolve(
            action="workflow_activate",
            view_id="view-2",
            requested_surface_ids=("surface.b", "surface.c"),
        )

        self.assertTrue(decision.allowed)
        self.assertEqual(frozenset({"view-2"}), decision.allowed_view_ids)
        self.assertEqual(
            frozenset({"surface.a", "surface.b"}),
            decision.visible_surface_ids,
        )
        self.assertEqual(
            ["system", "tenant:tenant-1"],
            [step.path for step in decision.source_path],
        )
        self.assertEqual(
            ["surface.c"],
            decision.details["requested_surfaces_outside_ceiling"],
        )

    def test_lower_scope_cannot_broaden_boolean_or_set_ceiling(self) -> None:
        self._save(
            "tenant",
            {
                "allow_assign": False,
                "allowed_view_ids": ["view-1"],
            },
        )

        with self.assertRaisesRegex(
            ViewPolicyError,
            "allow_assign, allowed_view_ids",
        ):
            self._save(
                "group",
                {
                    "allow_assign": True,
                    "allowed_view_ids": ["view-1", "view-2"],
                },
                "group-1",
            )

    def test_action_and_requested_view_are_bounded_independently(self) -> None:
        self._save(
            "tenant",
            {
                "allow_assign": False,
                "allowed_view_ids": ["view-1"],
            },
        )

        assignment = self._resolve(action="assign", view_id="view-1")
        selection = self._resolve(action="select", view_id="view-2")

        self.assertFalse(assignment.allowed)
        self.assertIn("disabled by Policy", assignment.reason or "")
        self.assertFalse(selection.allowed)
        self.assertIn("outside the effective Policy ceiling", selection.reason or "")

    def test_edit_cannot_store_surfaces_outside_the_policy_ceiling(self) -> None:
        self._save(
            "tenant",
            {"visible_surface_ids": ["surface.a", "surface.b"]},
        )

        decision = self._resolve(
            action="edit",
            requested_surface_ids=("surface.a", "surface.c"),
        )

        self.assertFalse(decision.allowed)
        self.assertIn("surfaces are outside", decision.reason or "")
        self.assertEqual(
            [{"id": "surface.c", "sources": ["tenant:tenant-1"]}],
            decision.details["surface_provenance"],
        )

    def test_assignment_uses_the_target_group_policy(self) -> None:
        self._save(
            "group",
            {"allow_assign": False},
            "group-2",
        )

        decision = self.provider.resolve_view_action(
            self.session,
            request=ViewGovernanceRequest(
                tenant_id="tenant-1",
                action="assign",
                actor=self.actor,
                target_scope=DefinitionScopeRef("group", "group-2"),
                view_id="view-1",
                candidate_view_ids=("view-1",),
            ),
        )

        self.assertFalse(decision.allowed)
        self.assertEqual("group:group-2", decision.source_path[-1].path)

    def test_malformed_policy_fails_closed_without_echoing_record(self) -> None:
        self.session.add(
            PolicyOverride(
                policy_family="view",
                target_key="*",
                tenant_id="tenant-1",
                scope_type="tenant",
                scope_id="tenant-1",
                scope_key="tenant:tenant-1",
                policy={"visible_surface_ids": "secret-value"},
            )
        )
        self.session.flush()

        decision = self._resolve(action="select", view_id="view-1")
        payload = view_policy_response_payload(
            view_policy_state(
                self.session,
                tenant_id="tenant-1",
                scope_type="tenant",
            )
        )

        self.assertFalse(decision.allowed)
        self.assertEqual(frozenset(), decision.allowed_view_ids)
        self.assertEqual(frozenset(), decision.visible_surface_ids)
        self.assertEqual(
            {"configuration_status": "invalid_fail_closed"},
            payload["policy"],
        )
        self.assertNotIn("secret-value", repr(decision.to_dict()))
        self.assertNotIn("secret-value", repr(payload))


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