"""Reporting reconciliation against histories produced by frozen services.""" from datetime import timedelta from decimal import Decimal from uuid import uuid4 from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission, Group from django.db import connection from django.test import TestCase from django.test.utils import CaptureQueriesContext from django.utils import timezone from apps.commercial.test_payment import PaymentFixture, shared_payment_data from apps.inventory import tests as inventory_fixture from apps.access.models import Role, UserRoleAssignment from apps.organization.assignment_services import create_assignment from apps.organization.test_lifecycle import make_tree from apps.service import test_handover as handover_fixture from apps.service.models import ServiceCase from apps.reporting import permissions as p from apps.reporting import service_dashboard as dashboard from apps.reporting import service_analytics as service from apps.reporting import inventory_analytics as inventory from apps.reporting import commercial_analytics as commercial from apps.reporting.management_reports import management_tables from apps.reporting.filters import ReportFilters, period from apps.reporting.exports import safe_cell from apps.reporting.scope import cases def grant_report(user, company, center=None, permissions=None, region=None, department=None): assignment = create_assignment(user=user, company=company, region=center.region if center else region, service_center=center, department=department) role = Role.objects.create(code="R-" + uuid4().hex, name="Reporting") names = permissions or [code for code, _ in p.PERMISSIONS] role.permissions.set(Permission.objects.filter(content_type__app_label="reporting", codename__in=names)) UserRoleAssignment.objects.create(user=user, role=role, organization_assignment=assignment) return assignment, role class ReportingTests(PaymentFixture, TestCase): @classmethod def setUpTestData(cls): shared_payment_data(cls) cls.reader = get_user_model().objects.create_user(username="report-reader", is_staff=True) cls.reader_assignment, cls.reader_role = grant_report(cls.reader, cls.company, cls.center) def test_all_report_queries_execute_on_real_completed_history(self): for builder in (dashboard.dashboard_tables, service.service_tables, inventory.inventory_tables, commercial.commercial_tables, management_tables): for report in builder(self.actor, {}): with self.subTest(report=report.key, builder=builder.__name__): self.assertIsInstance(list(report.rows[:2]), list) self.assertEqual(len(dashboard.turnaround(self.actor, {})), 6) self.assertEqual(commercial.quotation_approval_turnaround(self.actor, {})["count"], 1) def test_open_count_reconciles_current_cases(self): result = dashboard.dashboard(self.reader, {}) self.assertEqual(result["current_open"], ServiceCase.objects.exclude(status__in=dashboard.TERMINAL).count()) self.assertEqual(result["received_in_period"], 1) def test_consumption_reconciles_posted_evidence(self): from apps.inventory.models import PartsDisposition report = next(t for t in inventory.inventory_tables(self.reader, {}) if t.key == "usage_parts") row = report.rows.get() self.assertEqual(row["consumed"], PartsDisposition.objects.get(kind="CONSUMED").quantity) self.assertEqual(row["net_consumption"], row["consumed"]) self.assertEqual(row["returned_unused"], 0) def test_current_inventory_reconciles_frozen_queries(self): from apps.inventory.queries import stock_on_hand, available_stock for row in inventory.positions(self.actor, {}): self.assertEqual(row.on_hand, stock_on_hand(actor=self.actor, location=row.location, spare_part=row.spare_part)) self.assertEqual(row.available, available_stock(actor=self.actor, location=row.location, spare_part=row.spare_part)) self.assertEqual(row.serialized_units + row.nonserialized_quantity, row.on_hand) def test_partial_payment_reconciles_current_debt(self): self.pay(amount="40") row = commercial.settlements(self.reader, {}).get() self.assertEqual((row.paid_amount, row.balance_due, row.settlement_state, row.clearance), (40, 60, "PARTIALLY_PAID", "UNCLEARED")) total = commercial.collection_totals(self.reader, {}).get() self.assertEqual((total["received"], total["net_collected"], total["valid_allocations"], total["receipts"]), (40, 40, 40, 1)) def test_reversal_preserves_receipt_but_removes_settlement(self): payment = self.pay() self.reverse_payment(payment) result = commercial.collection_totals(self.reader, {}).get() self.assertEqual((result["received"], result["reversed"], result["net_collected"], result["valid_allocations"], result["receipts"]), (100, 100, 0, 0, 1)) row = commercial.settlements(self.reader, {}).get() self.assertEqual((row.balance_due, row.settlement_state), (100, "UNPAID")) def test_due_release_delivery_retains_debt(self): self.release_due() self.deliver() row = commercial.settlements(self.reader, {}).get() self.assertEqual((row.clearance, row.balance_due), ("DUE_RELEASE", 100)) report = next(t for t in commercial.commercial_tables(self.reader, {}) if t.key == "outstanding_summary") self.assertEqual(report.rows.get()["delivered_due"], 1) def test_paid_delivered_closed_history(self): self.pay() self.return_unused(self.issue_row) self.deliver() self.case.refresh_from_db() handover_fixture.close(self) self.assertEqual(commercial.settlements(self.reader, {}).get().clearance, "PAID") counts = dashboard.dashboard(self.reader, {}) self.assertEqual((counts["current_open"], counts["closed_in_period"], counts["delivered_in_period"]), (0, 1, 1)) self.assertEqual(dashboard.turnaround(self.reader, {})[-1]["count"], 1) def test_unused_return_does_not_reduce_consumption(self): self.return_unused(self.issue_row) row = next(t for t in inventory.inventory_tables(self.reader, {}) if t.key == "usage_parts").rows.get() self.assertEqual((row["consumed"], row["returned_unused"], row["net_consumption"]), (1, 1, 1)) def test_same_job_number_at_two_centers_does_not_merge_usage(self): from apps.service import tests as intake, test_diagnosis as diagnosis, test_engineer_assignment as engineer from apps.inventory import services as stock original = self.case self.case = intake.intake(self, service_center=self.center2) self.assertEqual(original.job_number, self.case.job_number) path = create_assignment(user=self.engineer, company=self.company) UserRoleAssignment.objects.create(user=self.engineer, role=self.role, organization_assignment=path) self.assignment = engineer.assign(self) assessment = diagnosis.begin(self) diagnosis.add(self, assessment) self.assessment = diagnosis.complete(self, assessment) self.location = stock.create_location(actor=self.actor, company=self.company, service_center=self.center2, code="SECOND-REPORT", name="Second report store", location_type="STORE") self.billed() report = next(t for t in inventory.inventory_tables(self.actor, {}) if t.key == "usage_case") rows = list(report.rows) self.assertEqual(len(rows), 2) self.assertEqual({row["issue__service_case_id"] for row in rows}, {original.pk, self.case.pk}) self.assertEqual({row["consumed"] for row in rows}, {1}) def test_payment_dates_are_independent_of_invoice_cohort(self): self.pay() tomorrow = timezone.localdate() + timedelta(days=1) self.assertFalse(commercial.payments(self.reader, {"date_from": tomorrow}).exists()) self.assertEqual(commercial.settlements(self.reader, {"date_from": tomorrow}).count(), 1) def test_payment_received_period_does_not_use_finalized_date(self): future = timezone.now() + timedelta(days=2) with patch("apps.commercial.payment_services.timezone.now", return_value=future): self.pay() dates = {"date_from": timezone.localdate(future), "date_to": timezone.localdate(future)} self.assertEqual(commercial.payments(self.reader, dates).count(), 1) invoice_report = next(t for t in commercial.commercial_tables(self.reader, dates) if t.key == "invoices") self.assertEqual(invoice_report.rows.count(), 0) def test_later_reversal_updates_received_cohort_without_double_subtraction(self): payment = self.pay() today = timezone.localdate() future = timezone.now() + timedelta(days=2) with patch("apps.commercial.payment_services.timezone.now", return_value=future): self.reverse_payment(payment) dates = {"date_from": today, "date_to": today} totals = commercial.collection_totals(self.reader, dates).get() self.assertEqual((totals["received"], totals["reversed"], totals["net_collected"]), (100, 100, 0)) reversals = next(t for t in commercial.commercial_tables(self.reader, dates) if t.key == "reversal_events") self.assertEqual(reversals.rows.count(), 0) def test_invoice_model_snapshot_survives_current_master_rename(self): original = self.invoice.context_snapshot["quotation"]["model"] self.model.name = "Renamed after finalization" self.model.save(update_fields=["name"]) report = next(t for t in commercial.commercial_tables(self.reader, {}) if t.key == "invoice_model") self.assertEqual(report.rows.get()["context_snapshot__quotation__model"], original) def test_reporting_permission_anchor_has_no_table(self): self.assertNotIn("reporting_reportaccess", connection.introspection.table_names()) self.assertEqual(Permission.objects.filter(content_type__app_label="reporting").count(), 6) def test_reports_reject_business_post_requests(self): self.client.force_login(self.reader) self.assertEqual(self.client.post("/reports/", {"status": "CLOSED"}).status_code, 405) def test_current_queue_ignores_event_period(self): future = {"date_from": timezone.localdate() + timedelta(days=1)} counts = dashboard.dashboard(self.reader, future) self.assertEqual((counts["current_open"], counts["received_in_period"]), (1, 0)) def test_intake_complaint_uses_device_category_not_applicability(self): from apps.service_catalog.models import ComplaintSymptomProductCategory from apps.service.tests import intake from apps.service.services import add_service_case_complaint case = intake(self) add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) rows = service.complaint_frequency(self.reader, {}, dimension="product_category") self.assertTrue(rows.exists()) self.assertEqual({r["dimension_id"] for r in rows}, {self.model.category_id}) sql = str(rows.query).lower() self.assertNotIn(ComplaintSymptomProductCategory._meta.db_table.lower(), sql) self.assertNotIn("servicecategory", sql) def test_unknown_cause_is_reporting_label_without_master(self): from apps.service_catalog.models import RootCause before = RootCause.objects.count() row = service.repair_action_diagnosis_cooccurrence(self.reader, {}).get() self.assertIsNone(row["root_cause_id"]) self.assertEqual(row["root_cause_label"], "Unknown / Unconfirmed") self.assertEqual(row["performed_actions"], 1) self.assertEqual(RootCause.objects.count(), before) def test_anonymous_redirect_and_authenticated_unauthorized_denial(self): self.assertEqual(self.client.get("/reports/").status_code, 302) user = get_user_model().objects.create_user(username="no-reports", is_staff=True) self.client.force_login(user) self.assertEqual(self.client.get("/reports/").status_code, 403) def test_direct_reporting_login_defaults_to_dashboard(self): self.reader.set_password("synthetic-report-login-only") self.reader.save(update_fields=["password"]) response = self.client.post("/reports/login/", {"username": self.reader.username, "password": "synthetic-report-login-only"}) self.assertRedirects(response, "/reports/") def test_groups_direct_permissions_and_staff_do_not_grant_scope(self): user = get_user_model().objects.create_user(username="native-only", is_staff=True) permissions = Permission.objects.filter(content_type__app_label="reporting") user.user_permissions.set(permissions) group = Group.objects.create(name="Native reports") group.permissions.set(permissions) user.groups.add(group) self.assertFalse(cases(user, p.OPERATIONAL, {}).exists()) def test_wrong_company_and_forged_case_filters_remain_empty(self): company, _, center, _ = make_tree("REPORT-FOREIGN") user = get_user_model().objects.create_user(username="foreign-report") grant_report(user, company, center) self.assertFalse(cases(user, p.OPERATIONAL, {"company": self.company.pk, "case": self.case.pk}).exists()) self.client.force_login(user) response = self.client.get("/reports/", {"case": str(self.case.pk)}) self.assertEqual(response.status_code, 200) self.assertNotContains(response, self.case.job_number) def test_permission_scope_split_does_not_combine(self): company, _, center, _ = make_tree("REPORT-SPLIT") user = get_user_model().objects.create_user(username="split-report") grant_report(user, company, center, ["view_operational_dashboard"]) grant_report(user, self.company, self.center, ["view_service_analytics"]) self.assertFalse(cases(user, p.OPERATIONAL, {}).exists()) self.assertTrue(cases(user, p.SERVICE, {}).exists()) def test_inactive_path_denied(self): from apps.organization.assignment_services import deactivate_assignment deactivate_assignment(assignment=self.reader_assignment) self.assertFalse(cases(self.reader, p.OPERATIONAL, {}).exists()) def test_active_superuser_and_database_fresh_inactive_superuser(self): self.assertTrue(cases(self.actor, p.OPERATIONAL, {}).exists()) get_user_model().objects.filter(pk=self.actor.pk).update(is_active=False) self.assertFalse(cases(self.actor, p.OPERATIONAL, {}).exists()) def test_export_requires_report_and_export_scope_intersection(self): company, _, center, _ = make_tree("REPORT-EXPORT") user = get_user_model().objects.create_user(username="export-split") grant_report(user, self.company, self.center, ["view_operational_dashboard"]) grant_report(user, company, center, ["export_reports"]) self.client.force_login(user) self.assertEqual(self.client.get("/reports/", {"export": "csv"}).status_code, 403) def test_csv_and_screen_reconcile_without_personal_details(self): self.client.force_login(self.reader) response = self.client.get("/reports/", {"export": "csv"}) self.assertEqual(response.status_code, 200) content = b"".join(response.streaming_content).decode("utf-8-sig") self.assertIn(self.case.job_number, content) self.assertNotIn("recipient_mobile", content) self.assertNotIn("customer_name", content) def test_query_string_manipulation_rejected(self): self.client.force_login(self.reader) for url in ("/reports/?company=bad", "/reports/?department=" + str(uuid4()), "/reports/?status=CLOSED&status=RECEIVED", "/reports/?export=xlsx", "/reports/?page=-1", "/reports/?date_from=2026-10-01&date_to=2026-09-01"): with self.subTest(url=url): self.assertEqual(self.client.get(url).status_code, 400) self.assertEqual(self.client.get("/reports/?table=forged").status_code, 404) def test_navigation_preserves_filters_and_labels_unsupported_resets(self): from urllib.parse import urlsplit, parse_qs self.client.force_login(self.reader) response = self.client.get("/reports/", {"company": str(self.company.pk), "model": str(self.model.pk), "date_from": "2025-01-01"}) for label, link in response.context["sections"]: query = parse_qs(urlsplit(link).query) self.assertEqual(query["company"], [str(self.company.pk)]) self.assertEqual(query["date_from"], ["2025-01-01"]) if "inventory" in link: self.assertIn("clears model", label) self.assertNotIn("model", query) else: self.assertEqual(query["model"], [str(self.model.pk)]) def test_unsupported_inventory_product_filter_rejected(self): self.client.force_login(self.reader) self.assertEqual(self.client.get("/reports/inventory/", {"model": str(self.model.pk)}).status_code, 400) def test_representative_pages_have_bounded_queries(self): self.client.force_login(self.reader) checks = (("operational", "cases", 24), ("operational", "engineer_queue", 24), ("service", "complaints_complaint", 10), ("service", "diagnoses", 10), ("inventory", "positions", 8), ("commercial", "invoices", 10), ("commercial", "outstanding", 10)) for section, key, budget in checks: with self.subTest(report=key), CaptureQueriesContext(connection) as queries: response = self.client.get(f"/reports/{section}/", {"table": key}) self.assertEqual(response.status_code, 200) self.assertLessEqual(len(queries), budget, (key, len(queries))) print(f"Reporting query count {key}: {len(queries)} (budget {budget})") def test_diagnosis_engineer_invoice_outstanding_query_counts_do_not_grow(self): from apps.service import tests as intake, test_diagnosis as diagnosis, test_engineer_assignment as engineer self.client.force_login(self.reader) reports = [("service", "diagnoses"), ("operational", "engineer_queue"), ("commercial", "invoices"), ("commercial", "outstanding")] counts = [] for section, key in reports: with CaptureQueriesContext(connection) as captured: self.assertEqual(self.client.get(f"/reports/{section}/", {"table": key}).status_code, 200) counts.append(len(captured)) self.case = intake.intake(self) self.assignment = engineer.assign(self) assessment = diagnosis.begin(self) diagnosis.add(self, assessment) self.assessment = diagnosis.complete(self, assessment) row = self.billed() self.ready() self.invoice = self.finalize(row) for index, (section, key) in enumerate(reports): with CaptureQueriesContext(connection) as captured: response = self.client.get(f"/reports/{section}/", {"table": key}) self.assertEqual(response.status_code, 200) self.assertEqual(len(captured), counts[index], key) def test_read_only_reports_and_exports(self): from django.apps import apps labels = {"service", "inventory", "commercial", "devices", "customers", "service_catalog"} def snapshot(): return {model._meta.label: list(model.objects.order_by("pk").values()) for model in apps.get_models() if model._meta.app_label in labels} before = snapshot() self.client.force_login(self.reader) for section in ("operational", "service", "inventory", "commercial", "management"): response = self.client.get(f"/reports/{section}/") self.assertEqual(response.status_code, 200) response = self.client.get(f"/reports/{section}/", {"export": "csv"}) self.assertEqual(response.status_code, 200) list(response.streaming_content) self.assertEqual(snapshot(), before) class FilterTests(TestCase): def test_valid_range_and_bad_values(self): self.assertTrue(ReportFilters({"date_from": "2026-09-01", "date_to": "2026-09-30"}).is_valid()) for data in ({"date_from": "2026-09-31"}, {"date_to": "9999-12-31"}, {"status": "INVALID"}, {"responsibility": "FREE"}, {"company": "invalid"}): self.assertFalse(ReportFilters(data).is_valid()) def test_csv_formula_escape(self): for value in ("=1+1", "+cmd", "-1+2", "@SUM(A1)", " \t=cmd", "\ttext", "\n=cmd"): self.assertTrue(safe_cell(value).startswith("'")) self.assertEqual(safe_cell("বাংলা"), "বাংলা") self.assertEqual(safe_cell(None), "") self.assertEqual(safe_cell(Decimal("12.50")), "12.50")