from __future__ import annotations import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_views.backend.service import ( ViewsConflictError, ViewsValidationError, archive_definition, create_assignment, create_definition, create_revision, get_revision, list_definitions, normalize_visible_surface_ids, publish_revision, resolve_effective_view, select_view, update_assignment, ) def surface_catalogue() -> tuple[ViewSurface, ...]: return ( ViewSurface( id="access.module", module_id="access", kind="module", label="Access", ), ViewSurface( id="access.nav.admin", module_id="access", kind="navigation", label="Administration", parent_id="access.module", ), ViewSurface( id="access.route.admin", module_id="access", kind="route", label="Administration", parent_id="access.module", ), ViewSurface( id="views.module", module_id="views", kind="module", label="Views", ), ViewSurface( id="views.selector", module_id="views", kind="selector", label="View selector", parent_id="views.module", required=True, ), ViewSurface( id="views.admin.system", module_id="views", kind="section", label="System Views", parent_id="views.module", ), ViewSurface( id="views.admin.tenant", module_id="views", kind="section", label="Tenant Views", parent_id="views.module", ), ViewSurface( id="files.module", module_id="files", kind="module", label="Files", ), ViewSurface( id="files.nav.files", module_id="files", kind="navigation", label="Files", parent_id="files.module", ), ViewSurface( id="files.route.files", module_id="files", kind="route", label="Files", parent_id="files.module", ), ) def ordinary_surface_ids() -> list[str]: return ["files.nav.files", "files.route.files"] def lockout_safe_surface_ids() -> list[str]: return [ "access.nav.admin", "access.route.admin", "views.admin.system", "views.admin.tenant", "files.nav.files", "files.route.files", ] class ViewsServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(self.engine) self.session = Session(self.engine) self.catalogue = surface_catalogue() def tearDown(self) -> None: self.session.close() self.engine.dispose() def create_published_definition( self, *, name: str, scope_type: str = "tenant", visible_surface_ids: list[str] | None = None, ): definition = create_definition( self.session, tenant_id="tenant-1", scope_type=scope_type, definition_key=None, name=name, description=None, visible_surface_ids=visible_surface_ids or ordinary_surface_ids(), catalogue=self.catalogue, actor_id="account-admin", ) revision = get_revision( self.session, definition_id=definition.id, ) publish_revision( self.session, definition, revision, catalogue=self.catalogue, actor_id="account-admin", ) return definition def test_normalization_adds_required_surfaces_and_ancestors(self) -> None: normalized = normalize_visible_surface_ids( ordinary_surface_ids(), catalogue=self.catalogue, ) self.assertIn("files.module", normalized) self.assertIn("views.selector", normalized) self.assertIn("views.module", normalized) def test_published_group_view_is_available_to_group_members(self) -> None: definition = create_definition( self.session, tenant_id="tenant-1", scope_type="group", scope_id="group-1", definition_key=None, name="Group workspace", description=None, visible_surface_ids=ordinary_surface_ids(), catalogue=self.catalogue, actor_id="account-designer", ) revision = get_revision(self.session, definition_id=definition.id) publish_revision( self.session, definition, revision, catalogue=self.catalogue, actor_id="account-designer", ) member = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-member", group_ids=("group-1",), ) outsider = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-outsider", group_ids=(), ) self.assertEqual([definition.id], [view.id for view in member.available_views]) self.assertEqual((), outsider.available_views) self.assertEqual( [definition.id], [ item.id for item in list_definitions( self.session, tenant_id="tenant-1", scope_type="group", scope_id="group-1", include_inherited=False, ) ], ) def test_normalization_requires_navigation_and_route(self) -> None: with self.assertRaisesRegex(ViewsValidationError, "navigation"): normalize_visible_surface_ids( ["files.route.files"], catalogue=self.catalogue, ) with self.assertRaisesRegex(ViewsValidationError, "route"): normalize_visible_surface_ids( ["files.nav.files"], catalogue=self.catalogue, ) def test_default_assignment_can_be_left_by_the_user(self) -> None: definition = self.create_published_definition(name="Files task") create_assignment( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition=definition, revision_id=None, mode="default", priority=0, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) initial = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", ) self.assertEqual(definition.id, initial.effective.view_id) self.assertFalse(initial.effective.locked) selected = select_view( self.session, tenant_id="tenant-1", account_id="account-user", group_ids=(), view_id=None, ) self.assertIsNone(selected.effective.view_id) self.assertEqual( "user_selection", selected.effective.provenance[0]["source"], ) def test_tenant_default_precedes_system_default(self) -> None: system_view = self.create_published_definition( name="System task", scope_type="system", ) tenant_view = self.create_published_definition(name="Tenant task") create_assignment( self.session, tenant_id="tenant-1", scope_type="system", scope_id=None, definition=system_view, revision_id=None, mode="default", priority=100, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) create_assignment( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition=tenant_view, revision_id=None, mode="default", priority=0, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", ) self.assertEqual(tenant_view.id, state.effective.view_id) self.assertEqual( {system_view.id, tenant_view.id}, {option.id for option in state.available_views}, ) def test_more_specific_available_assignment_cannot_override_required_mode( 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", ) create_assignment( self.session, tenant_id="tenant-1", scope_type="user", scope_id="account-user", definition=definition, revision_id=None, mode="available", priority=100, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", ) self.assertEqual(definition.id, state.effective.view_id) self.assertTrue(state.effective.locked) self.assertEqual( "required_assignment", 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") with self.assertRaisesRegex(ViewsValidationError, "administration escape"): create_assignment( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition=definition, revision_id=None, mode="required", priority=0, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) def test_retired_surfaces_fall_back_to_full_interface_with_diagnostics( self, ) -> None: definition = self.create_published_definition(name="Retired task") create_assignment( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition=definition, revision_id=None, mode="default", priority=0, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) active_catalogue = tuple( surface for surface in self.catalogue if surface.module_id != "files" ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=active_catalogue, ) self.assertIsNone(state.effective.view_id) self.assertEqual( {"view.stale_surfaces", "view.no_reachable_surface"}, {diagnostic.code for diagnostic in state.diagnostics}, ) self.assertEqual( "invalid_view_fallback", state.effective.provenance[-1]["source"], ) def test_required_assignment_is_locked_and_blocks_archive(self) -> None: definition = self.create_published_definition( name="Required task", visible_surface_ids=lockout_safe_surface_ids(), ) create_assignment( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition=definition, revision_id=None, mode="required", priority=0, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", ) self.assertEqual(definition.id, state.effective.view_id) self.assertTrue(state.effective.locked) with self.assertRaisesRegex(ViewsConflictError, "required assignments"): archive_definition( self.session, definition, actor_id="account-admin", ) with self.assertRaisesRegex(ViewsConflictError, "cannot be changed"): select_view( self.session, tenant_id="tenant-1", account_id="account-user", group_ids=(), view_id=None, ) def test_historical_revision_pin_survives_assignment_edits(self) -> None: definition = self.create_published_definition( name="Pinned task", visible_surface_ids=lockout_safe_surface_ids(), ) pinned_revision = get_revision( self.session, definition_id=definition.id, ) assignment = create_assignment( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition=definition, revision_id=pinned_revision.id, mode="required", priority=0, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) next_revision = create_revision( self.session, definition, visible_surface_ids=ordinary_surface_ids(), catalogue=self.catalogue, actor_id="account-admin", ) publish_revision( self.session, definition, next_revision, catalogue=self.catalogue, actor_id="account-admin", ) updated = update_assignment( self.session, assignment, updates={ "revision_id": pinned_revision.id, "priority": 25, }, catalogue=self.catalogue, actor_id="account-admin", ) self.assertEqual(pinned_revision.id, updated.revision_id) self.assertEqual(25, updated.priority) if __name__ == "__main__": unittest.main()