GovOPlaN/govoplan-views
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Enforce Policy ceilings in View resolution

Browse Source
This commit is contained in:
zemion
2026-07-31 17:55:46 +02:00
parent 554705c3c2
commit 303f1556c2
5 changed files with 619 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_views.py
+187 -4
View File
@@ -5,6 +5,10 @@ import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.policy import (
PolicySourceStep,
ViewGovernanceDecision,
)
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_views.backend.service import (
@@ -112,6 +116,72 @@ def lockout_safe_surface_ids() -> list[str]:
]
class StaticViewPolicy:
def __init__(
self,
*,
allowed_view_ids: set[str] | None = None,
visible_surface_ids: set[str] | None = None,
denied_actions: set[str] | None = None,
malformed: bool = False,
) -> None:
self.allowed_view_ids = allowed_view_ids
self.visible_surface_ids = visible_surface_ids
self.denied_actions = denied_actions or set()
self.malformed = malformed
def resolve_view_action(self, session=None, *, request):
del session
diagnostics = (
(
{
"code": "view_policy.invalid",
"severity": "error",
"message": "Malformed View policy.",
},
)
if self.malformed
else ()
)
return ViewGovernanceDecision(
allowed=request.action not in self.denied_actions and not self.malformed,
reason=(
"The requested action is denied by Policy."
if request.action in self.denied_actions or self.malformed
else None
),
allowed_view_ids=(
frozenset(request.candidate_view_ids).intersection(
self.allowed_view_ids
)
if self.allowed_view_ids is not None
else None
),
visible_surface_ids=(
frozenset(request.candidate_surface_ids).intersection(
self.visible_surface_ids
)
if self.visible_surface_ids is not None
else None
),
source_path=(
PolicySourceStep(
scope_type="tenant",
scope_id="tenant-1",
label="Tenant View policy",
applied_fields=("allowed_view_ids", "visible_surface_ids"),
),
),
diagnostics=diagnostics,
details={
"hidden_surface_ids": sorted(
set(request.candidate_surface_ids)
- (self.visible_surface_ids or set(request.candidate_surface_ids))
)
},
)
class ViewsServiceTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
@@ -190,6 +260,122 @@ class ViewsServiceTests(unittest.TestCase):
self.assertIn("views.selector", normalized)
self.assertIn("views.module", normalized)
def test_policy_filters_available_and_default_views_before_selection(self) -> None:
blocked = self.create_published_definition(name="Blocked")
allowed = self.create_published_definition(name="Allowed")
self.assign(
blocked,
scope_type="tenant",
scope_id="tenant-1",
mode="default",
priority=10,
)
self.assign(
allowed,
scope_type="tenant",
scope_id="tenant-1",
mode="default",
)
state = resolve_effective_view(
self.session,
tenant_id="tenant-1",
account_id="account-1",
catalogue=self.catalogue,
governance_policy=StaticViewPolicy(
allowed_view_ids={allowed.id},
),
)
self.assertEqual(allowed.id, state.effective.view_id)
self.assertEqual([allowed.id], [item.id for item in state.available_views])
def test_policy_surface_ceiling_applies_without_a_selected_view(self) -> None:
state = resolve_effective_view(
self.session,
tenant_id="tenant-1",
account_id="account-1",
catalogue=self.catalogue,
governance_policy=StaticViewPolicy(
visible_surface_ids={
"files.module",
"files.nav.files",
"files.route.files",
},
),
)
self.assertIsNone(state.effective.view_id)
self.assertTrue(state.effective.projection_active)
self.assertIn("files.nav.files", state.effective.visible_surface_ids)
self.assertIn("views.selector", state.effective.visible_surface_ids)
self.assertEqual(
"policy_view_ceiling",
state.effective.provenance[-1]["source"],
)
def test_unreachable_or_malformed_policy_falls_back_to_full_interface(self) -> None:
unreachable = resolve_effective_view(
self.session,
tenant_id="tenant-1",
account_id="account-1",
catalogue=self.catalogue,
governance_policy=StaticViewPolicy(
visible_surface_ids={"views.selector"},
),
)
malformed = resolve_effective_view(
self.session,
tenant_id="tenant-1",
account_id="account-1",
catalogue=self.catalogue,
governance_policy=StaticViewPolicy(malformed=True),
)
self.assertFalse(unreachable.effective.projection_active)
self.assertIn(
"view.policy_ceiling_unreachable",
{item.code for item in unreachable.diagnostics},
)
self.assertFalse(malformed.effective.projection_active)
self.assertIn(
"view_policy.invalid",
{item.code for item in malformed.diagnostics},
)
def test_policy_denied_workflow_falls_back_to_normal_default(self) -> None:
default = self.create_published_definition(name="Default")
workflow = self.create_published_definition(name="Workflow")
self.assign(
default,
scope_type="tenant",
scope_id="tenant-1",
mode="default",
)
self.assign(
workflow,
scope_type="tenant",
scope_id="tenant-1",
mode="available",
)
state = resolve_effective_view(
self.session,
tenant_id="tenant-1",
account_id="account-1",
catalogue=self.catalogue,
workflow_view_id=workflow.id,
governance_policy=StaticViewPolicy(
denied_actions={"workflow_activate"},
),
)
self.assertEqual(default.id, state.effective.view_id)
self.assertIn(
"view.workflow_policy_denied",
{item.code for item in state.diagnostics},
)
def test_published_group_view_is_available_to_group_members(self) -> None:
definition = create_definition(
self.session,
@@ -614,10 +800,7 @@ class ViewsServiceTests(unittest.TestCase):
self.assertEqual(default.id, after_membership_removal.effective.view_id)
self.assertIn(
"view.workflow_selection_unavailable",
{
diagnostic.code
for diagnostic in after_membership_removal.diagnostics
},
{diagnostic.code for diagnostic in after_membership_removal.diagnostics},
)
def test_resolution_invalidation_token_tracks_inputs(self) -> None: