1237 lines
39 KiB
Python
1237 lines
39 KiB
Python
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()
|