Govern hierarchical View projections
This commit is contained in:
@@ -0,0 +1,227 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user