"""Separate-connection PostgreSQL intake/lifecycle/snapshot serialization.""" from django.contrib.auth import get_user_model from django.test import TransactionTestCase from apps.catalog import services as catalog from apps.catalog.identification import set_identification_policy from apps.customers import services as customers from apps.devices import services as devices from apps.organization import services as organization from apps.organization import test_assignment_concurrency as concurrency from apps.service_catalog import services as taxonomy from . import services from .models import ServiceCase, ServiceCaseComplaint from .tests import setup, intake, warranty class IntakeConcurrencyTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): setup(self) def test_same_center_allocates_distinct_jobs(self): self.run_concurrent(lambda: intake(self), lambda: intake(self), expected="success") self.assertEqual(list(ServiceCase.objects.order_by("job_number").values_list("job_number", flat=True)), ["JOB-00000001", "JOB-00000002"]) def test_different_centers_share_customer_without_counter_serialization(self): self.run_concurrent(lambda: intake(self), lambda: intake(self, service_center=self.center2, device=self.device2), expected="success", should_block=False) self.assertEqual(list(ServiceCase.objects.values_list("job_number", flat=True)), ["JOB-00000001", "JOB-00000001"]) def test_company_deactivation_then_intake_rejects(self): self.run_concurrent(lambda: organization.deactivate_company(company=self.company), lambda: intake(self), expected="validation") self.assertFalse(ServiceCase.objects.exists()) def test_center_deactivation_then_intake_rejects(self): self.run_concurrent(lambda: organization.deactivate_service_center(service_center=self.center), lambda: intake(self), expected="validation") def test_customer_deactivation_then_intake_rejects(self): self.run_concurrent(lambda: customers.deactivate_customer(customer=self.customer), lambda: intake(self), expected="validation") def test_device_deactivation_then_intake_rejects(self): self.run_concurrent(lambda: devices.deactivate_device(device=self.device), lambda: intake(self), expected="validation") def test_creator_deactivation_then_intake_rejects(self): self.run_concurrent(lambda: get_user_model().objects.filter(pk=self.user.pk).update(is_active=False), lambda: intake(self), expected="validation") def test_category_deactivation_then_intake_rejects(self): self.run_concurrent(lambda: catalog.deactivate_category(category=self.category), lambda: intake(self), expected="validation") def test_policy_change_then_intake_revalidates(self): self.run_concurrent(lambda: set_identification_policy(product_model=self.model, imei1_requirement="REQUIRED", imei2_requirement="OPTIONAL", serial_requirement="OPTIONAL"), lambda: intake(self), expected="validation") def test_intake_then_company_deactivation_preserves_case(self): self.run_concurrent(lambda: intake(self), lambda: organization.deactivate_company(company=self.company), expected="success") self.assertEqual(ServiceCase.objects.get().status, "RECEIVED") def test_intake_then_device_deactivation_preserves_snapshot(self): warranty(self) self.run_concurrent(lambda: intake(self), lambda: devices.deactivate_device(device=self.device), expected="success") self.assertTrue(ServiceCase.objects.get().warranty_snapshot.recorded_coverage) def test_warranty_replacement_then_snapshot_captures_new(self): warranty(self) self.run_concurrent(lambda: warranty(self, reference="SYNTHETIC-NEW"), lambda: intake(self), expected="success") self.assertEqual(ServiceCase.objects.get().warranty_snapshot.coverage_reference, "SYNTHETIC-NEW") def test_snapshot_then_warranty_replacement_retains_old(self): warranty(self) self.run_concurrent(lambda: intake(self), lambda: warranty(self, reference="SYNTHETIC-NEW"), expected="success") self.assertEqual(ServiceCase.objects.get().warranty_snapshot.coverage_reference, "SYNTHETIC-COVERAGE") def test_warranty_clear_then_snapshot_is_uncovered(self): warranty(self) self.run_concurrent(lambda: devices.clear_device_warranty_coverage(device=self.device), lambda: intake(self), expected="success") self.assertFalse(ServiceCase.objects.get().warranty_snapshot.recorded_coverage) def test_snapshot_then_warranty_clear_retains_coverage(self): warranty(self) self.run_concurrent(lambda: intake(self), lambda: devices.clear_device_warranty_coverage(device=self.device), expected="success") self.assertTrue(ServiceCase.objects.get().warranty_snapshot.recorded_coverage) def test_purchase_invalidation_then_snapshot_has_no_dependent_coverage(self): evidence = devices.set_device_purchase_evidence(device=self.device) devices.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) warranty(self, coverage_source="PURCHASE_EVIDENCE") self.run_concurrent(lambda: devices.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-CHANGED"), lambda: intake(self), expected="success") self.assertFalse(ServiceCase.objects.get().warranty_snapshot.recorded_coverage) def test_complaint_deactivation_then_add_rejects(self): case = intake(self) self.run_concurrent(lambda: taxonomy.deactivate_complaint(complaint=self.symptom), lambda: services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom), expected="validation") def test_category_deactivation_then_complaint_add_rejects(self): case = intake(self) self.run_concurrent(lambda: catalog.deactivate_category(category=self.category), lambda: services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom), expected="validation") def test_applicability_change_then_add_rejects(self): case = intake(self) self.run_concurrent(lambda: taxonomy.set_complaint_applicability(complaint=self.symptom, applies_to_all_product_categories=False, product_categories=[]), lambda: services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom), expected="validation") def test_complaint_add_then_deactivation_preserves_history(self): case = intake(self) self.run_concurrent(lambda: services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom), lambda: taxonomy.deactivate_complaint(complaint=self.symptom), expected="success") self.assertEqual(ServiceCaseComplaint.objects.count(), 1) def test_cancel_then_child_add_rejects(self): case = intake(self) self.run_concurrent(lambda: services.cancel_service_case(service_case=case), lambda: services.add_service_case_accessory(service_case=case, accessory_type="BOX"), expected="validation") self.assertFalse(case.accessories.exists()) def test_child_add_then_cancellation_preserves_child(self): case = intake(self) self.run_concurrent(lambda: services.add_service_case_accessory(service_case=case, accessory_type="BOX"), lambda: services.cancel_service_case(service_case=case), expected="success") self.assertEqual(case.accessories.count(), 1) def test_duplicate_complaints_serialize_and_reject(self): case = intake(self) def add(): return services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) self.run_concurrent(add, add, expected="validation") self.assertEqual(case.complaints.count(), 1) def test_foreign_ownership_affinity_wins_before_intake(self): self.run_concurrent(lambda: devices.assign_device_owner(device=self.device, customer=self.outsider), lambda: intake(self), expected="validation") self.assertFalse(ServiceCase.objects.exists())