from __future__ import annotations 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 ( ViewsConflictError, ViewsValidationError, archive_definition, create_assignment, create_definition, create_revision, get_revision, list_definitions, normalize_view_presentation, 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 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:") 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 test_view_presentation_is_normalized_and_versioned(self) -> None: definition = create_definition( self.session, tenant_id="tenant-1", scope_type="tenant", scope_id=None, definition_key=None, name="Product navigation", description=None, visible_surface_ids=ordinary_surface_ids(), catalogue=self.catalogue, actor_id="account-admin", presentation={ "navigation_mode": "grouped", "product_area_order": ["work", "records-documents"], "product_area_labels": {"work": "My work"}, "quick_access_recommended_tool_ids": ["tasks.work"], "quick_access_focused_tool_ids": ["tasks.work", "mail.messages"], }, available_product_area_ids=("work", "records-documents"), ) revision = get_revision(self.session, definition_id=definition.id) self.assertEqual("grouped", revision.presentation["navigation_mode"]) self.assertEqual( ["tasks.work"], revision.presentation["quick_access_recommended_tool_ids"], ) compatible_revision = create_revision( self.session, definition, visible_surface_ids=lockout_safe_surface_ids(), catalogue=self.catalogue, actor_id="legacy-client", available_product_area_ids=("work", "records-documents"), ) self.assertEqual( revision.presentation, compatible_revision.presentation, "omitting presentation must preserve the previous revision contract", ) revision = compatible_revision publish_revision( self.session, definition, revision, catalogue=self.catalogue, actor_id="account-admin", ) self.assign( definition, scope_type="tenant", scope_id=None, ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-1", catalogue=self.catalogue, ) self.assertEqual("My work", state.effective.presentation["product_area_labels"]["work"]) def test_view_presentation_rejects_unknown_or_unavailable_fields(self) -> None: with self.assertRaises(ViewsValidationError): normalize_view_presentation({"unknown": True}) with self.assertRaises(ViewsValidationError): normalize_view_presentation( {"product_area_order": ["unavailable"]}, available_product_area_ids=("work",), ) with self.assertRaises(ViewsValidationError): normalize_view_presentation( {"quick_access_recommended_tool_ids": ["not namespaced"]} ) def create_published_definition( self, *, name: str, scope_type: str = "tenant", scope_id: str | None = None, visible_surface_ids: list[str] | None = None, ): definition = create_definition( self.session, tenant_id="tenant-1", scope_type=scope_type, scope_id=scope_id, 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 assign( self, definition, *, scope_type: str, scope_id: str | None, mode: str = "default", priority: int = 0, ): return create_assignment( self.session, tenant_id="tenant-1", scope_type=scope_type, scope_id=scope_id, definition=definition, revision_id=None, mode=mode, priority=priority, is_active=True, metadata={}, catalogue=self.catalogue, actor_id="account-admin", ) 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_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, 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_user_and_group_views_hide_tenant_effective_module_roots(self) -> None: group_view = self.create_published_definition( name="Group files", scope_type="group", scope_id="group-1", visible_surface_ids=ordinary_surface_ids(), ) user_view = self.create_published_definition( name="Personal files", scope_type="user", scope_id="account-user", visible_surface_ids=ordinary_surface_ids(), ) self.assign( group_view, scope_type="group", scope_id="group-1", ) self.assign( user_view, scope_type="user", scope_id="account-user", ) cases = ( ("account-member", ("group-1",), group_view.id), ("account-user", (), user_view.id), ) for account_id, group_ids, expected_view_id in cases: with self.subTest(account_id=account_id): state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id=account_id, group_ids=group_ids, catalogue=self.catalogue, ) self.assertEqual(expected_view_id, state.effective.view_id) self.assertIn("files.module", state.effective.visible_surface_ids) self.assertNotIn("access.module", state.effective.visible_surface_ids) def test_group_view_cannot_restore_a_tenant_unavailable_module(self) -> None: definition = self.create_published_definition( name="Group files", scope_type="group", scope_id="group-1", visible_surface_ids=ordinary_surface_ids(), ) self.assign( definition, scope_type="group", scope_id="group-1", ) 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-member", group_ids=("group-1",), catalogue=active_catalogue, ) self.assertIsNone(state.effective.view_id) self.assertNotIn("files.module", state.effective.visible_surface_ids) self.assertIn( "view.stale_surfaces", {diagnostic.code for diagnostic in state.diagnostics}, ) 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_scope_precedence_is_system_tenant_group_user(self) -> None: definitions = { "system": self.create_published_definition( name="System", scope_type="system", ), "tenant": self.create_published_definition(name="Tenant"), "group": self.create_published_definition( name="Group", scope_type="group", scope_id="group-1", ), "user": self.create_published_definition( name="User", scope_type="user", scope_id="account-user", ), } self.assign( definitions["system"], scope_type="system", scope_id=None, priority=1000, ) self.assign( definitions["tenant"], scope_type="tenant", scope_id=None, priority=100, ) self.assign( definitions["group"], scope_type="group", scope_id="group-1", priority=10, ) self.assign( definitions["user"], scope_type="user", scope_id="account-user", priority=0, ) cases = ( ("account-outsider", (), "tenant"), ("account-member", ("group-1",), "group"), ("account-user", ("group-1",), "user"), ) for account_id, group_ids, expected_scope in cases: with self.subTest(account_id=account_id): state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id=account_id, group_ids=group_ids, ) self.assertEqual( definitions[expected_scope].id, state.effective.view_id, ) def test_required_assignment_limits_workflow_selection(self) -> None: default = self.create_published_definition(name="Default") workflow = self.create_published_definition(name="Workflow") required = self.create_published_definition( name="Required", visible_surface_ids=lockout_safe_surface_ids(), ) self.assign( default, scope_type="tenant", scope_id=None, mode="default", ) self.assign( workflow, scope_type="tenant", scope_id=None, mode="available", ) workflow_state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", workflow_view_id=workflow.id, ) self.assertEqual(workflow.id, workflow_state.effective.view_id) self.assertEqual( "workflow_selection", workflow_state.effective.provenance[0]["source"], ) self.assign( required, scope_type="tenant", scope_id=None, mode="required", ) required_state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue, workflow_view_id=workflow.id, ) workflow_revision = get_revision( self.session, definition_id=workflow.id, ) required_revision = get_revision( self.session, definition_id=required.id, ) self.assertEqual(workflow.id, required_state.effective.view_id) self.assertTrue(required_state.effective.locked) self.assertEqual( "required_assignment_ceiling", required_state.effective.provenance[0]["source"], ) self.assertEqual( set(workflow_revision.visible_surface_ids) & set(required_revision.visible_surface_ids), set(required_state.effective.visible_surface_ids), ) def test_workflow_can_pin_a_historical_view_revision(self) -> None: definition = self.create_published_definition(name="Workflow") pinned_revision = get_revision( self.session, definition_id=definition.id, ) self.assign( definition, scope_type="tenant", scope_id=None, mode="available", ) current_revision = create_revision( self.session, definition, visible_surface_ids=["access.nav.admin", "access.route.admin"], catalogue=self.catalogue, actor_id="account-admin", ) publish_revision( self.session, definition, current_revision, catalogue=self.catalogue, actor_id="account-admin", ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue, workflow_view_id=definition.id, workflow_revision_id=pinned_revision.id, ) self.assertEqual(definition.id, state.effective.view_id) self.assertEqual(pinned_revision.id, state.effective.revision_id) self.assertIn("files.route.files", state.effective.visible_surface_ids) self.assertNotIn("access.route.admin", state.effective.visible_surface_ids) def test_workflow_step_overlay_only_narrows_the_selected_view(self) -> None: definition = self.create_published_definition( name="Workflow", visible_surface_ids=[ "access.nav.admin", "access.route.admin", "files.nav.files", "files.route.files", ], ) self.assign( definition, scope_type="tenant", scope_id=None, mode="available", ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue, workflow_view_id=definition.id, workflow_surface_ids=( "files.nav.files", "files.route.files", "unknown.route", ), ) self.assertIn("files.module", state.effective.visible_surface_ids) self.assertIn("files.nav.files", state.effective.visible_surface_ids) self.assertIn("files.route.files", state.effective.visible_surface_ids) self.assertIn("views.selector", state.effective.visible_surface_ids) self.assertNotIn("access.route.admin", state.effective.visible_surface_ids) self.assertIn( "view.workflow_overlay_unknown_surfaces", {diagnostic.code for diagnostic in state.diagnostics}, ) def test_missing_workflow_revision_falls_back_to_normal_selection(self) -> None: default = self.create_published_definition(name="Default") workflow = self.create_published_definition(name="Workflow") self.assign( default, scope_type="tenant", scope_id=None, mode="default", ) self.assign( workflow, scope_type="tenant", scope_id=None, mode="available", ) state = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue, workflow_view_id=workflow.id, workflow_revision_id="missing-revision", ) self.assertEqual(default.id, state.effective.view_id) self.assertIn( "view.workflow_revision_unavailable", {diagnostic.code for diagnostic in state.diagnostics}, ) def test_workflow_view_falls_back_after_group_authorization_is_lost( self, ) -> None: default = self.create_published_definition(name="Default") workflow = self.create_published_definition(name="Group workflow") self.assign( default, scope_type="tenant", scope_id=None, mode="default", ) self.assign( workflow, scope_type="group", scope_id="group-1", mode="available", ) authorized = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", group_ids=("group-1",), catalogue=self.catalogue, workflow_view_id=workflow.id, ) after_membership_removal = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", group_ids=(), catalogue=self.catalogue, workflow_view_id=workflow.id, ) self.assertEqual(workflow.id, authorized.effective.view_id) 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}, ) def test_resolution_invalidation_token_tracks_inputs(self) -> None: definition = self.create_published_definition(name="Files") self.assign( definition, scope_type="tenant", scope_id=None, mode="available", ) baseline = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue, ) workflow = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue, workflow_view_id=definition.id, ) reduced_catalogue = resolve_effective_view( self.session, tenant_id="tenant-1", account_id="account-user", catalogue=self.catalogue[:-1], ) self.assertIsNotNone(baseline.invalidation_token) self.assertNotEqual( baseline.invalidation_token, workflow.invalidation_token, ) self.assertNotEqual( baseline.invalidation_token, reduced_catalogue.invalidation_token, ) 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()