"""Phase 5D.3: report pages read the same evidence in fewer round trips. Every optimized read is compared with a verbatim copy of the code it replaced: * ``access()`` vs the legacy ``accessible()`` view/export probes; * ``service_rates()`` / the service card vs the legacy ``.aggregate()`` / ``.count()``; * ``page_of()`` vs ``Paginator.get_page()``; * ``configuration.navigation.visible()`` vs its previous boolean expression. The frozen Phase 3D budgets themselves live, unchanged, in ``apps.reporting.tests.test_reports`` and ``tests.test_phase3d_audit``. """ import itertools from datetime import timedelta from types import SimpleNamespace from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser, Permission from django.db import connection from django.db.models import Count, Q, Sum from django.http import Http404 from django.core.paginator import Paginator from django.test import RequestFactory, TestCase from django.test.utils import CaptureQueriesContext from django.utils import timezone from django.urls import reverse from apps.access.models import Role from apps.access.services import create_role_assignment, set_role_permissions from apps.commercial.test_payment import PaymentFixture, shared_payment_data from apps.configuration import navigation as config_navigation from apps.operations import context as shell_context 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.reporting import permissions as p from apps.reporting import service_analytics as service from apps.reporting.filters import period from apps.reporting.query import combined_exists, combined_totals, totals from apps.reporting.scope import centers, companies from apps.reporting.tests.test_reports import grant_report from apps.reporting.views import ANALYTICAL_FILTERS, PER_PAGE, REPORTS, access, page_of from apps.service import test_quality_control as qc, test_repair as repair # --- Verbatim legacy oracles (pre-Phase 5D.3) --------------------------------- def legacy_accessible(user, permission, export=False): return centers(user, permission, export=export).exists() or companies(user, permission, export=export).exists() def legacy_service_rates(user, filters, *, permission=p.SERVICE, export=False): repair_rows = period(service.repairs(user, filters, permission=permission, export=export).filter(status="COMPLETED"), "completed_at", filters) repair_counts = repair_rows.aggregate(denominator=Count("pk"), numerator=Count("pk", filter=Q(outcome="REPAIRED"))) qc_rows = period(service.quality_controls(user, filters, permission=permission, export=export).filter(status="COMPLETED", first_completed_attempt=True), "completed_at", filters) first_pass = qc_rows.aggregate(denominator=Count("pk"), numerator=Count("pk", filter=Q(outcome="PASSED"))) started = period(service.repairs(user, filters, permission=permission, export=export), "started_at", filters) rework = started.aggregate(denominator=Count("pk"), numerator=Count("pk", filter=Q(after_failed_qc=True))) result = [] for label, counts in (("Repair success (completion period)", repair_counts), ("First-pass QC yield (first completion period)", first_pass), ("Repair attempts after failed QC (start period)", rework)): result.append(dict(metric=label, **counts, percent=100 * counts["numerator"] / counts["denominator"] if counts["denominator"] else None)) return result def legacy_visible(user, can): if not user.is_authenticated or not user.is_active: return False return (config_navigation.system_administrator(user) or can("sla.manage_slapolicy", True) or can("communications.manage_templates", True) or can("frontdesk.manage_slots") or (user.is_staff and user.has_perm("inventory.change_inventorylocation") and can("inventory.manage_inventory") and can("inventory.view_stock"))) def date_filters(): today = timezone.localdate() yield "none", {} yield "today", {"date_from": today, "date_to": today} yield "until-yesterday", {"date_to": today - timedelta(days=1)} yield "from-tomorrow", {"date_from": today + timedelta(days=1)} yield "wide", {"date_from": today - timedelta(days=30), "date_to": today + timedelta(days=30)} class ReportingRoundTripEquivalence(PaymentFixture, TestCase): """Completed service, diagnosis, repair, QC, invoice and payment history.""" @classmethod def setUpTestData(cls): shared_payment_data(cls) User = get_user_model() cls.other_company, cls.other_region, cls.other_center, _ = make_tree("RTOTHER") cls.sibling = ServiceCenter.objects.create(company=cls.company, region=cls.center.region, code="RTSIB", name="Sibling", center_type="OWN") cls.department = Department.objects.create(company=cls.company, code="RTDEPT", name="Reporting department") everything = [code for code, _ in p.PERMISSIONS] no_export = [code for code in everything if code != "export_reports"] cls.personas = {} def persona(name, **grant): user = User.objects.create_user(username="rt-" + name, is_staff=True) if grant: grant_report(user, **grant) cls.personas[name] = user persona("center", company=cls.company, center=cls.center) persona("center-no-export", company=cls.company, center=cls.center, permissions=no_export) persona("sibling-center", company=cls.company, center=cls.sibling) persona("region", company=cls.company, region=cls.center.region) persona("company", company=cls.company) persona("department", company=cls.company, department=cls.department) persona("other-company", company=cls.other_company, center=cls.other_center) persona("service-only", company=cls.company, center=cls.center, permissions=["view_service_analytics"]) persona("nothing") cls.personas["superuser"] = User.objects.create_superuser(username="rt-superuser") def test_combined_access_probe_equals_legacy_view_and_export_probes(self): checked = 0 for name, user in self.personas.items(): for section, (_, permission, _) in REPORTS.items(): with self.subTest(persona=name, section=section): self.assertEqual(access(user, permission), (legacy_accessible(user, permission), legacy_accessible(user, permission, export=True))) checked += 1 self.assertEqual(checked, len(self.personas) * len(REPORTS)) def test_access_is_one_round_trip(self): with CaptureQueriesContext(connection) as captured: access(self.personas["center"], p.SERVICE) self.assertEqual(len(captured), 1) def test_combined_exists_equals_each_exists_including_empty(self): user = self.personas["center"] probes = {"centers": centers(user, p.SERVICE), "companies": companies(user, p.SERVICE), "empty": centers(user, p.SERVICE).filter(pk=None), "other": centers(self.personas["other-company"], p.SERVICE).filter(company=self.company)} self.assertEqual(combined_exists(**probes), {name: rows.exists() for name, rows in probes.items()}) def test_combined_totals_equal_aggregate_for_counts_sums_and_empty_sets(self): from apps.commercial.models import ServicePayment self.pay(amount="40") self.pay(amount="15") rows = ServicePayment.objects.all() reads = { "all": (rows, dict(n=Count("pk"), gross=Sum("amount"))), "posted": (rows, dict(posted=Count("pk", filter=Q(status="POSTED")), total=Sum("amount", default=0))), "empty": (rows.filter(pk=None), dict(n=Count("pk"), gross=Sum("amount"), zero=Sum("amount", default=0))), } combined = combined_totals(**{name: totals(qs, **metrics) for name, (qs, metrics) in reads.items()}) self.assertEqual(combined, {name: qs.aggregate(**metrics) for name, (qs, metrics) in reads.items()}) self.assertEqual(combined["all"]["n"], 2) self.assertIsNone(combined["empty"]["gross"]) def test_service_rates_equal_legacy_for_every_persona_and_date_boundary(self): for name, user in self.personas.items(): for label, filters in date_filters(): for extra in ({}, {"service_center": self.center.pk}, {"company": self.other_company.pk}, {"engineer": self.engineer.pk}): combined = {**filters, **extra} with self.subTest(persona=name, period=label, extra=extra): self.assertEqual(service.service_rates(user, combined), legacy_service_rates(user, combined)) def test_service_page_card_rates_rows_and_export_link_equal_legacy(self): for name, user in self.personas.items(): self.client.force_login(user) for table in ("complaints_complaint", "diagnoses", "diagnosis_records", "cooccurrence_diagnosis_root_cause"): for label, filters in list(date_filters())[:4]: with self.subTest(persona=name, table=table, period=label): response = self.client.get("/reports/service/", {"table": table, **{k: str(v) for k, v in filters.items()}}) viewable = legacy_accessible(user, p.SERVICE) if not viewable: self.assertEqual(response.status_code, 403) continue self.assertEqual(response.status_code, 200) rate_filters = {k: v for k, v in filters.items() if k not in ANALYTICAL_FILTERS} self.assertEqual(response.context["rates"], legacy_service_rates(user, rate_filters)) self.assertEqual(response.context["cards"], {"Distinct performed actions": service.performed_actions(user, filters).count()}) self.assertEqual(response.context["can_export"], legacy_accessible(user, p.SERVICE, export=True)) selected = next(t for t in service.service_tables(user, filters) if t.key == table) legacy_page = Paginator(selected.rows, PER_PAGE).get_page(1) self.assertEqual(response.context["page"].paginator.count, legacy_page.paginator.count) self.assertEqual(response.context["rows"], list(selected.values(list(legacy_page.object_list)))) def test_every_section_page_matches_legacy_count_rows_and_access(self): tables = {"operational": "cases", "inventory": "positions", "commercial": "invoices", "management": "workflow", "service": "diagnoses"} for name, user in self.personas.items(): self.client.force_login(user) for section, table in tables.items(): _, permission, builder = REPORTS[section] with self.subTest(persona=name, section=section): response = self.client.get(f"/reports/{section}/", {"table": table}) if not legacy_accessible(user, permission): self.assertEqual(response.status_code, 403) continue self.assertEqual(response.status_code, 200) selected = next(t for t in builder(user, {}, permission=permission) if t.key == table) legacy_page = Paginator(selected.rows, PER_PAGE).get_page(1) self.assertEqual(response.context["page"].paginator.count, legacy_page.paginator.count) self.assertEqual(response.context["rows"], list(selected.values(list(legacy_page.object_list)))) self.assertEqual(response.context["can_export"], legacy_accessible(user, permission, export=True)) if section == "management": self.assertEqual(response.context["rates"], legacy_service_rates(user, {}, permission=permission)) def test_page_of_equals_paginator_for_every_table_of_every_builder(self): from apps.reporting.management_reports import management_tables from apps.reporting.service_dashboard import dashboard_tables from apps.reporting.inventory_analytics import inventory_tables from apps.reporting.commercial_analytics import commercial_tables self.pay(amount="25") checked = 0 for user in (self.actor, self.personas["center"], self.personas["other-company"]): for builder in (dashboard_tables, service.service_tables, inventory_tables, commercial_tables, management_tables): for report in builder(user, {}): with self.subTest(user=user.username, report=report.key): legacy = Paginator(report.rows, PER_PAGE).get_page(1) page = page_of(report.rows, 1) self.assertEqual(page.paginator.count, legacy.paginator.count) self.assertEqual(list(page.object_list), list(legacy.object_list)) checked += 1 self.assertGreater(checked, 150) def test_scope_isolation_is_preserved(self): center = self.personas["center"] other = self.personas["other-company"] sibling = self.personas["sibling-center"] self.assertGreater(service.performed_actions(center, {}).count(), 0) for outsider in (other, sibling): with self.subTest(outsider=outsider.username): self.client.force_login(outsider) response = self.client.get("/reports/service/", {"table": "diagnoses"}) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["cards"], {"Distinct performed actions": 0}) self.assertEqual(response.context["page"].paginator.count, 0) self.assertTrue(all(rate["denominator"] == 0 for rate in response.context["rates"])) def test_export_requires_export_permission_and_unauthorized_is_denied(self): self.client.force_login(self.personas["center-no-export"]) page = self.client.get("/reports/service/", {"table": "diagnoses"}) self.assertEqual(page.status_code, 200) self.assertFalse(page.context["can_export"]) self.assertNotContains(page, "Export this scoped report as CSV") self.assertEqual(self.client.get("/reports/service/", {"table": "diagnoses", "export": "csv"}).status_code, 403) self.client.force_login(self.personas["center"]) page = self.client.get("/reports/service/", {"table": "diagnoses"}) self.assertTrue(page.context["can_export"]) self.assertContains(page, "Export this scoped report as CSV") self.assertEqual(self.client.get("/reports/service/", {"table": "diagnoses", "export": "csv"}).status_code, 200) self.client.force_login(self.personas["service-only"]) for section in ("operational", "inventory", "commercial", "management"): with self.subTest(section=section): self.assertEqual(self.client.get(f"/reports/{section}/").status_code, 403) self.client.force_login(self.personas["nothing"]) for section in REPORTS: with self.subTest(section=section): self.assertEqual(self.client.get(f"/reports/{section}/").status_code, 403) self.assertEqual(self.client.get(f"/reports/{section}/", {"export": "csv"}).status_code, 403) class ReworkRatesEquivalence(TestCase): """Non-trivial ratios: one failed QC, a rework attempt and a passing QC.""" @classmethod def setUpTestData(cls): fixture = type("Fixture", (), {})() qc.setup_qc(fixture) first = qc.prepared(fixture) qc.check(fixture, first, result="FAIL") qc.fail_qc(fixture, first) fixture.case.refresh_from_db() fixture.rework, _ = repair.prepared(fixture) repair.complete(fixture, fixture.rework) second = qc.prepared(fixture) qc.fill(fixture, second) qc.pass_qc(fixture, second) fixture.reader = get_user_model().objects.create_superuser(username="rt-rework-reader") for key, value in vars(fixture).items(): setattr(cls, key, value) def test_rates_equal_legacy_and_keep_their_meaning(self): rates = service.service_rates(self.reader, {}) self.assertEqual(rates, legacy_service_rates(self.reader, {})) self.assertEqual([r["percent"] for r in rates], [100, 0, 50]) self.assertEqual([(r["numerator"], r["denominator"]) for r in rates], [(2, 2), (0, 1), (1, 2)]) for label, filters in date_filters(): with self.subTest(period=label): self.assertEqual(service.service_rates(self.reader, filters), legacy_service_rates(self.reader, filters)) def test_service_page_reads_card_and_rates_in_one_round_trip(self): self.client.force_login(self.reader) legacy = legacy_service_rates(self.reader, {}) with CaptureQueriesContext(connection) as captured: response = self.client.get("/reports/service/", {"table": "actions"}) self.assertEqual(response.context["rates"], legacy) rate_reads = [q for q in captured if "service_servicerepairexecution" in q["sql"] and "CROSS JOIN" in q["sql"]] self.assertEqual(len(rate_reads), 1) class PageOfEquivalence(TestCase): """page_of() returns exactly what Paginator.get_page() returns, or 404s where the view did.""" def test_page_of_matches_paginator_for_every_size_and_page(self): ordered = Permission.objects.order_by("pk") ids = list(ordered.values_list("pk", flat=True)) self.assertGreater(len(ids), 2 * PER_PAGE + 1) for size in (0, 1, PER_PAGE - 1, PER_PAGE, PER_PAGE + 1, 2 * PER_PAGE, 2 * PER_PAGE + 1): rows = ordered.filter(pk__in=ids[:size]).values("pk", "codename") legacy = Paginator(rows, PER_PAGE) for number in range(1, legacy.num_pages + 3): with self.subTest(size=size, page=number): expected = legacy.get_page(number) if expected.number != number: with self.assertRaises(Http404): page_of(rows, number) continue page = page_of(rows, number) self.assertEqual((page.number, page.paginator.count, page.paginator.num_pages, page.has_next(), page.has_previous()), (expected.number, expected.paginator.count, expected.paginator.num_pages, expected.has_next(), expected.has_previous())) self.assertEqual(list(page.object_list), list(expected.object_list)) def test_rows_and_total_are_one_query_at_every_size(self): ordered = Permission.objects.order_by("pk") ids = list(ordered.values_list("pk", flat=True)) for size, number in ((0, 1), (10, 1), (PER_PAGE, 1), (PER_PAGE + 1, 1), (PER_PAGE + 1, 2), (2 * PER_PAGE + 1, 3)): rows = ordered.filter(pk__lte=ids[size - 1]) if size else ordered.filter(pk__lt=0) with self.subTest(size=size, page=number), CaptureQueriesContext(connection) as captured: page_of(rows.values("pk", "codename"), number) self.assertEqual(len(captured), 1) def test_extreme_page_is_404_and_never_reaches_postgresql_as_an_offset(self): from apps.reporting.views import MAX_OFFSET rows = Permission.objects.order_by("pk").values("pk") last_safe = MAX_OFFSET // PER_PAGE + 1 # offset <= bigint max: harmless empty read for number in (last_safe + 1, 2 ** 63, 2 ** 64, int("9" * 4000)): with self.subTest(digits=len(str(number))), CaptureQueriesContext(connection) as captured: with self.assertRaises(Http404): page_of(rows, number) self.assertEqual(len(captured), 0) with CaptureQueriesContext(connection) as captured, self.assertRaises(Http404): page_of(rows, last_safe) self.assertEqual(len(captured), 1) # The connection is still usable: no aborted transaction was left behind. self.assertTrue(Permission.objects.exists()) def test_grouped_rows_total_counts_groups_not_underlying_rows(self): rows = Permission.objects.order_by().values("content_type__app_label").annotate(count=Count("pk")).order_by("content_type__app_label") page = page_of(rows, 1) self.assertEqual(page.paginator.count, rows.count()) self.assertEqual(list(page.object_list), list(rows[:PER_PAGE])) class ConfigurationVisibility(TestCase): """Phase 5D.2 declared the SLA policy capability as 'sla.manage_slopolicy', which does not exist.""" def test_every_declared_shell_capability_is_a_real_permission(self): known = {f"{app}.{codename}" for app, codename in Permission.objects.values_list("content_type__app_label", "codename")} declared = set(config_navigation.CONFIGURATION_CAPABILITIES) for mode in shell_context.capability_questions().values(): declared |= mode self.assertEqual(sorted(declared - known), []) def test_visible_equals_previous_expression_for_every_input(self): capability_keys = [("sla.manage_slapolicy", True), ("communications.manage_templates", True), ("frontdesk.manage_slots", False), ("inventory.manage_inventory", False), ("inventory.view_stock", False)] for bits in itertools.product((False, True), repeat=len(capability_keys) + 4): authenticated, staff, superuser, has_perm = bits[:4] granted = dict(zip(capability_keys, bits[4:])) user = SimpleNamespace(is_authenticated=authenticated, is_active=True, is_staff=staff, is_superuser=superuser, has_perm=lambda permission, has_perm=has_perm: has_perm) can = lambda permission, company_only=False, granted=granted: granted.get((permission, company_only), False) with self.subTest(bits=bits): self.assertEqual(config_navigation.visible(SimpleNamespace(user=user), can), legacy_visible(user, can)) def test_sla_policy_manager_sees_sla_policies_and_setup_in_the_shell(self): company, region, center, _ = make_tree("RTSLA") user = get_user_model().objects.create_user(username="rt-sla-manager") role = Role.objects.create(code="RT-SLA", name="SLA policy manager") set_role_permissions(role=role, permissions=list(Permission.objects.filter(content_type__app_label="sla", codename="manage_slapolicy"))) create_role_assignment(user=user, role=role, organization_assignment=create_assignment(user=user, company=company)) from apps.sla.services import save_policy policy = save_policy(actor=user, company=company, code="rc-policy", name="Scoped RC policy", effective_from=timezone.localdate(), target_minutes=60, warning_minutes=10) request = RequestFactory().get("/reports/") request.user = user with patch.object(shell_context.q, "capability_map", wraps=shell_context.q.capability_map) as resolver: workspace = shell_context.shell(request)["workspace"] resolver.assert_called_once() groups = {group["label"]: {link["label"]: link["url"] for link in group["links"]} for group in workspace["groups"]} policies = reverse("sla:policies") setup = reverse("configuration:index") self.assertIn("System", groups) self.assertIn("SLA policies", groups["System"]) self.assertIn("Administration / Settings (Configuration center)", groups["System"]) self.assertEqual(groups["System"]["SLA policies"], policies) self.assertEqual(groups["System"]["Administration / Settings (Configuration center)"], setup) self.client.force_login(user) for url in (policies, setup): self.assertEqual(self.client.get(url).status_code, 200) self.assertContains(self.client.get(policies), policy.name) edit = reverse("sla:policy_edit", args=[policy.pk]) self.assertEqual(self.client.get(edit).status_code, 200) # A valid center grant cannot satisfy this company-only capability. center_user = get_user_model().objects.create_user(username="rt-sla-center") create_role_assignment(user=center_user, role=role, organization_assignment=create_assignment( user=center_user, company=company, region=region, service_center=center)) no_permission = get_user_model().objects.create_user(username="rt-sla-denied") for denied in (center_user, no_permission): denied_request = RequestFactory().get("/reports/") denied_request.user = denied denied_urls = {link["url"] for group in shell_context.shell(denied_request)["workspace"]["groups"] for link in group["links"]} self.assertNotIn(policies, denied_urls) self.assertNotIn(setup, denied_urls) self.client.force_login(denied) # The list is a scoped empty 200; individual policy access is 404. response = self.client.get(policies) self.assertEqual(response.status_code, 200) self.assertFalse(response.context["policies"].exists()) self.assertNotContains(response, policy.name) self.assertEqual(self.client.get(setup).status_code, 403) self.assertEqual(self.client.get(edit).status_code, 404) self.assertEqual(self.client.post(edit, {}).status_code, 404) anonymous = RequestFactory().get("/reports/") anonymous.user = AnonymousUser() self.assertEqual(shell_context.shell(anonymous), {"workspace": None})