"""Cross-phase audit: real transitions, immutable evidence and rollback boundaries.""" from datetime import date from unittest.mock import patch from django.apps import apps from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.core.exceptions import ValidationError, PermissionDenied from django.db import IntegrityError, transaction from django.db.models.deletion import Collector, ProtectedError from django.test import TestCase, Client from django.urls import reverse from apps.access import services as access from apps.access.models import Role from apps.organization import assignment_services as postings from apps.devices import services as devices from apps.service_catalog import services as taxonomy from apps.service_catalog.models import FaultDiagnosis, RootCause, RepairAction from apps.service import services as intake, engineer_services as engineer, diagnostic_services as diagnosis from apps.service import repair_services as repair, quality_control_services as qc, handover_services as delivery from apps.service import queries as iq, engineer_queries as eq, diagnostic_queries as dq from apps.service import repair_queries as rq, quality_control_queries as qq, handover_queries as hq from apps.service.models import (ServiceCase, ServiceEngineerAssignment, ServiceDiagnosticAssessment, ServiceDiagnosticFinding, ServiceRepairExecution, ServiceRepairAction, ServiceQualityControl, ServiceCaseDeliveryRelease, ServiceCaseHandover, ServiceCaseClosure) from apps.service.quality_control_checklist import CHECKLIST from apps.service.test_engineer_assignment import setup_engineers STATES = ("RECEIVED", "ASSIGNED", "DIAGNOSING", "DIAGNOSED", "REPAIRING", "REPAIRED", "QC_PENDING", "QC_IN_PROGRESS", "QC_PASSED", "READY_FOR_DELIVERY", "DELIVERED", "CLOSED", "CANCELLED") TRANSITIONS = { "assign": {"RECEIVED": "ASSIGNED"}, "reassign": {"ASSIGNED": "ASSIGNED"}, "unassign": {"ASSIGNED": "RECEIVED"}, "diagnose": {"ASSIGNED": "DIAGNOSING"}, "diagnosis_complete": {"DIAGNOSING": "DIAGNOSED"}, "diagnosis_abandon": {"DIAGNOSING": "ASSIGNED"}, "repair": {"DIAGNOSED": "REPAIRING"}, "repair_success": {"REPAIRING": "REPAIRED"}, "repair_failure": {"REPAIRING": "DIAGNOSED"}, "repair_abandon": {"REPAIRING": "DIAGNOSED"}, "qc_submit": {"REPAIRED": "QC_PENDING"}, "qc_begin": {"QC_PENDING": "QC_IN_PROGRESS"}, "qc_pass": {"QC_IN_PROGRESS": "QC_PASSED"}, "qc_fail": {"QC_IN_PROGRESS": "DIAGNOSED"}, "qc_abandon": {"QC_IN_PROGRESS": "QC_PENDING"}, "release": {"QC_PASSED": "READY_FOR_DELIVERY"}, "deliver": {"READY_FOR_DELIVERY": "DELIVERED"}, "close": {"DELIVERED": "CLOSED"}, "cancel": {state: "CANCELLED" for state in ("RECEIVED", "ASSIGNED", "DIAGNOSING", "DIAGNOSED", "REPAIRING", "CANCELLED")}, "intake_edit": {"RECEIVED": "RECEIVED"}, "condition": {"RECEIVED": "RECEIVED"}, "accessory": {"RECEIVED": "RECEIVED"}, "complaint_remove": {"RECEIVED": "RECEIVED"}, "diagnosis_note": {"DIAGNOSING": "DIAGNOSING"}, "repair_note": {"REPAIRING": "REPAIRING"}, "qc_note": {"QC_IN_PROGRESS": "QC_IN_PROGRESS"}, } FORWARD = ("assign", "diagnose", "diagnosis_complete", "repair", "repair_success", "qc_submit", "qc_begin", "qc_pass", "release", "deliver", "close") def setup_audit(test): # Includes real Company trees, Customer, Device, intake and independent engineers. setup_engineers(test) test.complaint = intake.add_service_case_complaint(service_case=test.case, complaint_symptom=test.symptom, customer_description="Synthetic intermittent fault") test.received = [intake.add_service_case_accessory(service_case=test.case, accessory_type="CABLE", quantity=2), intake.add_service_case_accessory(service_case=test.case, accessory_type="CASE", quantity=1)] intake.add_service_case_intake_condition(service_case=test.case, condition_type="COSMETIC", description="Synthetic intake condition") test.fault = FaultDiagnosis.objects.create(code="AUDIT-FAULT", name="Synthetic fault", applies_to_all_product_categories=True) test.action_type = RepairAction.objects.create(code="AUDIT-ACTION", name="Synthetic action", applies_to_all_product_categories=True) test.inspector = get_user_model().objects.create_user(username="synthetic-audit-inspector") test.audit_role = Role.objects.create(code="AUDIT-REVIEW", name="Synthetic review and delivery") access.set_role_permissions(role=test.audit_role, permissions=list(Permission.objects.filter(content_type__app_label="service", codename__in=["perform_quality_control", "handover_servicecase", "close_servicecase"]))) test.audit_path = postings.create_assignment(user=test.inspector, company=test.company, region=test.region, service_center=test.center) test.audit_role_assignment = access.create_role_assignment(user=test.inspector, role=test.audit_role, organization_assignment=test.audit_path) def snapshot(): return {model._meta.label: list(model.objects.order_by("pk").values()) for model in apps.get_app_config("service").get_models()} def latest(test, model): return model.objects.filter(service_case=test.case).order_by("-created_at", "-pk").first() or model() def operate(test, operation): case = ServiceCase.objects.get(pk=test.case.pk) actor = test.engineer if operation == "assign": return engineer.assign_service_case_engineer(service_case=case, engineer=actor, assigned_by=test.user) if operation == "reassign": return engineer.reassign_service_case_engineer(service_case=case, new_engineer=test.engineer2, reassigned_by=test.user) if operation == "unassign": return engineer.unassign_service_case_engineer(service_case=case, unassigned_by=test.user) if operation == "cancel": return intake.cancel_service_case(service_case=case, cancelled_by=test.user, reason="Synthetic audit cancellation") if operation == "intake_edit": return intake.update_service_case_intake(service_case=case, intake_note="Synthetic edit") if operation == "condition": return intake.add_service_case_intake_condition(service_case=case, condition_type="COSMETIC", description="Synthetic observation") if operation == "accessory": return intake.add_service_case_accessory(service_case=case, accessory_type="BOX") if operation == "complaint_remove": return intake.remove_service_case_complaint(service_case=case, complaint=test.complaint) if operation == "diagnose": return diagnosis.begin_service_case_diagnosis(service_case=case, actor=actor) if operation.startswith("diagnosis_"): assessment = latest(test, ServiceDiagnosticAssessment) if operation == "diagnosis_complete": return diagnosis.complete_service_case_diagnosis(assessment=assessment, actor=actor) if operation == "diagnosis_note": return diagnosis.update_diagnostic_assessment(assessment=assessment, actor=actor, technical_note="Synthetic note") return diagnosis.abandon_service_case_diagnosis(assessment=assessment, actor=test.user, reason="Synthetic interruption") if operation == "repair": return repair.begin_service_case_repair(service_case=case, actor=actor) if operation.startswith("repair_"): execution = latest(test, ServiceRepairExecution) if operation == "repair_abandon": return repair.abandon_service_case_repair(repair_execution=execution, actor=test.user, reason="Synthetic interruption") if operation == "repair_note": return repair.update_repair_execution(repair_execution=execution, actor=actor, note="Synthetic note") return repair.complete_service_case_repair(repair_execution=execution, actor=actor, outcome="REPAIRED" if operation == "repair_success" else "NOT_REPAIRED") if operation == "qc_submit": return qc.submit_for_quality_control(service_case=case, inspector=test.inspector) if operation == "qc_begin": return qc.begin_quality_control(service_case=case, inspector=test.inspector) if operation.startswith("qc_"): attempt = latest(test, ServiceQualityControl) if operation == "qc_abandon": return qc.abandon_quality_control(quality_control=attempt, actor=test.user, reason="Synthetic interruption") if operation == "qc_note": return qc.update_quality_control(quality_control=attempt, inspector=test.inspector, summary="Synthetic note") if operation == "qc_fail": qc.set_quality_control_check(quality_control=attempt, inspector=test.inspector, check_code="BOOT", result="FAIL") attempt.refresh_from_db() return qc.complete_quality_control_fail(quality_control=attempt, inspector=test.inspector, failure_note="Synthetic failure") return qc.complete_quality_control_pass(quality_control=attempt, inspector=test.inspector) if operation == "close": return delivery.close_service_case(service_case=case, actor=test.inspector) args = dict(service_case=case, actor=test.inspector, expected_device=test.device, expected_device_identity=hq.device_identity_fingerprint(test.device)) if operation == "release": return delivery.mark_service_case_ready_for_delivery(**args) if operation == "deliver": return delivery.handover_service_case(**args, recipient_type="CUSTOMER", verification_method="CUSTOMER_RECORD", accessories=[dict(intake_accessory=a, returned_quantity=a.quantity) for a in test.received]) raise AssertionError(operation) def prepare_children(test, operation, row): if operation == "diagnose": diagnosis.add_diagnostic_finding(assessment=row, actor=test.engineer, fault_diagnosis=test.fault, root_cause=None) elif operation == "repair": action = repair.add_repair_action(repair_execution=row, actor=test.engineer, repair_action=test.action_type) repair.mark_repair_action_performed(action=action, actor=test.engineer) elif operation == "qc_begin": for code, _, _ in CHECKLIST: row.refresh_from_db() qc.set_quality_control_check(quality_control=row, inspector=test.inspector, check_code=code, result="PASS") row.refresh_from_db() qc.set_quality_control_complaint_check(quality_control=row, inspector=test.inspector, complaint=test.complaint, result="RESOLVED") def advance(test, operation): row = operate(test, operation) prepare_children(test, operation, row) test.case.refresh_from_db() return row def reach(test, state): if state == "CANCELLED": advance(test, "cancel") return for operation in FORWARD: test.case.refresh_from_db() if test.case.status == state: return advance(test, operation) test.case.refresh_from_db() if test.case.status != state: raise AssertionError((test.case.status, state)) def invariants(test): case = ServiceCase.objects.get(pk=test.case.pk) test.assertEqual(ServiceEngineerAssignment.objects.filter(service_case=case, ended_at=None).count(), int(case.status not in ("RECEIVED", "CLOSED", "DELIVERED", "CANCELLED"))) test.assertEqual(ServiceDiagnosticAssessment.objects.filter(service_case=case, completed_at=None, abandoned_at=None).count(), int(case.status == "DIAGNOSING")) test.assertEqual(ServiceRepairExecution.objects.filter(service_case=case, status="OPEN").count(), int(case.status == "REPAIRING")) test.assertEqual(ServiceQualityControl.objects.filter(service_case=case, status="IN_PROGRESS").count(), int(case.status == "QC_IN_PROGRESS")) test.assertEqual(ServiceCaseHandover.objects.filter(service_case=case).count(), int(case.status in ("DELIVERED", "CLOSED"))) test.assertEqual(ServiceCaseClosure.objects.filter(service_case=case).count(), int(case.status == "CLOSED")) class Phase3AMatrixTests(TestCase): @classmethod def setUpTestData(cls): setup_audit(cls) def matrix_test(state): def test(self): reach(self, state) for operation, allowed in TRANSITIONS.items(): with self.subTest(state=state, operation=operation): before = snapshot() with transaction.atomic(): if state in allowed: operate(self, operation) self.assertEqual(ServiceCase.objects.get(pk=self.case.pk).status, allowed[state]) invariants(self) else: with self.assertRaises((ValidationError, PermissionDenied)): operate(self, operation) self.assertEqual(snapshot(), before) transaction.set_rollback(True) self.assertEqual(snapshot(), before) return test for state in STATES: setattr(Phase3AMatrixTests, "test_matrix_" + state.lower(), matrix_test(state)) class Phase3AIntegrationTests(TestCase): @classmethod def setUpTestData(cls): setup_audit(cls) def test_golden_unknown_cause_history_and_actors(self): self.assertEqual(RootCause.objects.count(), 0) expected = iter(STATES[1:12]) records = [] for operation in FORWARD: row = advance(self, operation) records.append(row) self.assertEqual(self.case.status, next(expected)) invariants(self) assessment = latest(self, ServiceDiagnosticAssessment) execution = latest(self, ServiceRepairExecution) attempt = latest(self, ServiceQualityControl) handover = hq.get_service_case_handover(self.case) closure = hq.get_service_case_closure(self.case) self.assertIsNone(assessment.findings.get().root_cause) self.assertEqual(RootCause.objects.count(), 0) self.assertEqual(execution.diagnostic_assessment, assessment) self.assertEqual(attempt.repair_execution, execution) self.assertEqual(handover.delivery_release.quality_control, attempt) self.assertEqual(closure.handover, handover) self.assertEqual(assessment.completed_by, self.engineer) self.assertEqual(execution.completed_by, self.engineer) self.assertEqual(attempt.inspector, self.inspector) self.assertEqual(handover.handed_over_by, self.inspector) self.assertEqual(closure.closed_by, self.inspector) times = [self.case.received_at, assessment.started_at, assessment.completed_at, execution.started_at, execution.completed_at, attempt.started_at, attempt.completed_at, handover.delivery_release.readied_at, handover.delivered_at, closure.closed_at] self.assertEqual(times, sorted(times)) assignment = eq.engineer_assignment_history(self.case).get() self.assertEqual(assignment.ended_at, handover.delivered_at) self.assertEqual(assignment.ended_by, self.inspector) self.assertEqual(handover.accessories.count(), 2) self.assertEqual(list(self.case.accessories.order_by("quantity").values_list("quantity", flat=True)), [1, 2]) def test_qc_rework_to_closure_preserves_both_cycles(self): reach(self, "QC_IN_PROGRESS") first_repair = latest(self, ServiceRepairExecution) first_qc = advance(self, "qc_fail") old_repair = ServiceRepairExecution.objects.filter(pk=first_repair.pk).values().get() old_qc = ServiceQualityControl.objects.filter(pk=first_qc.pk).values().get() for operation in ("repair", "repair_success", "qc_submit", "qc_begin", "qc_pass", "release", "deliver", "close"): advance(self, operation) invariants(self) self.assertEqual(ServiceRepairExecution.objects.filter(pk=first_repair.pk).values().get(), old_repair) self.assertEqual(ServiceQualityControl.objects.filter(pk=first_qc.pk).values().get(), old_qc) self.assertEqual(list(rq.repair_execution_history(self.case).values_list("outcome", flat=True)), ["REPAIRED", "REPAIRED"]) self.assertEqual(list(qq.quality_control_history(self.case).values_list("outcome", flat=True)), ["FAILED", "PASSED"]) self.assertNotEqual(hq.get_service_case_handover(self.case).delivery_release.quality_control_id, first_qc.pk) def test_abandonments_and_unsuccessful_repair_then_success(self): reach(self, "DIAGNOSING") assessment = advance(self, "diagnosis_abandon") advance(self, "diagnose") advance(self, "diagnosis_complete") advance(self, "repair") abandoned = advance(self, "repair_abandon") advance(self, "repair") failed = advance(self, "repair_failure") advance(self, "repair") advance(self, "repair_success") advance(self, "qc_submit") advance(self, "qc_begin") abandoned_qc = advance(self, "qc_abandon") advance(self, "qc_begin") for operation in ("qc_pass", "release", "deliver", "close"): advance(self, operation) for row in (assessment, abandoned, failed, abandoned_qc): row.refresh_from_db() with self.assertRaises(ValidationError): row.save() self.assertEqual(ServiceDiagnosticAssessment.objects.count(), 2) self.assertEqual(ServiceRepairExecution.objects.count(), 3) self.assertEqual(ServiceQualityControl.objects.count(), 2) invariants(self) def test_late_rollback_every_major_transition(self): # A failure after child/evidence writes but before final case persistence. for operation in FORWARD: before = snapshot() with patch.object(ServiceCase, "_persist", side_effect=RuntimeError("Synthetic late failure")), self.assertRaises(RuntimeError): operate(self, operation) self.assertEqual(snapshot(), before, operation) advance(self, operation) invariants(self) def test_late_rollback_qc_fail_and_cancellation(self): for target, operation in (("DIAGNOSING", "cancel"), ("QC_IN_PROGRESS", "qc_fail")): with self.subTest(operation=operation), transaction.atomic(): reach(self, target) before = snapshot() # Outer transaction also rolls back the explicit failed-check evidence. with self.assertRaises(RuntimeError), transaction.atomic(): with patch.object(ServiceCase, "_persist", side_effect=RuntimeError("Synthetic late failure")): operate(self, operation) self.assertEqual(snapshot(), before) transaction.set_rollback(True) def test_late_intake_job_number_rollback(self): before = snapshot() with patch.object(intake, "_capture_warranty", side_effect=RuntimeError("Synthetic late failure")), self.assertRaises(RuntimeError): intake.create_service_case(company=self.company, service_center=self.center, customer=self.customer, device=self.device, created_by=self.user, intake_channel="WALK_IN") self.assertEqual(snapshot(), before) def test_identifier_change_preserves_technical_chain_but_blocks_released_delivery(self): reach(self, "QC_PASSED") original = self.case.device_id devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-AUDIT-A") advance(self, "release") devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-AUDIT-B") with self.assertRaises(ValidationError): operate(self, "deliver") self.case.refresh_from_db() self.assertEqual(self.case.device_id, original) self.assertEqual(self.case.status, "READY_FOR_DELIVERY") self.assertEqual(qq.quality_control_history(self.case).count(), 1) def test_historical_taxonomy_deactivation_does_not_block_delivery(self): reach(self, "QC_PASSED") taxonomy.deactivate_fault_diagnosis(diagnosis=self.fault) taxonomy.deactivate_repair_action(repair_action=self.action_type) taxonomy.deactivate_complaint(complaint=self.symptom) for operation in ("release", "deliver", "close"): advance(self, operation) invariants(self) def test_final_history_ordinary_orm_save_delete_and_protected_references(self): reach(self, "CLOSED") for model in apps.get_app_config("service").get_models(): if model.__name__ == "ServiceCaseNumberSequence": continue for row in model.objects.all(): with self.assertRaises(ValidationError): row.save() with self.assertRaises(ValidationError): row.delete() for obj in (self.customer, self.device, self.center, self.user, self.engineer, self.inspector, self.fault, self.action_type, self.symptom, latest(self, ServiceDiagnosticAssessment), latest(self, ServiceRepairExecution), latest(self, ServiceQualityControl), hq.get_service_case_handover(self.case)): with self.assertRaises(ProtectedError): Collector(using="default").collect([obj]) def test_privileged_update_boundary_is_not_a_tamper_proof_ledger(self): reach(self, "CLOSED") before = snapshot() with transaction.atomic(): self.assertEqual(ServiceCaseClosure.objects.filter(service_case=self.case).update(note="Synthetic privileged overwrite"), 1) transaction.set_rollback(True) self.assertEqual(snapshot(), before) for value in ("INVALID", "REOPENED", "closed", ""): with self.subTest(value=value), self.assertRaises(IntegrityError), transaction.atomic(): ServiceCase.objects.filter(pk=self.case.pk).update(status=value) def test_complete_case_history_query_budget(self): reach(self, "CLOSED") # Compose existing queries; no new reporting abstraction is introduced. with self.assertNumQueries(16): case = iq.service_cases_for_customer(self.customer).get(pk=self.case.pk) str(case.customer); str(case.device); str(case.service_center); str(case.created_by); str(case.warranty_snapshot) assignments = list(eq.engineer_assignment_history(case)) assessments = list(dq.diagnostic_assessment_history(case)) findings = list(dq.diagnostic_findings_for_assessment(assessments[0])) repairs = list(rq.repair_execution_history(case)) actions = list(rq.repair_action_history(repairs[0])) controls = list(qq.quality_control_history(case)) checks = list(qq.quality_control_checks(controls[0])) complaints = list(qq.quality_control_complaint_checks(controls[0])) handover = hq.get_service_case_handover(case) closure = hq.get_service_case_closure(case) list(case.complaints.select_related("complaint_symptom")) list(case.intake_conditions.all()) list(case.accessories.all()) list(case.device.identifiers.all()) list(hq.handover_accessories(handover)) for row in assignments: str(row.engineer); str(row.ended_by) for row in findings: str(row.fault_diagnosis); str(row.root_cause) for row in actions: str(row.repair_action); str(row.performed_by) for row in complaints: str(row.complaint.complaint_symptom) str(handover.handed_over_by); str(closure.closed_by) self.assertEqual(len(checks), len(CHECKLIST)) def test_warranty_and_purchase_changes_do_not_rewrite_intake_context(self): original = self.case.warranty_snapshot devices.set_device_warranty_coverage(device=self.device, coverage_start_date=date(2020, 1, 1), coverage_end_date=date(2099, 1, 1), coverage_source="MANUFACTURER_RECORD", reference="SYNTHETIC-AUDIT") original.refresh_from_db() self.assertFalse(original.recorded_coverage) covered = intake.create_service_case(company=self.company, service_center=self.center, customer=self.customer, device=self.device, created_by=self.user, intake_channel="WALK_IN") before = type(original).objects.filter(service_case=covered).values().get() devices.clear_device_warranty_coverage(device=self.device) devices.set_device_warranty_coverage(device=self.device, coverage_start_date=date(2021, 1, 1), coverage_end_date=date(2098, 1, 1), coverage_source="MANUFACTURER_RECORD", reference="SYNTHETIC-REPLACED") devices.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-PURCHASE") self.assertEqual(type(original).objects.filter(service_case=covered).values().get(), before) def test_presenter_ownership_history_and_later_transfer_are_separate(self): from apps.devices.models import CustomerDeviceRelationship devices.assign_device_owner(device=self.device, customer=self.presenter) before = self.case.customer_id, self.case.device_id, self.case.created_by_id devices.transfer_device_ownership(device=self.device, new_customer=self.customer, reason="Synthetic owner change") ownership = list(CustomerDeviceRelationship.objects.order_by("pk").values()) reach(self, "CLOSED") self.assertEqual((self.case.customer_id, self.case.device_id, self.case.created_by_id), before) self.assertEqual(list(CustomerDeviceRelationship.objects.order_by("pk").values()), ownership) def test_foreign_company_affinity_and_mixed_intake_objects_reject(self): devices.assign_device_owner(device=self.device2, customer=self.outsider) for values in (dict(customer=self.outsider), dict(service_center=self.other_center), dict(device=self.device2)): with self.subTest(values=values), self.assertRaises(ValidationError): intake.create_service_case(**dict(dict(company=self.company, service_center=self.center, customer=self.customer, device=self.device, created_by=self.user, intake_channel="WALK_IN"), **values)) def test_foreign_company_and_sibling_center_actors_cannot_work_or_query(self): outsider = get_user_model().objects.create_user(username="synthetic-foreign-actor") for company, center in ((self.other_company, self.other_center), (self.company, self.center2)): posting = postings.create_assignment(user=outsider, company=company, region=center.region, service_center=center) access.create_role_assignment(user=outsider, role=self.role, organization_assignment=posting) access.create_role_assignment(user=outsider, role=self.audit_role, organization_assignment=posting) reach(self, "ASSIGNED") with self.assertRaises(ValidationError): diagnosis.begin_service_case_diagnosis(service_case=self.case, actor=outsider) self.assertFalse(eq.assigned_service_cases_for_engineer(outsider).filter(pk=self.case.pk).exists()) for operation in FORWARD[1:7]: advance(self, operation) with self.assertRaises(ValidationError): qc.update_quality_control(quality_control=latest(self, ServiceQualityControl), inspector=outsider, summary="Synthetic forbidden") self.assertFalse(qq.qc_cases_for_inspector(outsider).filter(pk=self.case.pk).exists()) advance(self, "qc_pass") with self.assertRaises(PermissionDenied): delivery.mark_service_case_ready_for_delivery(service_case=self.case, actor=outsider, expected_device=self.device, expected_device_identity=hq.device_identity_fingerprint(self.device)) self.assertFalse(hq.authorized_delivery_cases(actor=outsider).filter(pk=self.case.pk).exists()) def test_required_complaint_resolution_blocks_qc_pass(self): reach(self, "QC_IN_PROGRESS") row = latest(self, ServiceQualityControl) qc.set_quality_control_complaint_check(quality_control=row, inspector=self.inspector, complaint=self.complaint, result="NOT_RESOLVED") with self.assertRaises(ValidationError): operate(self, "qc_pass") invariants(self) def test_operational_queues_and_details_have_bounded_queries(self): def measure(): state = ServiceCase.objects.get(pk=self.case.pk).status checks = { "RECEIVED": [lambda: iq.active_service_cases_for_service_center(self.center)], "ASSIGNED": [lambda: eq.assigned_service_cases_for_engineer(self.engineer)], "DIAGNOSING": [lambda: dq.diagnostic_assessment_history(self.case)], "REPAIRING": [lambda: rq.repair_execution_history(self.case)], "QC_PENDING": [lambda: qq.qc_pending_cases_for_center(self.center, inspector=self.inspector)], "QC_IN_PROGRESS": [lambda: qq.quality_control_history(self.case)], "READY_FOR_DELIVERY": [lambda: hq.ready_for_delivery_cases_for_center(self.center, actor=self.inspector)], "DELIVERED": [lambda: hq.delivered_cases_for_center(self.center, actor=self.inspector)], "CLOSED": [lambda: hq.closed_cases_for_center(self.center, actor=self.inspector)], } for query in checks.get(state, []): with self.subTest(state=state), self.assertNumQueries(1): rows = list(query()) self.assertEqual(len(rows), 1) row = rows[0] if isinstance(row, ServiceCase): str(row.customer); str(row.device); str(row.service_center) measure() for operation in FORWARD: advance(self, operation) measure()