Add governed view revision resolution

This commit is contained in:
2026-07-31 02:48:57 +02:00
parent 2df42fb0f8
commit 5c5aeecc19
8 changed files with 496 additions and 43 deletions
+173 -3
View File
@@ -398,7 +398,7 @@ class ViewsServiceTests(unittest.TestCase):
state.effective.view_id,
)
def test_required_then_workflow_then_user_and_default_precedence(self) -> None:
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(
@@ -443,12 +443,182 @@ class ViewsServiceTests(unittest.TestCase):
catalogue=self.catalogue,
workflow_view_id=workflow.id,
)
self.assertEqual(required.id, required_state.effective.view_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",
"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")