"""Shared-shell capability evaluation: correctness guards and query profile. Phase 5D.1 measured the Phase 4D shared shell's authorization query cost. The navigation shell is the only part of a page that resolves capabilities for permissions the page itself does not use, so its behaviour is guarded here independently of the Phase 3D report budgets. The Phase 3D budgets are deliberately NOT restated or relaxed here. They are enforced, unchanged, by the two frozen tests below, which pass since Phase 5D.3 (see docs/RELEASE_CHECKLIST.md, Release blocker 1): apps.reporting.tests.test_reports.ReportingTests .test_representative_pages_have_bounded_queries tests.test_phase3d_audit.CommercialReconciliationAudit .test_performance_all_representative_pages_and_streaming Nothing in this module caches an authorization result, so there is no request-scoped or cross-request cache to leak; the only memoization that exists is the per-request ``(permission, company_only)`` capability dict owned by ``apps.operations.context.shell``. """ import itertools from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.db import connection from django.test import TestCase from django.test.utils import CaptureQueriesContext from django.urls import reverse from apps.access import authorization as az from apps.access.authorization import authorized_queryset, is_authorized from apps.access.models import Role from apps.access.services import create_role_assignment, set_role_permissions from apps.catalog.tests import make_catalog from apps.organization.assignment_services import create_assignment from apps.organization.models import Department, ServiceCenter from apps.organization.test_lifecycle import make_tree from apps.operations import queries as operations_queries from apps.operations.context import shell, capability_questions class ShellCapabilityFixture(TestCase): """Two companies, two centers and one department, plus scoped personas.""" @classmethod def setUpTestData(cls): cls.company, cls.region, cls.center, _ = make_tree("SHELL") cls.other_company, _, cls.other_center, _ = make_tree("SHELLOTHER") cls.department = Department.objects.create(company=cls.company, code="SHELL", name="Shell department") make_catalog("SHELL") _unique = itertools.count() def persona(self, username, permissions, *, center=None, region=None, department=None): username = f"{username}-{next(self._unique)}" user = get_user_model().objects.create_user(username=username) role = Role.objects.create(code=f"SHELL-{username}", name=f"Shell {username}") set_role_permissions(role=role, permissions=list(permissions)) # A service-center assignment must name its region; derive it when omitted. if center is not None and region is None and department is None: region = center.region path = create_assignment(user=user, company=self.company, region=region, department=department, service_center=center) create_role_assignment(user=user, role=role, organization_assignment=path) return user def granted(self, *codenames): return Permission.objects.filter(content_type__app_label__in=[c.split(".")[0] for c in codenames], codename__in=[c.split(".")[1] for c in codenames]) def scope_reader(self): return self.persona("shell-reader", self.granted("service.view_servicecase"), center=self.center) def company_reader(self): return self.persona("shell-company", self.granted("service.view_servicecase")) def department_reader(self): return self.persona("shell-department", self.granted("service.view_servicecase"), department=self.department) def second_center_reader(self): """Same company and region, but a different center than the case's.""" sibling = ServiceCenter.objects.create(company=self.company, region=self.region, code="SHELLB", name="Shell sibling", center_type="OWN") return self.persona("shell-sibling", self.granted("service.view_servicecase"), center=sibling), sibling def navigation_labels(self, user): self.client.force_login(user) response = self.client.get(reverse("operations:dashboard")) self.assertEqual(response.status_code, 200) workspace = response.context["workspace"] for group in workspace["groups"]: for link in group["links"]: if link["label"] in ("Service cases", "Customers"): kind = "cases" if link["label"] == "Service cases" else "customers" self.assertEqual(link["url"], reverse("operations:list", args=[kind])) return {(group["label"], link["label"]) for group in workspace["groups"] for link in group["links"]} class CapabilityOutcomeTests(ShellCapabilityFixture): """A. Capability results are exactly what the authoritative path returns.""" def test_capable_agrees_with_the_authoritative_querysets(self): for user in (self.scope_reader(), self.company_reader(), self.department_reader(), get_user_model().objects.create_superuser(username="shell-superuser")): for permission in ("service.view_servicecase", "service.change_servicecase", "devices.view_device"): with self.subTest(user=user.username, permission=permission): authoritative = (operations_queries.companies(user, permission).exists() or operations_queries.centers(user, permission).exists()) self.assertEqual(operations_queries.capable(user, permission), authoritative) def test_company_only_capability_never_reports_a_center_scope(self): reader = self.scope_reader() self.assertTrue(operations_queries.capable(reader, "service.view_servicecase")) self.assertFalse(operations_queries.capable(reader, "service.view_servicecase", company_only=True), "a center-scoped persona must not satisfy a company-only capability") def test_absent_permission_is_denied_even_for_a_superuser(self): superuser = get_user_model().objects.create_superuser(username="shell-superuser-2") self.assertFalse(operations_queries.capable(superuser, "service.no_such_permission")) def test_unknown_permission_identifier_is_denied(self): self.assertFalse(operations_queries.capable(self.company_reader(), "not-a-permission")) self.assertFalse(operations_queries.capable(self.company_reader(), "service.view_servicecase.extra")) def test_inactive_user_has_no_capability(self): user = self.scope_reader() get_user_model().objects.filter(pk=user.pk).update(is_active=False) self.assertFalse(operations_queries.capable(get_user_model().objects.get(pk=user.pk), "service.view_servicecase")) class NavigationVisibilityTests(ShellCapabilityFixture): """B and C. Permitted navigation stays visible; denied navigation stays hidden.""" def test_permitted_navigation_is_visible(self): user = self.scope_reader() from apps.service.tests import customer_create from apps.devices import services as devices from apps.devices.tests import policy from apps.service.services import create_service_case _, _, model, _ = make_catalog("NAVCASE") policy(model) case = create_service_case(company=self.company, service_center=self.center, customer=customer_create(self.company), device=devices.register_device(product_model=model), created_by=user, intake_channel="WALK_IN") with patch.object(operations_queries, "capability_map", wraps=operations_queries.capability_map) as resolver: labels = self.navigation_labels(user) resolver.assert_called_once() self.assertIn(("Operations", "Service cases"), labels) url = reverse("operations:list", args=["cases"]) self.assertContains(self.client.get(url), case.job_number) detail = reverse("operations:detail", args=["cases", case.pk]) self.assertEqual(self.client.get(detail).status_code, 200) denied = self.persona("no-cases", self.granted("devices.view_device"), center=self.center) self.assertNotIn(("Operations", "Service cases"), self.navigation_labels(denied)) response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["page"].paginator.count, 0) self.assertNotContains(response, case.job_number) self.assertEqual(self.client.get(detail).status_code, 404) self.assertEqual(self.client.post(detail, {}).status_code, 405) def test_denied_navigation_is_hidden(self): labels = self.navigation_labels(self.scope_reader()) self.assertNotIn(("Parts & inventory", "Inventory locations"), labels, "no inventory capability was granted") self.assertNotIn(("Parts & inventory", "Inventory overview"), labels) self.assertNotIn(("Commercial", "Invoices"), labels) def test_company_scope_does_not_unlock_center_only_navigation(self): # A company-scoped reader may manage customers but has no center queue. user = self.persona("shell-company-nav", self.granted("customers.view_customer")) with patch.object(operations_queries, "capability_map", wraps=operations_queries.capability_map) as resolver: labels = self.navigation_labels(user) resolver.assert_called_once() self.assertIn(("Operations", "Customers"), labels) self.assertNotIn(("Operations", "Service cases"), labels) self.assertNotIn(("Operations", "Walk-in queue"), labels) self.assertEqual(self.client.get(reverse("operations:list", args=["customers"])).status_code, 200) cases = self.client.get(reverse("operations:list", args=["cases"])) self.assertEqual(cases.context["page"].paginator.count, 0) self.assertEqual(self.client.get(reverse("frontdesk:desk", args=[self.center.pk])).status_code, 404) self.assertEqual(self.client.post(reverse("frontdesk:desk", args=[self.center.pk]), {}).status_code, 405) def test_every_rendered_link_requires_a_capability(self): """No navigation entry is emitted without an explicit capability check.""" self.client.force_login(self.scope_reader()) response = self.client.get(reverse("operations:dashboard")) workspace = response.context["workspace"] for group in workspace["groups"]: for link in group["links"]: with self.subTest(link=link["label"]): self.assertTrue(link["url"].startswith("/")) def test_unauthenticated_user_gets_no_workspace(self): response = self.client.get(reverse("operations:dashboard")) self.assertEqual(response.status_code, 302) self.assertIn(reverse("operations:login"), response["Location"]) def test_inactive_user_gets_no_workspace(self): user = self.scope_reader() self.assertTrue(self.navigation_labels(user)) get_user_model().objects.filter(pk=user.pk).update(is_active=False) self.client.force_login(get_user_model().objects.get(pk=user.pk)) response = self.client.get(reverse("operations:dashboard")) self.assertEqual(response.status_code, 302, "an inactive account must not reach the workspace on a remembered session") class ScopeIsolationTests(ShellCapabilityFixture): """D. Cross-company, cross-center and cross-department scope stays correct.""" def test_cross_company_service_center_is_never_authorized(self): user = self.scope_reader() self.assertTrue(is_authorized(user=user, permission="service.view_servicecase", target=self.center)) self.assertFalse(is_authorized(user=user, permission="service.view_servicecase", target=self.other_center)) self.assertFalse(is_authorized(user=user, permission="service.view_servicecase", target=self.other_company)) def test_company_scope_reaches_every_center_in_that_company_only(self): user = self.company_reader() self.assertTrue(is_authorized(user=user, permission="service.view_servicecase", target=self.center)) self.assertFalse(is_authorized(user=user, permission="service.view_servicecase", target=self.other_center)) def test_sibling_center_in_the_same_region_is_not_authorized(self): user, sibling = self.second_center_reader() self.assertTrue(is_authorized(user=user, permission="service.view_servicecase", target=sibling)) self.assertFalse(is_authorized(user=user, permission="service.view_servicecase", target=self.center)) def test_department_scope_authorizes_departments_not_service_centers(self): user = self.department_reader() self.assertTrue(is_authorized(user=user, permission="service.view_servicecase", target=self.department)) self.assertFalse(is_authorized(user=user, permission="service.view_servicecase", target=self.center), "department scope must not reach a service center") def test_department_scope_does_not_leak_to_other_departments(self): user = self.department_reader() self.assertTrue(is_authorized(user=user, permission="service.view_servicecase", target=self.department)) foreign = Department.objects.create(company=self.company, code="SHELL2", name="Other department") self.assertFalse(is_authorized(user=user, permission="service.view_servicecase", target=foreign)) def test_deactivating_a_center_removes_the_capability_immediately(self): user = self.scope_reader() self.assertTrue(operations_queries.capable(user, "service.view_servicecase")) ServiceCenter.objects.filter(pk=self.center.pk).update(is_active=False) fresh = get_user_model().objects.get(pk=user.pk) self.assertFalse(operations_queries.capable(fresh, "service.view_servicecase"), "a capability must be recomputed from current state, never remembered") def test_two_personas_never_share_an_authorization_result(self): """H. Different permissions for different users stay separated.""" capable_user = self.persona("shell-h-capable", self.granted("service.view_servicecase"), center=self.center) denied_user = self.persona("shell-h-denied", self.granted("devices.view_device"), center=self.center) self.assertTrue(operations_queries.capable(capable_user, "service.view_servicecase")) self.assertFalse(operations_queries.capable(denied_user, "service.view_servicecase")) self.assertFalse(operations_queries.capable(capable_user, "devices.view_device")) self.assertTrue(operations_queries.capable(denied_user, "devices.view_device")) def test_authorized_queryset_never_leaks_another_company(self): user = self.company_reader() self.assertEqual( set(authorized_queryset(user=user, permission="service.view_servicecase", queryset=ServiceCenter.objects.all()).values_list("pk", flat=True)), {self.center.pk}) class ShellQueryProfileTests(ShellCapabilityFixture): """E. The shell's own authorization work is bounded and fully attributed.""" def shell_capabilities(self, user): """Every capability the navigation shell resolves for this persona.""" self.client.force_login(user) response = self.client.get(reverse("operations:dashboard")) self.assertEqual(response.status_code, 200) return response.context["workspace"]["capabilities"] or {} def test_shell_resolves_each_capability_at_most_once_per_request(self): """The shell's own memoization must hold for the whole request.""" user = self.company_reader() self.client.force_login(user) with patch.object(operations_queries, "capability_map", wraps=operations_queries.capability_map) as resolver: response = self.client.get(reverse("operations:dashboard")) first = shell(response.wsgi_request) self.assertIs(first, shell(response.wsgi_request)) self.assertEqual(response.status_code, 200) resolver.assert_called_once() actor, permissions = resolver.call_args.args self.assertEqual(actor.pk, user.pk) questions = capability_questions() self.assertEqual(set(permissions), questions[False] | questions[True]) self.assertEqual(len(permissions), len(set(permissions))) capabilities = first["workspace"]["capabilities"] self.assertEqual(set(capabilities), {(permission, mode) for mode in questions for permission in questions[mode]}) self.assertTrue(capabilities["service.view_servicecase", False]) self.assertFalse(capabilities["commercial.view_serviceinvoice", False]) labels = {(group["label"], link["label"]) for group in first["workspace"]["groups"] for link in group["links"]} self.assertIn(("Operations", "Service cases"), labels) self.assertNotIn(("Commercial", "Invoices"), labels) def test_shell_authorization_work_is_bounded_by_two_queries_per_capability(self): """Current true invariant. ``capable()`` probes the company scope and then the service-center scope, so a shell that resolves N distinct capabilities spends at most 2N authorization queries. This asserts that bound; it deliberately does NOT restate or relax the Phase 3D report budgets, which the frozen Phase 3D tests enforce. See the release checklist, Release blocker 1. """ user = self.company_reader() self.client.force_login(user) with CaptureQueriesContext(connection) as captured: response = self.client.get(reverse("operations:dashboard")) self.assertEqual(response.status_code, 200) workspace = response.context["workspace"] capabilities = workspace["capabilities"] authorization = [q for q in captured.captured_queries if "userorganizationassignment" in q["sql"]] # The dashboard also authorizes its own cards, so bound only the shell's # contribution by the number of navigation capabilities it resolved. self.assertLessEqual(len(authorization), 2 * len(capabilities) + 12, f"{len(authorization)} authorization queries for {len(capabilities)} navigation capabilities") def test_shell_capability_memo_is_not_reused_by_a_later_request(self): """G. The per-request memo must not survive the request.""" user = self.company_reader() self.assertTrue(self.shell_capabilities(user)) get_user_model().objects.filter(pk=user.pk).update(is_active=False) self.client.force_login(get_user_model().objects.get(pk=user.pk)) response = self.client.get(reverse("operations:dashboard")) self.assertEqual(response.status_code, 302, "a later request must recompute from current state, not reuse the memo") def test_shell_capability_memo_is_not_shared_between_users(self): """F. The memo belongs to one request of one user.""" first = self.shell_capabilities(self.scope_reader()) second = self.shell_capabilities(self.company_reader()) self.assertIsNot(first, second, "each request must own its capability memo") self.assertTrue(first["service.view_servicecase", False]) self.assertTrue(second["service.view_servicecase", False]) denied = self.persona("shell-memo-denied", self.granted("devices.view_device"), center=self.center) third = self.shell_capabilities(denied) self.assertFalse(third["service.view_servicecase", False], "a persona without the permission must not inherit another persona's capability") self.assertTrue(third["devices.view_device", False]) def test_shell_is_memoized_on_the_request(self): """The shell is built once per request, not once per template render.""" from django.test import RequestFactory request = RequestFactory().get("/") request.user = self.company_reader() first = shell(request) self.assertIs(first, shell(request), "shell() must memoize on the request") self.assertTrue(first["workspace"]["groups"]) def test_shell_memo_is_not_written_for_anonymous_or_admin_requests(self): from django.test import RequestFactory factory = RequestFactory() anonymous = factory.get("/") from django.contrib.auth.models import AnonymousUser anonymous.user = AnonymousUser() self.assertEqual(shell(anonymous), {"workspace": None}) admin_path = factory.get("/admin/") admin_path.user = self.company_reader() self.assertEqual(shell(admin_path), {"workspace": None}) self.assertFalse(hasattr(admin_path, "_operation_shell"), "a non-workspace path must not be memoized either") def test_authorized_queryset_rejects_unsupported_shapes_and_cannot_be_reused_as_a_scope(self): """The authorization boundary still fails closed for combinators.""" user = self.company_reader() rows = authorized_queryset(user=user, permission="service.view_servicecase", queryset=ServiceCenter.objects.all()) self.assertFalse(authorized_queryset(user=user, permission="service.view_servicecase", queryset=rows.union(rows)).exists(), "a combined queryset is not a supported authorization input")