From d5b13decfcdb023b955f53d61ea02073dfaee602 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 29 Jul 2026 19:09:03 +0200 Subject: [PATCH] refactor: stage effective view resolution --- README.md | 10 +- src/govoplan_views/backend/capabilities.py | 2 + src/govoplan_views/backend/service.py | 729 ++++++++++++++------- tests/test_views.py | 177 +++++ 4 files changed, 694 insertions(+), 224 deletions(-) diff --git a/README.md b/README.md index 80b7011..2eb6c77 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,9 @@ pip install -e ../govoplan-views python -m unittest discover -s ../govoplan-views/tests -v ``` -Workflow declares the resolver as an optional contract so a future workflow -instance runtime can activate a pinned View revision. Workflow-driven -activation and Policy ceilings remain separate follow-up work; neither is -required for manually assigned and selected Views. +Workflow declares the resolver as an optional contract. The resolver accepts an +optional workflow-selected View and applies it below required administrative +assignments but above user/default selection. The View must already be +available to the account, so Workflow cannot bypass assignment policy. +Persisting a workflow instance's pinned View revision remains owned by the +Workflow module. diff --git a/src/govoplan_views/backend/capabilities.py b/src/govoplan_views/backend/capabilities.py index 91467f0..adfdb44 100644 --- a/src/govoplan_views/backend/capabilities.py +++ b/src/govoplan_views/backend/capabilities.py @@ -18,6 +18,7 @@ class ViewsResolverCapability(ViewResolver): tenant_id: str, account_id: str, group_ids: Iterable[str] = (), + workflow_view_id: str | None = None, ) -> EffectiveView: state = resolve_effective_view( session, @@ -25,6 +26,7 @@ class ViewsResolverCapability(ViewResolver): account_id=account_id, group_ids=group_ids, catalogue=self._registry.view_surfaces(), + workflow_view_id=workflow_view_id, ) return state.effective diff --git a/src/govoplan_views/backend/service.py b/src/govoplan_views/backend/service.py index 17dc7f0..be91397 100644 --- a/src/govoplan_views/backend/service.py +++ b/src/govoplan_views/backend/service.py @@ -8,8 +8,8 @@ from dataclasses import dataclass from datetime import datetime, timezone from typing import Literal -from sqlalchemy import or_ -from sqlalchemy.orm import Session +from sqlalchemy import and_, or_ +from sqlalchemy.orm import Session, joinedload from govoplan_core.core.views import ( VIEW_SURFACE_CONTRACT_VERSION, @@ -28,6 +28,8 @@ from govoplan_views.backend.db.models import ( ASSIGNMENT_MODES = frozenset({"available", "default", "required"}) ASSIGNMENT_SCOPES = frozenset({"system", "tenant", "group", "user"}) DEFINITION_SCOPES = frozenset({"system", "tenant", "group", "user"}) +MAX_EFFECTIVE_VIEW_ASSIGNMENTS = 1000 +MAX_EFFECTIVE_VIEW_GROUPS = 500 LOCKOUT_BASE_SURFACE_IDS = ( "access.module", "access.nav.admin", @@ -81,6 +83,18 @@ class EffectiveViewState: effective: EffectiveView available_views: tuple[EffectiveViewOption, ...] diagnostics: tuple[ViewDiagnostic, ...] = () + invalidation_token: str | None = None + + +ResolvedAssignment = tuple[ViewAssignment, ViewRevision] + + +@dataclass(frozen=True, slots=True) +class ViewSelection: + selected: ResolvedAssignment | None + locked: bool + provenance: tuple[dict[str, object], ...] + diagnostics: tuple[ViewDiagnostic, ...] = () def _now() -> datetime: @@ -660,19 +674,6 @@ def list_assignments( ).all() -def _assignment_revision( - session: Session, - assignment: ViewAssignment, -) -> ViewRevision | None: - revision_id = assignment.revision_id or assignment.definition.published_revision_id - if not revision_id: - return None - revision = session.get(ViewRevision, revision_id) - if revision is None or revision.definition_id != assignment.definition_id: - return None - return revision - - def _validate_assignment_revision( session: Session, definition: ViewDefinition, @@ -874,26 +875,6 @@ def delete_assignment(session: Session, assignment: ViewAssignment) -> None: session.flush() -def _assignment_applies( - assignment: ViewAssignment, - *, - tenant_id: str, - account_id: str, - group_ids: frozenset[str], -) -> bool: - if assignment.scope_type == "system": - return True - if assignment.tenant_id != tenant_id: - return False - if assignment.scope_type == "tenant": - return assignment.scope_id == tenant_id - if assignment.scope_type == "group": - return bool(assignment.scope_id and assignment.scope_id in group_ids) - if assignment.scope_type == "user": - return assignment.scope_id == account_id - return False - - def _assignment_order(assignment: ViewAssignment) -> tuple[int, int, float, str]: specificity = { "system": 0, @@ -923,57 +904,113 @@ def _preference( ) -def resolve_effective_view( +def _collect_effective_view_candidates( session: Session, *, tenant_id: str, account_id: str, - group_ids: Iterable[str] = (), - catalogue: Iterable[ViewSurface] | None = None, -) -> EffectiveViewState: - groups = frozenset(str(group_id) for group_id in group_ids) - candidates = ( + group_ids: frozenset[str], +) -> tuple[tuple[ViewAssignment, ...], bool]: + scope_filters = [ + and_( + ViewAssignment.scope_type == "system", + ViewAssignment.tenant_id.is_(None), + ), + and_( + ViewAssignment.scope_type == "tenant", + ViewAssignment.tenant_id == tenant_id, + ViewAssignment.scope_id == tenant_id, + ), + and_( + ViewAssignment.scope_type == "user", + ViewAssignment.tenant_id == tenant_id, + ViewAssignment.scope_id == account_id, + ), + ] + if group_ids: + scope_filters.append( + and_( + ViewAssignment.scope_type == "group", + ViewAssignment.tenant_id == tenant_id, + ViewAssignment.scope_id.in_(group_ids), + ) + ) + rows = ( session.query(ViewAssignment) .join(ViewDefinition, ViewAssignment.definition_id == ViewDefinition.id) + .options(joinedload(ViewAssignment.definition)) .filter( ViewAssignment.is_active.is_(True), ViewDefinition.status == "published", ViewDefinition.deleted_at.is_(None), - or_( - ViewAssignment.tenant_id == tenant_id, - ViewAssignment.tenant_id.is_(None), - ), + or_(*scope_filters), ) + .order_by(ViewAssignment.id.asc()) + .limit(MAX_EFFECTIVE_VIEW_ASSIGNMENTS + 1) .all() ) - applicable = [ - assignment + return ( + tuple(rows[:MAX_EFFECTIVE_VIEW_ASSIGNMENTS]), + len(rows) > MAX_EFFECTIVE_VIEW_ASSIGNMENTS, + ) + + +def _resolve_assignment_revisions( + session: Session, + assignments: Iterable[ViewAssignment], +) -> tuple[ResolvedAssignment, ...]: + candidates = tuple(assignments) + revision_ids = { + revision_id for assignment in candidates - if _assignment_applies( - assignment, - tenant_id=tenant_id, - account_id=account_id, - group_ids=groups, + if ( + revision_id := ( + assignment.revision_id + or assignment.definition.published_revision_id + ) ) - ] + } + revisions = ( + session.query(ViewRevision) + .filter(ViewRevision.id.in_(revision_ids)) + .all() + if revision_ids + else () + ) + by_id = {revision.id: revision for revision in revisions} + resolved: list[ResolvedAssignment] = [] + for assignment in candidates: + revision_id = ( + assignment.revision_id + or assignment.definition.published_revision_id + ) + revision = by_id.get(revision_id) + if ( + revision is not None + and revision.definition_id == assignment.definition_id + ): + resolved.append((assignment, revision)) + return tuple(resolved) - resolved_assignments: list[tuple[ViewAssignment, ViewRevision]] = [] - for assignment in applicable: - revision = _assignment_revision(session, assignment) - if revision is None: - continue - resolved_assignments.append((assignment, revision)) - option_assignments: dict[str, tuple[ViewAssignment, ViewRevision]] = {} - for assignment, revision in resolved_assignments: - existing = option_assignments.get(assignment.definition_id) +def _option_assignments( + assignments: Iterable[ResolvedAssignment], +) -> dict[str, ResolvedAssignment]: + options: dict[str, ResolvedAssignment] = {} + for assignment, revision in assignments: + existing = options.get(assignment.definition_id) if existing is None or _assignment_order(assignment) > _assignment_order( existing[0] ): - option_assignments[assignment.definition_id] = (assignment, revision) + options[assignment.definition_id] = (assignment, revision) + return options - ordered_options = sorted( - option_assignments.values(), + +def _available_view_options( + options: dict[str, ResolvedAssignment], +) -> tuple[EffectiveViewOption, ...]: + ordered = sorted( + options.values(), key=lambda item: ( -_assignment_order(item[0])[0], -_assignment_order(item[0])[1], @@ -981,186 +1018,438 @@ def resolve_effective_view( item[0].definition_id, ), ) - available_views = tuple( + return tuple( EffectiveViewOption( id=assignment.definition.id, name=assignment.definition.name, description=assignment.definition.description, revision_id=revision.id, ) - for assignment, revision in ordered_options + for assignment, revision in ordered ) - selected: tuple[ViewAssignment, ViewRevision] | None = None - locked = False - required = [pair for pair in resolved_assignments if pair[0].mode == "required"] - preference = _preference( - session, - tenant_id=tenant_id, - account_id=account_id, - ) - provenance: list[dict[str, object]] = [] + +def _assignment_provenance( + source: str, + assignment: ViewAssignment, + detail: str, +) -> dict[str, object]: + return { + "source": source, + "scope_type": assignment.scope_type, + "scope_id": assignment.scope_id, + "detail": detail, + } + + +def _select_effective_assignment( + assignments: tuple[ResolvedAssignment, ...], + options: dict[str, ResolvedAssignment], + *, + preference: ViewPreference | None, + account_id: str, + workflow_view_id: str | None, +) -> ViewSelection: + required = [ + pair for pair in assignments if pair[0].mode == "required" + ] if required: - selected = max(required, key=lambda pair: _assignment_order(pair[0])) - locked = True - provenance.append( - { - "source": "required_assignment", - "scope_type": selected[0].scope_type, - "scope_id": selected[0].scope_id, - "detail": f"Required View assignment {selected[0].id}", - } + selected = max( + required, + key=lambda pair: _assignment_order(pair[0]), ) - elif preference and preference.selection_kind == "none": - provenance.append( - { - "source": "user_selection", - "scope_type": "user", - "scope_id": account_id, - "detail": "Full interface selected", - } + return ViewSelection( + selected=selected, + locked=True, + provenance=( + _assignment_provenance( + "required_assignment", + selected[0], + f"Required View assignment {selected[0].id}", + ), + ), ) - elif ( - preference - and preference.selection_kind == "view" - and preference.view_id in option_assignments - ): - selected = option_assignments[preference.view_id] - provenance.append( - { - "source": "user_selection", - "scope_type": "user", - "scope_id": account_id, - "detail": f"Selected View {preference.view_id}", - } - ) - else: - defaults = [pair for pair in resolved_assignments if pair[0].mode == "default"] - if defaults: - selected = max(defaults, key=lambda pair: _assignment_order(pair[0])) - provenance.append( - { - "source": "default_assignment", - "scope_type": selected[0].scope_type, - "scope_id": selected[0].scope_id, - "detail": f"Default View assignment {selected[0].id}", - } - ) diagnostics: list[ViewDiagnostic] = [] - if selected is not None and catalogue is not None: - assignment, revision = selected - current_catalogue = tuple(catalogue) - known_surfaces = {surface.id: surface for surface in current_catalogue} - stale_ids = sorted(set(revision.visible_surface_ids) - set(known_surfaces)) - if stale_ids: - diagnostics.append( - ViewDiagnostic( - severity="warning", - code="view.stale_surfaces", - message=( - "The active View references surfaces that are no longer " - "announced by the active module graph." + if workflow_view_id is not None: + selected = options.get(workflow_view_id) + if selected is not None: + return ViewSelection( + selected=selected, + locked=False, + provenance=( + _assignment_provenance( + "workflow_selection", + selected[0], + f"Workflow selected View {workflow_view_id}", ), - surface_ids=tuple(stale_ids), - ) + ), ) - if locked: - try: - lockout_ids = lockout_required_surface_ids( - current_catalogue, - assignment_scope_type=assignment.scope_type, - ) - unavailable_escape_ids = sorted( - lockout_ids - set(revision.visible_surface_ids) - ) - lockout_message = ( - "The required View no longer retains all administration " - "escape surfaces." - if unavailable_escape_ids - else "" - ) - except ViewsValidationError as exc: - unavailable_escape_ids = () - lockout_message = str(exc) - if lockout_message: - diagnostics.append( - ViewDiagnostic( - severity="error", - code="view.lockout_escape_unavailable", - message=( - f"{lockout_message} The full authorized interface " - "is shown until the assignment is repaired." - ), - surface_ids=tuple(unavailable_escape_ids), - ) - ) - provenance.append( - { - "source": "invalid_view_fallback", - "scope_type": assignment.scope_type, - "scope_id": assignment.scope_id, - "detail": ( - "Required View administration escape is unavailable " - f"for revision {revision.id}" - ), - } - ) - selected = None - locked = False - known_visible = [ - known_surfaces[surface_id] - for surface_id in revision.visible_surface_ids - if surface_id in known_surfaces - ] - if selected is not None and ( - not any(surface.kind == "navigation" for surface in known_visible) - or not any(surface.kind == "route" for surface in known_visible) - ): - diagnostics.append( - ViewDiagnostic( - severity="error", - code="view.no_reachable_surface", - message=( - "The active View no longer contains a reachable navigation " - "entry and route. The full authorized interface is shown." - ), - ) + diagnostics.append( + ViewDiagnostic( + severity="warning", + code="view.workflow_selection_unavailable", + message=( + "The workflow-selected View is not available to this " + "account. Normal View selection was used." + ), ) - provenance.append( - { - "source": "invalid_view_fallback", - "scope_type": assignment.scope_type, - "scope_id": assignment.scope_id, - "detail": f"Invalid View revision {revision.id}", - } - ) - selected = None - locked = False + ) - if selected is None: - effective = EffectiveView( + if preference and preference.selection_kind == "none": + return ViewSelection( + selected=None, + locked=False, + provenance=( + { + "source": "user_selection", + "scope_type": "user", + "scope_id": account_id, + "detail": "Full interface selected", + }, + ), + diagnostics=tuple(diagnostics), + ) + if ( + preference + and preference.selection_kind == "view" + and preference.view_id in options + ): + selected = options[preference.view_id] + return ViewSelection( + selected=selected, + locked=False, + provenance=( + { + "source": "user_selection", + "scope_type": "user", + "scope_id": account_id, + "detail": f"Selected View {preference.view_id}", + }, + ), + diagnostics=tuple(diagnostics), + ) + + defaults = [ + pair for pair in assignments if pair[0].mode == "default" + ] + if defaults: + selected = max( + defaults, + key=lambda pair: _assignment_order(pair[0]), + ) + return ViewSelection( + selected=selected, + locked=False, + provenance=( + _assignment_provenance( + "default_assignment", + selected[0], + f"Default View assignment {selected[0].id}", + ), + ), + diagnostics=tuple(diagnostics), + ) + return ViewSelection( + selected=None, + locked=False, + provenance=(), + diagnostics=tuple(diagnostics), + ) + + +def _evaluate_selected_surfaces( + selection: ViewSelection, + *, + catalogue: tuple[ViewSurface, ...] | None, +) -> ViewSelection: + if selection.selected is None or catalogue is None: + return selection + assignment, revision = selection.selected + known_surfaces = {surface.id: surface for surface in catalogue} + stale_ids = sorted( + set(revision.visible_surface_ids) - set(known_surfaces) + ) + diagnostics = list(selection.diagnostics) + provenance = list(selection.provenance) + if stale_ids: + diagnostics.append( + ViewDiagnostic( + severity="warning", + code="view.stale_surfaces", + message=( + "The active View references surfaces that are no longer " + "announced by the active module graph." + ), + surface_ids=tuple(stale_ids), + ) + ) + + if selection.locked: + lockout_diagnostic = _required_lockout_diagnostic( + assignment, + revision, + catalogue=catalogue, + ) + if lockout_diagnostic is not None: + diagnostics.append(lockout_diagnostic) + provenance.append( + _fallback_provenance( + assignment, + detail=( + "Required View administration escape is unavailable " + f"for revision {revision.id}" + ), + ) + ) + return ViewSelection( + selected=None, + locked=False, + provenance=tuple(provenance), + diagnostics=tuple(diagnostics), + ) + + known_visible = [ + known_surfaces[surface_id] + for surface_id in revision.visible_surface_ids + if surface_id in known_surfaces + ] + if ( + not any(surface.kind == "navigation" for surface in known_visible) + or not any(surface.kind == "route" for surface in known_visible) + ): + diagnostics.append( + ViewDiagnostic( + severity="error", + code="view.no_reachable_surface", + message=( + "The active View no longer contains a reachable navigation " + "entry and route. The full authorized interface is shown." + ), + ) + ) + provenance.append( + _fallback_provenance( + assignment, + detail=f"Invalid View revision {revision.id}", + ) + ) + return ViewSelection( + selected=None, + locked=False, + provenance=tuple(provenance), + diagnostics=tuple(diagnostics), + ) + return ViewSelection( + selected=selection.selected, + locked=selection.locked, + provenance=tuple(provenance), + diagnostics=tuple(diagnostics), + ) + + +def _required_lockout_diagnostic( + assignment: ViewAssignment, + revision: ViewRevision, + *, + catalogue: tuple[ViewSurface, ...], +) -> ViewDiagnostic | None: + try: + required_ids = lockout_required_surface_ids( + catalogue, + assignment_scope_type=assignment.scope_type, + ) + missing_ids = tuple( + sorted(required_ids - set(revision.visible_surface_ids)) + ) + message = ( + "The required View no longer retains all administration escape surfaces." + if missing_ids + else None + ) + except ViewsValidationError as exc: + missing_ids = () + message = str(exc) + if message is None: + return None + return ViewDiagnostic( + severity="error", + code="view.lockout_escape_unavailable", + message=( + f"{message} The full authorized interface is shown until the " + "assignment is repaired." + ), + surface_ids=missing_ids, + ) + + +def _fallback_provenance( + assignment: ViewAssignment, + *, + detail: str, +) -> dict[str, object]: + return _assignment_provenance( + "invalid_view_fallback", + assignment, + detail, + ) + + +def _effective_view_from_selection(selection: ViewSelection) -> EffectiveView: + if selection.selected is None: + return EffectiveView( view_id=None, revision_id=None, name=None, visible_surface_ids=frozenset(), locked=False, - provenance=tuple(provenance), + provenance=selection.provenance, ) - else: - assignment, revision = selected - effective = EffectiveView( - view_id=assignment.definition.id, - revision_id=revision.id, - name=assignment.definition.name, - visible_surface_ids=frozenset(revision.visible_surface_ids), - locked=locked, - provenance=tuple(provenance), + assignment, revision = selection.selected + return EffectiveView( + view_id=assignment.definition.id, + revision_id=revision.id, + name=assignment.definition.name, + visible_surface_ids=frozenset(revision.visible_surface_ids), + locked=selection.locked, + provenance=selection.provenance, + ) + + +def _resolution_invalidation_token( + assignments: Iterable[ResolvedAssignment], + *, + preference: ViewPreference | None, + catalogue: tuple[ViewSurface, ...] | None, + workflow_view_id: str | None, +) -> str: + payload = { + "assignments": [ + { + "id": assignment.id, + "revision_id": revision.id, + "mode": assignment.mode, + "priority": assignment.priority, + "updated_at": assignment.updated_at.isoformat(), + } + for assignment, revision in sorted( + assignments, + key=lambda pair: pair[0].id, + ) + ], + "preference": ( + { + "selection_kind": preference.selection_kind, + "view_id": preference.view_id, + "updated_at": preference.updated_at.isoformat(), + } + if preference is not None + else None + ), + "surfaces": ( + sorted( + ( + surface.id, + surface.module_id, + surface.kind, + surface.parent_id, + surface.required, + ) + for surface in catalogue + ) + if catalogue is not None + else None + ), + "workflow_view_id": workflow_view_id, + } + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def resolve_effective_view( + session: Session, + *, + tenant_id: str, + account_id: str, + group_ids: Iterable[str] = (), + catalogue: Iterable[ViewSurface] | None = None, + workflow_view_id: str | None = None, +) -> EffectiveViewState: + groups = frozenset(str(group_id) for group_id in group_ids) + if len(groups) > MAX_EFFECTIVE_VIEW_GROUPS: + raise ViewsValidationError( + "Too many group memberships were supplied for View resolution" ) + current_catalogue = tuple(catalogue) if catalogue is not None else None + candidates, overflow = _collect_effective_view_candidates( + session, + tenant_id=tenant_id, + account_id=account_id, + group_ids=groups, + ) + if overflow: + diagnostic = ViewDiagnostic( + severity="error", + code="view.assignment_limit_exceeded", + message=( + "View resolution exceeded its assignment safety limit. The " + "full authorized interface is shown." + ), + ) + return EffectiveViewState( + effective=_effective_view_from_selection( + ViewSelection( + selected=None, + locked=False, + provenance=( + { + "source": "invalid_view_fallback", + "scope_type": "user", + "scope_id": account_id, + "detail": "View assignment safety limit exceeded", + }, + ), + diagnostics=(diagnostic,), + ) + ), + available_views=(), + diagnostics=(diagnostic,), + ) + resolved_assignments = _resolve_assignment_revisions( + session, + candidates, + ) + options = _option_assignments(resolved_assignments) + preference = _preference( + session, + tenant_id=tenant_id, + account_id=account_id, + ) + selection = _select_effective_assignment( + resolved_assignments, + options, + preference=preference, + account_id=account_id, + workflow_view_id=workflow_view_id, + ) + evaluated = _evaluate_selected_surfaces( + selection, + catalogue=current_catalogue, + ) return EffectiveViewState( - effective=effective, - available_views=available_views, - diagnostics=tuple(diagnostics), + effective=_effective_view_from_selection(evaluated), + available_views=_available_view_options(options), + diagnostics=evaluated.diagnostics, + invalidation_token=_resolution_invalidation_token( + resolved_assignments, + preference=preference, + catalogue=current_catalogue, + workflow_view_id=workflow_view_id, + ), ) diff --git a/tests/test_views.py b/tests/test_views.py index 8b5e605..c4d51a1 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -128,12 +128,14 @@ class ViewsServiceTests(unittest.TestCase): *, 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, @@ -154,6 +156,30 @@ class ViewsServiceTests(unittest.TestCase): ) 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(), @@ -311,6 +337,157 @@ class ViewsServiceTests(unittest.TestCase): {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_then_workflow_then_user_and_default_precedence(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, + ) + self.assertEqual(required.id, required_state.effective.view_id) + self.assertTrue(required_state.effective.locked) + self.assertEqual( + "required_assignment", + required_state.effective.provenance[0]["source"], + ) + + 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: