From 303f1556c208b053222209bd1189ac02a1189e8c Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 31 Jul 2026 17:55:46 +0200 Subject: [PATCH] Enforce Policy ceilings in View resolution --- src/govoplan_views/backend/capabilities.py | 2 + src/govoplan_views/backend/router.py | 117 +++++++ src/govoplan_views/backend/schemas.py | 3 + src/govoplan_views/backend/service.py | 344 +++++++++++++++++++-- tests/test_views.py | 191 +++++++++++- 5 files changed, 619 insertions(+), 38 deletions(-) diff --git a/src/govoplan_views/backend/capabilities.py b/src/govoplan_views/backend/capabilities.py index 26d1dcf..7bddfb4 100644 --- a/src/govoplan_views/backend/capabilities.py +++ b/src/govoplan_views/backend/capabilities.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Iterable from govoplan_core.core.modules import ModuleContext +from govoplan_core.core.policy import view_governance_policy from govoplan_core.core.views import EffectiveView, ViewResolver from govoplan_views.backend.service import resolve_effective_view @@ -31,6 +32,7 @@ class ViewsResolverCapability(ViewResolver): workflow_view_id=workflow_view_id, workflow_revision_id=workflow_revision_id, workflow_surface_ids=workflow_surface_ids, + governance_policy=view_governance_policy(self._registry), ) return state.effective diff --git a/src/govoplan_views/backend/router.py b/src/govoplan_views/backend/router.py index 973df79..3736aea 100644 --- a/src/govoplan_views/backend/router.py +++ b/src/govoplan_views/backend/router.py @@ -6,6 +6,11 @@ from sqlalchemy.orm import Session from govoplan_core.audit.logging import audit_event from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope from govoplan_core.core.access import CAPABILITY_ACCESS_DIRECTORY, AccessDirectory +from govoplan_core.core.policy import ( + DefinitionScopeRef, + ViewGovernanceRequest, + view_governance_policy, +) from govoplan_core.core.views import VIEW_SURFACE_CONTRACT_VERSION, ViewSurface from govoplan_core.db.session import get_session from govoplan_views.backend.manifest import ( @@ -81,6 +86,49 @@ def _catalogue() -> tuple[ViewSurface, ...]: return get_registry().view_surfaces() +def _view_governance_policy(): + return view_governance_policy(get_registry()) + + +def _enforce_view_policy_action( + session: Session, + principal: ApiPrincipal, + *, + action: str, + scope_type: str, + scope_id: str | None, + view_id: str | None = None, + surface_ids: tuple[str, ...] = (), +) -> None: + policy = _view_governance_policy() + if policy is None: + return + target_id = scope_id + if scope_type == "tenant": + target_id = target_id or principal.tenant_id + decision = policy.resolve_view_action( + session, + request=ViewGovernanceRequest( + tenant_id=principal.tenant_id, + action=action, # type: ignore[arg-type] + actor=principal.principal, + target_scope=DefinitionScopeRef( + scope_type, # type: ignore[arg-type] + target_id, + ), + view_id=view_id, + candidate_view_ids=(view_id,) if view_id is not None else (), + candidate_surface_ids=tuple(surface.id for surface in _catalogue()), + requested_surface_ids=surface_ids, + ), + ) + if not decision.allowed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=decision.reason or f"Policy denied View action: {action}", + ) + + def _optional_access_directory() -> AccessDirectory | None: registry = get_registry() if not registry.has_capability(CAPABILITY_ACCESS_DIRECTORY): @@ -432,6 +480,7 @@ def _effective_response(state: EffectiveViewState) -> EffectiveViewResponse: active_revision_id=effective.revision_id, active_view_name=effective.name, visible_surface_ids=sorted(effective.visible_surface_ids), + projection_active=effective.projection_active, locked=effective.locked, available_views=[ EffectiveViewOptionResponse( @@ -490,6 +539,7 @@ def api_effective_view( account_id=principal.account_id, group_ids=principal.group_ids, catalogue=_catalogue(), + governance_policy=_view_governance_policy(), ) ) @@ -515,6 +565,7 @@ def api_workflow_effective_view( workflow_view_id=payload.view_id, workflow_revision_id=payload.revision_id, workflow_surface_ids=payload.visible_surface_ids, + governance_policy=_view_governance_policy(), ) ) @@ -534,6 +585,7 @@ def api_select_view( group_ids=principal.group_ids, view_id=payload.view_id, catalogue=_catalogue(), + governance_policy=_view_governance_policy(), ) _audit( session, @@ -651,6 +703,14 @@ def api_create_definition( payload.scope_id, ) _require_definition_write(principal, payload.scope_type, target_id) + _enforce_view_policy_action( + session, + principal, + action="derive", + scope_type=payload.scope_type, + scope_id=target_id, + surface_ids=tuple(payload.visible_surface_ids), + ) try: definition = create_definition( session, @@ -700,6 +760,14 @@ def api_update_definition( definition.scope_type, definition.scope_id, ) + _enforce_view_policy_action( + session, + principal, + action="edit", + scope_type=definition.scope_type, + scope_id=definition.scope_id, + view_id=definition.id, + ) update_definition( session, definition, @@ -775,6 +843,15 @@ def api_create_revision( definition.scope_type, definition.scope_id, ) + _enforce_view_policy_action( + session, + principal, + action="edit", + scope_type=definition.scope_type, + scope_id=definition.scope_id, + view_id=definition.id, + surface_ids=tuple(payload.visible_surface_ids), + ) revision = create_revision( session, definition, @@ -821,6 +898,14 @@ def api_publish_revision( definition.scope_type, definition.scope_id, ) + _enforce_view_policy_action( + session, + principal, + action="edit", + scope_type=definition.scope_type, + scope_id=definition.scope_id, + view_id=definition.id, + ) revision = get_revision( session, definition_id=definition.id, @@ -868,6 +953,14 @@ def api_archive_definition( definition.scope_type, definition.scope_id, ) + _enforce_view_policy_action( + session, + principal, + action="edit", + scope_type=definition.scope_type, + scope_id=definition.scope_id, + view_id=definition.id, + ) archive_definition( session, definition, @@ -977,6 +1070,14 @@ def api_create_assignment( tenant_id=principal.tenant_id, definition_id=payload.definition_id, ) + _enforce_view_policy_action( + session, + principal, + action="assign", + scope_type=payload.scope_type, + scope_id=payload.scope_id, + view_id=definition.id, + ) assignment = create_assignment( session, tenant_id=principal.tenant_id, @@ -1033,6 +1134,14 @@ def api_update_assignment( assignment_id=assignment_id, ) _require_assignment_write(principal, assignment.scope_type) + _enforce_view_policy_action( + session, + principal, + action="assign", + scope_type=assignment.scope_type, + scope_id=assignment.scope_id, + view_id=assignment.definition_id, + ) update_assignment( session, assignment, @@ -1076,6 +1185,14 @@ def api_delete_assignment( assignment_id=assignment_id, ) _require_assignment_write(principal, assignment.scope_type) + _enforce_view_policy_action( + session, + principal, + action="assign", + scope_type=assignment.scope_type, + scope_id=assignment.scope_id, + view_id=assignment.definition_id, + ) assignment_details = { "scope_type": assignment.scope_type, "scope_id": assignment.scope_id, diff --git a/src/govoplan_views/backend/schemas.py b/src/govoplan_views/backend/schemas.py index b2d2b21..b430570 100644 --- a/src/govoplan_views/backend/schemas.py +++ b/src/govoplan_views/backend/schemas.py @@ -163,6 +163,8 @@ class ViewProvenanceResponse(BaseModel): scope_type: str | None = None scope_id: str | None = None detail: str | None = None + surface_ids: list[str] = Field(default_factory=list) + applied_fields: list[str] = Field(default_factory=list) class ViewDiagnosticResponse(BaseModel): @@ -177,6 +179,7 @@ class EffectiveViewResponse(BaseModel): active_revision_id: str | None = None active_view_name: str | None = None visible_surface_ids: list[str] = Field(default_factory=list) + projection_active: bool = False locked: bool = False available_views: list[EffectiveViewOptionResponse] = Field(default_factory=list) provenance: list[ViewProvenanceResponse] = Field(default_factory=list) diff --git a/src/govoplan_views/backend/service.py b/src/govoplan_views/backend/service.py index 9cf3d73..8a2b2ae 100644 --- a/src/govoplan_views/backend/service.py +++ b/src/govoplan_views/backend/service.py @@ -11,6 +11,13 @@ from typing import Literal from sqlalchemy import and_, or_ from sqlalchemy.orm import Session, joinedload +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.policy import ( + DefinitionScopeRef, + ViewGovernanceDecision, + ViewGovernancePolicy, + ViewGovernanceRequest, +) from govoplan_core.core.views import ( VIEW_SURFACE_CONTRACT_VERSION, EffectiveView, @@ -968,15 +975,12 @@ def _resolve_assignment_revisions( for assignment in candidates if ( revision_id := ( - assignment.revision_id - or assignment.definition.published_revision_id + assignment.revision_id or assignment.definition.published_revision_id ) ) } revisions = ( - session.query(ViewRevision) - .filter(ViewRevision.id.in_(revision_ids)) - .all() + session.query(ViewRevision).filter(ViewRevision.id.in_(revision_ids)).all() if revision_ids else () ) @@ -984,14 +988,10 @@ def _resolve_assignment_revisions( resolved: list[ResolvedAssignment] = [] for assignment in candidates: revision_id = ( - assignment.revision_id - or assignment.definition.published_revision_id + assignment.revision_id or assignment.definition.published_revision_id ) revision = by_id.get(revision_id) - if ( - revision is not None - and revision.definition_id == assignment.definition_id - ): + if revision is not None and revision.definition_id == assignment.definition_id: resolved.append((assignment, revision)) return tuple(resolved) @@ -1055,9 +1055,7 @@ def _select_effective_assignment( initial_diagnostics: Iterable[ViewDiagnostic] = (), fallback: ViewSelection | None = None, ) -> ViewSelection: - required = [ - pair for pair in assignments if pair[0].mode == "required" - ] + required = [pair for pair in assignments if pair[0].mode == "required"] required_selection = ( max( required, @@ -1102,8 +1100,7 @@ def _select_effective_assignment( fallback=fallback, ) if not any( - item.code == "view.workflow_revision_unavailable" - for item in diagnostics + item.code == "view.workflow_revision_unavailable" for item in diagnostics ): diagnostics.append( ViewDiagnostic( @@ -1164,9 +1161,7 @@ def _select_effective_assignment( diagnostics=tuple(diagnostics), ) - defaults = [ - pair for pair in assignments if pair[0].mode == "default" - ] + defaults = [pair for pair in assignments if pair[0].mode == "default"] if defaults: selected = max( defaults, @@ -1202,9 +1197,7 @@ def _evaluate_selected_surfaces( return selection assignment, revision = selection.selected known_surfaces = {surface.id: surface for surface in catalogue} - stale_ids = sorted( - set(revision.visible_surface_ids) - set(known_surfaces) - ) + stale_ids = sorted(set(revision.visible_surface_ids) - set(known_surfaces)) diagnostics = list(selection.diagnostics) provenance = list(selection.provenance) visible_surface_ids = set(revision.visible_surface_ids) @@ -1219,9 +1212,7 @@ def _evaluate_selected_surfaces( unknown_overlay_ids = sorted(requested_overlay - set(known_surfaces)) bounded_overlay_ids = requested_overlay & visible_surface_ids unavailable_overlay_ids = sorted( - requested_overlay - - set(unknown_overlay_ids) - - bounded_overlay_ids + requested_overlay - set(unknown_overlay_ids) - bounded_overlay_ids ) protected_surface_ids = { surface.id @@ -1345,9 +1336,8 @@ def _evaluate_selected_surfaces( for surface_id in visible_surface_ids if surface_id in known_surfaces ] - if ( - not any(surface.kind == "navigation" for surface in known_visible) - or not any(surface.kind == "route" for surface in known_visible) + if not any(surface.kind == "navigation" for surface in known_visible) or not any( + surface.kind == "route" for surface in known_visible ): diagnostics.append( ViewDiagnostic( @@ -1410,9 +1400,7 @@ def _required_lockout_diagnostic( catalogue, assignment_scope_type=assignment.scope_type, ) - missing_ids = tuple( - sorted(required_ids - set(revision.visible_surface_ids)) - ) + missing_ids = tuple(sorted(required_ids - set(revision.visible_surface_ids))) message = ( "The required View no longer retains all administration escape surfaces." if missing_ids @@ -1452,8 +1440,9 @@ def _effective_view_from_selection(selection: ViewSelection) -> EffectiveView: view_id=None, revision_id=None, name=None, - visible_surface_ids=frozenset(), + visible_surface_ids=selection.visible_surface_ids or frozenset(), locked=False, + projection_active=selection.visible_surface_ids is not None, provenance=selection.provenance, ) assignment, revision = selection.selected @@ -1467,10 +1456,198 @@ def _effective_view_from_selection(selection: ViewSelection) -> EffectiveView: else frozenset(revision.visible_surface_ids) ), locked=selection.locked, + projection_active=True, provenance=selection.provenance, ) +def _policy_diagnostics( + decision: ViewGovernanceDecision, +) -> tuple[ViewDiagnostic, ...]: + result: list[ViewDiagnostic] = [] + for item in decision.diagnostics: + severity = str(item.get("severity") or "warning") + result.append( + ViewDiagnostic( + severity="error" if severity == "error" else "warning", + code=str(item.get("code") or "view.policy_diagnostic"), + message=str( + item.get("message") + or "View Policy changed the effective interface projection." + ), + surface_ids=tuple( + str(value) for value in item.get("surface_ids", ()) if str(value) + ), + ) + ) + return tuple(result) + + +def _policy_provenance( + decision: ViewGovernanceDecision, +) -> tuple[dict[str, object], ...]: + surface_sources: dict[str, set[str]] = {} + for item in decision.details.get("surface_provenance", ()): + if not isinstance(item, dict) or not item.get("id"): + continue + surface_id = str(item["id"]) + for source in item.get("sources", ()): + surface_sources.setdefault(str(source), set()).add(surface_id) + return tuple( + { + "source": "policy_view_ceiling", + "scope_type": step.scope_type, + "scope_id": step.scope_id, + "detail": step.label, + "surface_ids": tuple(sorted(surface_sources.get(step.path, set()))), + "applied_fields": step.applied_fields, + } + for step in decision.source_path + ) + + +def _apply_policy_surface_ceiling( + selection: ViewSelection, + *, + decision: ViewGovernanceDecision | None, + catalogue: tuple[ViewSurface, ...] | None, +) -> ViewSelection: + if decision is None: + return selection + diagnostics = (*selection.diagnostics, *_policy_diagnostics(decision)) + provenance = (*selection.provenance, *_policy_provenance(decision)) + if any(item.code == "view_policy.invalid" for item in diagnostics): + return ViewSelection( + selected=None, + locked=False, + provenance=( + *provenance, + { + "source": "invalid_view_fallback", + "scope_type": "user", + "scope_id": None, + "detail": "Malformed View Policy record", + }, + ), + diagnostics=diagnostics, + ) + if decision.visible_surface_ids is None or catalogue is None: + return ViewSelection( + selected=selection.selected, + locked=selection.locked, + provenance=provenance, + diagnostics=diagnostics, + visible_surface_ids=selection.visible_surface_ids, + surface_ceiling_ids=selection.surface_ceiling_ids, + fallback=selection.fallback, + ) + + by_id = {surface.id: surface for surface in catalogue} + ceiling = set(decision.visible_surface_ids).intersection(by_id) + protected = {surface.id for surface in catalogue if surface.required} + if selection.locked and selection.selected is not None: + try: + protected.update( + lockout_required_surface_ids( + catalogue, + assignment_scope_type=selection.selected[0].scope_type, + ) + ) + except ViewsValidationError: + pass + ceiling.update(protected) + pending = list(ceiling) + while pending: + surface = by_id.get(pending.pop()) + if surface is None or not surface.parent_id: + continue + if surface.parent_id not in ceiling and surface.parent_id in by_id: + ceiling.add(surface.parent_id) + pending.append(surface.parent_id) + + selected_ids = ( + set(selection.visible_surface_ids) + if selection.visible_surface_ids is not None + else ( + set(selection.selected[1].visible_surface_ids) + if selection.selected is not None + else set(by_id) + ) + ) + bounded = selected_ids.intersection(ceiling) + bounded.update(protected.intersection(selected_ids)) + known_visible = [by_id[surface_id] for surface_id in bounded if surface_id in by_id] + if not any(surface.kind == "navigation" for surface in known_visible) or not any( + surface.kind == "route" for surface in known_visible + ): + diagnostic = ViewDiagnostic( + severity="error", + code="view.policy_ceiling_unreachable", + message=( + "The effective View Policy contains no reachable navigation and " + "route. The normal authorized interface is shown." + ), + ) + return ViewSelection( + selected=None, + locked=False, + provenance=( + *provenance, + { + "source": "invalid_view_fallback", + "scope_type": "user", + "scope_id": None, + "detail": "View Policy ceiling was not reachable", + }, + ), + diagnostics=(*diagnostics, diagnostic), + ) + return ViewSelection( + selected=selection.selected, + locked=selection.locked, + provenance=provenance, + diagnostics=diagnostics, + visible_surface_ids=frozenset(bounded), + surface_ceiling_ids=selection.surface_ceiling_ids, + fallback=selection.fallback, + ) + + +def _resolve_policy_decision( + policy: ViewGovernancePolicy | None, + session: Session, + *, + tenant_id: str, + account_id: str, + group_ids: frozenset[str], + action: str, + view_id: str | None, + candidate_view_ids: Iterable[str], + catalogue: tuple[ViewSurface, ...] | None, + requested_surface_ids: Iterable[str] = (), +) -> ViewGovernanceDecision | None: + if policy is None: + return None + return policy.resolve_view_action( + session, + request=ViewGovernanceRequest( + tenant_id=tenant_id, + action=action, # type: ignore[arg-type] + actor=PrincipalRef( + account_id=account_id, + membership_id=None, + tenant_id=tenant_id, + group_ids=group_ids, + ), + target_scope=DefinitionScopeRef("user", account_id), + view_id=view_id, + candidate_view_ids=tuple(candidate_view_ids), + candidate_surface_ids=tuple(surface.id for surface in catalogue or ()), + requested_surface_ids=tuple(requested_surface_ids), + ), + ) + + def _resolution_invalidation_token( assignments: Iterable[ResolvedAssignment], *, @@ -1479,6 +1656,7 @@ def _resolution_invalidation_token( workflow_view_id: str | None, workflow_revision_id: str | None, workflow_surface_ids: Iterable[str], + policy_decisions: Iterable[ViewGovernanceDecision] = (), ) -> str: payload = { "assignments": [ @@ -1522,6 +1700,7 @@ def _resolution_invalidation_token( "workflow_surface_ids": sorted( str(surface_id) for surface_id in workflow_surface_ids ), + "policy_decisions": [decision.to_dict() for decision in policy_decisions], } encoded = json.dumps( payload, @@ -1541,6 +1720,7 @@ def resolve_effective_view( workflow_view_id: str | None = None, workflow_revision_id: str | None = None, workflow_surface_ids: Iterable[str] = (), + governance_policy: ViewGovernancePolicy | None = None, ) -> EffectiveViewState: groups = frozenset(str(group_id) for group_id in group_ids) workflow_surfaces = tuple( @@ -1594,8 +1774,75 @@ def resolve_effective_view( candidates, ) options = _option_assignments(resolved_assignments) + policy_decision = _resolve_policy_decision( + governance_policy, + session, + tenant_id=tenant_id, + account_id=account_id, + group_ids=groups, + action="view", + view_id=None, + candidate_view_ids=options, + catalogue=current_catalogue, + ) + policy_diagnostics = ( + _policy_diagnostics(policy_decision) if policy_decision is not None else () + ) + policy_invalid = any( + item.code == "view_policy.invalid" for item in policy_diagnostics + ) + if policy_decision is not None and not policy_invalid: + allowed_ids = ( + policy_decision.allowed_view_ids if policy_decision.allowed else frozenset() + ) + if allowed_ids is not None: + resolved_assignments = tuple( + pair + for pair in resolved_assignments + if pair[0].definition_id in allowed_ids + ) + options = _option_assignments(resolved_assignments) workflow_options = dict(options) workflow_diagnostics: tuple[ViewDiagnostic, ...] = () + workflow_policy_decision = ( + _resolve_policy_decision( + governance_policy, + session, + tenant_id=tenant_id, + account_id=account_id, + group_ids=groups, + action="workflow_activate", + view_id=workflow_view_id, + candidate_view_ids=options, + catalogue=current_catalogue, + requested_surface_ids=workflow_surfaces, + ) + if workflow_view_id is not None + else None + ) + if workflow_policy_decision is not None: + workflow_diagnostics = _policy_diagnostics(workflow_policy_decision) + if ( + workflow_policy_decision is not None + and not workflow_policy_decision.allowed + and not any( + str(item.get("code")) == "view_policy.invalid" + for item in workflow_policy_decision.diagnostics + ) + ): + workflow_diagnostics = ( + *workflow_diagnostics, + ViewDiagnostic( + severity="warning", + code="view.workflow_policy_denied", + message=( + workflow_policy_decision.reason + or "Policy did not allow the workflow-selected View." + ), + ), + ) + workflow_view_id = None + workflow_revision_id = None if workflow_view_id is not None and workflow_revision_id is not None: option = options.get(workflow_view_id) pinned_revision = ( @@ -1655,6 +1902,11 @@ def resolve_effective_view( catalogue=current_catalogue, workflow_surface_ids=workflow_surfaces, ) + evaluated = _apply_policy_surface_ceiling( + evaluated, + decision=policy_decision, + catalogue=current_catalogue, + ) return EffectiveViewState( effective=_effective_view_from_selection(evaluated), available_views=_available_view_options(options), @@ -1666,6 +1918,11 @@ def resolve_effective_view( workflow_view_id=workflow_view_id, workflow_revision_id=workflow_revision_id, workflow_surface_ids=workflow_surfaces, + policy_decisions=tuple( + decision + for decision in (policy_decision, workflow_policy_decision) + if decision is not None + ), ), ) @@ -1678,13 +1935,16 @@ def select_view( group_ids: Iterable[str], view_id: str | None, catalogue: Iterable[ViewSurface] | None = None, + governance_policy: ViewGovernancePolicy | None = None, ) -> EffectiveViewState: + current_catalogue = tuple(catalogue) if catalogue is not None else None current = resolve_effective_view( session, tenant_id=tenant_id, account_id=account_id, group_ids=group_ids, - catalogue=catalogue, + catalogue=current_catalogue, + governance_policy=governance_policy, ) if current.effective.locked and view_id != current.effective.view_id: raise ViewsConflictError( @@ -1693,6 +1953,21 @@ def select_view( available_ids = {option.id for option in current.available_views} if view_id is not None and view_id not in available_ids: raise ViewsValidationError("The selected View is not available to this account") + selection_policy = _resolve_policy_decision( + governance_policy, + session, + tenant_id=tenant_id, + account_id=account_id, + group_ids=frozenset(str(group_id) for group_id in group_ids), + action="select", + view_id=view_id, + candidate_view_ids=available_ids, + catalogue=current_catalogue, + ) + if selection_policy is not None and not selection_policy.allowed: + raise ViewsValidationError( + selection_policy.reason or "Policy does not allow View selection" + ) preference = _preference( session, @@ -1714,7 +1989,8 @@ def select_view( tenant_id=tenant_id, account_id=account_id, group_ids=group_ids, - catalogue=catalogue, + catalogue=current_catalogue, + governance_policy=governance_policy, ) diff --git a/tests/test_views.py b/tests/test_views.py index 540b6af..be48a94 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -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: