diff --git a/src/govoplan_views/backend/service.py b/src/govoplan_views/backend/service.py index 4ce953f..1c4645d 100644 --- a/src/govoplan_views/backend/service.py +++ b/src/govoplan_views/backend/service.py @@ -942,7 +942,8 @@ def resolve_effective_view( diagnostics: list[ViewDiagnostic] = [] if selected is not None and catalogue is not None: assignment, revision = selected - known_surfaces = {surface.id: surface for surface in catalogue} + current_catalogue = tuple(catalogue) + known_surfaces = {surface.id: surface for surface in current_catalogue} stale_ids = sorted(set(revision.visible_surface_ids) - set(known_surfaces)) if stale_ids: diagnostics.append( @@ -956,14 +957,58 @@ def resolve_effective_view( surface_ids=tuple(stale_ids), ) ) + if locked: + try: + lockout_ids = lockout_required_surface_ids( + current_catalogue, + assignment_scope_type=assignment.scope_type, + ) + unavailable_escape_ids = sorted( + lockout_ids - set(revision.visible_surface_ids) + ) + lockout_message = ( + "The required View no longer retains all administration " + "escape surfaces." + if unavailable_escape_ids + else "" + ) + except ViewsValidationError as exc: + unavailable_escape_ids = () + lockout_message = str(exc) + if lockout_message: + diagnostics.append( + ViewDiagnostic( + severity="error", + code="view.lockout_escape_unavailable", + message=( + f"{lockout_message} The full authorized interface " + "is shown until the assignment is repaired." + ), + surface_ids=tuple(unavailable_escape_ids), + ) + ) + provenance.append( + { + "source": "invalid_view_fallback", + "scope_type": assignment.scope_type, + "scope_id": assignment.scope_id, + "detail": ( + "Required View administration escape is unavailable " + f"for revision {revision.id}" + ), + } + ) + selected = None + locked = False known_visible = [ known_surfaces[surface_id] for surface_id in revision.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 selected is not None and ( + not any(surface.kind == "navigation" for surface in known_visible) + or not any(surface.kind == "route" for surface in known_visible) + ): diagnostics.append( ViewDiagnostic( severity="error", diff --git a/tests/test_views.py b/tests/test_views.py index b025d79..7660d12 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -310,6 +310,48 @@ class ViewsServiceTests(unittest.TestCase): state.effective.provenance[0]["source"], ) + def test_required_view_falls_back_when_admin_escape_surface_is_removed( + self, + ) -> None: + definition = self.create_published_definition( + name="Required task", + scope_type="system", + visible_surface_ids=lockout_safe_surface_ids(), + ) + create_assignment( + self.session, + tenant_id="tenant-1", + scope_type="system", + scope_id=None, + definition=definition, + revision_id=None, + mode="required", + priority=0, + is_active=True, + metadata={}, + catalogue=self.catalogue, + actor_id="account-admin", + ) + catalogue_without_admin_route = tuple( + surface + for surface in self.catalogue + if surface.id != "access.route.admin" + ) + + state = resolve_effective_view( + self.session, + tenant_id="tenant-1", + account_id="account-user", + catalogue=catalogue_without_admin_route, + ) + + self.assertIsNone(state.effective.view_id) + self.assertFalse(state.effective.locked) + self.assertIn( + "view.lockout_escape_unavailable", + {diagnostic.code for diagnostic in state.diagnostics}, + ) + def test_required_assignment_rejects_revision_without_escape_surfaces(self) -> None: definition = self.create_published_definition(name="Unsafe task")