559 lines
17 KiB
Python
559 lines
17 KiB
Python
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()
|