"""Synthetic intake integrity, snapshots, administration and rollback.""" import uuid from datetime import date, datetime, timedelta, timezone as tz from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.core.exceptions import ValidationError 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 django.utils import timezone from apps.access.authorization import is_authorized from apps.catalog.tests import make_catalog from apps.catalog import services as catalog from apps.customers.tests import create as customer_create from apps.customers import services as customers from apps.devices import services as devices from apps.devices.models import CustomerDeviceRelationship, DevicePurchaseEvidence, DeviceWarrantyCoverage from apps.devices.tests import policy, synthetic_imei from apps.organization.test_lifecycle import make_tree from apps.organization.models import ServiceCenter from apps.organization import services as organization from apps.service_catalog.models import ComplaintSymptom from apps.service_catalog import services as taxonomy from . import services, queries from .models import (ServiceCase, ServiceCaseComplaint, ServiceCaseIntakeCondition, ServiceCaseAccessory, ServiceCaseWarrantySnapshot, ServiceCaseNumberSequence, MAX_JOB_NUMBER) def setup(test): test.company, test.region, test.center, _ = make_tree("INTAKE") test.other_company, _, test.other_center, _ = make_tree("OTHER") test.center2 = ServiceCenter.objects.create(company=test.company, region=test.region, code="SECOND", name="Second", center_type="OWN") test.customer = customer_create(test.company) test.presenter = customer_create(test.company) test.outsider = customer_create(test.other_company) test.brand, test.category, test.model, _ = make_catalog("INTAKE") policy(test.model) test.device = devices.register_device(product_model=test.model) test.device2 = devices.register_device(product_model=test.model) test.user = get_user_model().objects.create_user(username="synthetic-intake-actor") test.symptom = ComplaintSymptom.objects.create(code="SYNTHETIC", name="Synthetic complaint", applies_to_all_product_categories=True) def intake(test, **kwargs): return services.create_service_case(**dict(dict(company=test.company, service_center=test.center, customer=test.customer, device=test.device, created_by=test.user, intake_channel="WALK_IN"), **kwargs)) def warranty(test, **kwargs): return devices.set_device_warranty_coverage(**dict(dict(device=test.device, coverage_start_date=date(2025, 1, 1), coverage_end_date=date(2030, 12, 31), coverage_source="MANUFACTURER_RECORD", reference="SYNTHETIC-COVERAGE"), **kwargs)) class IntakeTests(TestCase): def setUp(self): setup(self) def test_case_defaults_uuid_and_empty_snapshot(self): case = intake(self, customer_reference=" Synthetic reference ", intake_note=" Synthetic note ") self.assertIsInstance(case.pk, uuid.UUID) self.assertEqual(case.job_number, "JOB-00000001") self.assertEqual(case.status, "RECEIVED") self.assertEqual(case.created_by, self.user) self.assertTrue(timezone.is_aware(case.received_at)) self.assertEqual(case.customer_reference, "Synthetic reference") self.assertEqual(case.intake_note, "Synthetic note") self.assertFalse(case.warranty_snapshot.recorded_coverage) self.assertIsNone(case.warranty_snapshot.coverage_source) self.assertIsNone(case.warranty_snapshot.coverage_reference) def test_center_scoped_numbering(self): first, second, other = intake(self), intake(self), intake(self, service_center=self.center2) self.assertEqual([row.job_number for row in [first, second, other]], ["JOB-00000001", "JOB-00000002", "JOB-00000001"]) def test_all_channels(self): for channel in ServiceCase.Channel.values: self.assertEqual(intake(self, intake_channel=channel).intake_channel, channel) def test_cross_company_inputs_reject(self): for kwargs in [dict(service_center=self.other_center), dict(customer=self.outsider), dict(company=self.other_company)]: with self.subTest(kwargs=kwargs), self.assertRaises(ValidationError): intake(self, **kwargs) self.assertFalse(ServiceCaseNumberSequence.objects.exists()) def test_presenter_need_not_be_owner(self): owner = devices.assign_device_owner(device=self.device, customer=self.customer) case = intake(self, customer=self.presenter) self.assertEqual(case.customer, self.presenter) self.assertEqual(CustomerDeviceRelationship.objects.get(), owner) def test_foreign_affinity_rejects_even_after_ownership_ends(self): devices.assign_device_owner(device=self.device, customer=self.outsider) devices.end_device_ownership(device=self.device) with self.assertRaises(ValidationError): intake(self) def test_intake_does_not_establish_hidden_ownership_affinity(self): intake(self) case = intake(self, company=self.other_company, service_center=self.other_center, customer=self.outsider) self.assertEqual(case.company, self.other_company) self.assertFalse(CustomerDeviceRelationship.objects.exists()) def test_unsaved_or_wrong_creator_rejected(self): for actor in [None, get_user_model()(username="unsaved")]: with self.assertRaises(ValidationError): intake(self, created_by=actor) def test_inactive_creator_rejected(self): get_user_model().objects.filter(pk=self.user.pk).update(is_active=False) with self.assertRaises(ValidationError): intake(self) def test_inactive_customer_rejected_from_stale_object(self): customers.deactivate_customer(customer=self.customer) with self.assertRaises(ValidationError): intake(self) def test_inactive_device_rejected_from_stale_object(self): devices.deactivate_device(device=self.device) with self.assertRaises(ValidationError): intake(self) def test_inactive_company_and_center_rejected(self): organization.deactivate_service_center(service_center=self.center) with self.assertRaises(ValidationError): intake(self) organization.deactivate_company(company=self.company) with self.assertRaises(ValidationError): intake(self, service_center=self.center2) def test_inactive_catalog_rejected(self): catalog.deactivate_brand(brand=self.brand) with self.assertRaises(ValidationError): intake(self) def test_current_identification_policy_required_at_intake(self): from apps.catalog.identification import set_identification_policy set_identification_policy(product_model=self.model, imei1_requirement="REQUIRED", imei2_requirement="OPTIONAL", serial_requirement="OPTIONAL") with self.assertRaises(ValidationError): intake(self) def test_unconfigured_policy_rejected(self): self.model.identification_policy.delete() with self.assertRaises(ValidationError): intake(self) def test_received_timestamp_validation_and_utc_normalization(self): for timestamp in [datetime(2025, 1, 1), timezone.now()+timedelta(minutes=1), "2025-01-01"]: with self.subTest(timestamp=timestamp), self.assertRaises(ValidationError): intake(self, received_at=timestamp) timestamp = datetime(2025, 1, 2, 1, tzinfo=tz(timedelta(hours=6))) case = intake(self, received_at=timestamp) self.assertEqual(case.received_at.date(), date(2025, 1, 1)) def test_atomic_snapshot_failure_rolls_back_first_counter(self): with patch("apps.service.services._capture_warranty", side_effect=RuntimeError("Synthetic failure")): with self.assertRaises(RuntimeError): intake(self) self.assertFalse(ServiceCase.objects.exists()) self.assertFalse(ServiceCaseWarrantySnapshot.objects.exists()) self.assertFalse(ServiceCaseNumberSequence.objects.exists()) self.assertEqual(intake(self).job_number, "JOB-00000001") def test_snapshot_failure_restores_existing_counter(self): intake(self) with patch("apps.service.services._capture_warranty", side_effect=ValidationError("Synthetic failure")): with self.assertRaises(ValidationError): intake(self) self.assertEqual(intake(self).job_number, "JOB-00000002") self.assertEqual(ServiceCaseWarrantySnapshot.objects.count(), 2) def test_corrupt_and_exhausted_counter_reject_without_case(self): intake(self) for value in [1, MAX_JOB_NUMBER+1]: ServiceCaseNumberSequence.objects.filter(service_center=self.center).update(next_value=value) with self.assertRaises(ValidationError): intake(self) self.assertEqual(ServiceCaseNumberSequence.objects.get(service_center=self.center).next_value, value) self.assertEqual(ServiceCase.objects.count(), 1) def test_number_not_reused_after_privileged_deletion(self): case = intake(self) # Unsupported bypass only simulates committed removal; the allocator survives. ServiceCaseWarrantySnapshot._base_manager.filter(service_case=case).delete() ServiceCase._base_manager.filter(pk=case.pk).delete() self.assertEqual(intake(self).job_number, "JOB-00000002") def test_case_facts_immutable_and_direct_save_disabled(self): case = intake(self) for field, value in [("company", self.other_company), ("service_center", self.center2), ("customer", self.presenter), ("device", self.device2), ("job_number", "JOB-99999999"), ("received_at", case.received_at-timedelta(days=1))]: case.refresh_from_db(); setattr(case, field, value) with self.subTest(field=field), self.assertRaises(ValidationError): case.full_clean() with self.assertRaises(ValidationError): case.save() def test_intake_edit_and_revision_precondition(self): case = intake(self) updated = services.update_service_case_intake(service_case=case, intake_note=" New note ") self.assertEqual(updated.intake_note, "New note") with self.assertRaises(ValidationError): services.update_service_case_intake(service_case=case, intake_note="Stale", expected_updated_at=case.updated_at) def test_cancellation_preserves_children_and_is_idempotent(self): case = intake(self) complaint = services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) condition = services.add_service_case_intake_condition(service_case=case, condition_type="OTHER", description="Synthetic condition") accessory = services.add_service_case_accessory(service_case=case, accessory_type="CABLE", quantity=2) before = (list(ServiceCaseComplaint.objects.values()), list(ServiceCaseIntakeCondition.objects.values()), list(ServiceCaseAccessory.objects.values()), list(ServiceCaseWarrantySnapshot.objects.values())) cancelled = services.cancel_service_case(service_case=case, reason=" Synthetic cancellation ") repeated = services.cancel_service_case(service_case=case, reason="Do not overwrite") self.assertEqual(cancelled.cancelled_at, repeated.cancelled_at) self.assertEqual(repeated.cancellation_reason, "Synthetic cancellation") self.assertEqual(before, (list(ServiceCaseComplaint.objects.values()), list(ServiceCaseIntakeCondition.objects.values()), list(ServiceCaseAccessory.objects.values()), list(ServiceCaseWarrantySnapshot.objects.values()))) self.assertEqual(queries.service_case_by_job_number(service_center=self.center, job_number=case.job_number), case) self.assertFalse(queries.active_service_cases_for_service_center(self.center).exists()) def test_cancelled_case_rejects_all_mutations_and_reopening(self): case = intake(self) complaint = services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) cancelled = services.cancel_service_case(service_case=case) operations = [lambda: services.update_service_case_intake(service_case=case, intake_note="Changed"), lambda: services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom), lambda: services.remove_service_case_complaint(service_case=case, complaint=complaint), lambda: services.add_service_case_accessory(service_case=case, accessory_type="BOX"), lambda: services.add_service_case_intake_condition(service_case=case, condition_type="OTHER", description="Synthetic")] for operation in operations: with self.assertRaises(ValidationError): operation() cancelled.status = "RECEIVED"; cancelled.cancelled_at = None with self.assertRaises(ValidationError): cancelled.full_clean() def test_complaint_applicability_and_duplicate(self): case = intake(self) record = services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom, customer_description=" Synthetic wording ") self.assertEqual(record.customer_description, "Synthetic wording") with self.assertRaises(ValidationError): services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) restricted = ComplaintSymptom.objects.create(code="RESTRICTED", name="Restricted") with self.assertRaises(ValidationError): services.add_service_case_complaint(service_case=case, complaint_symptom=restricted) taxonomy.set_complaint_applicability(complaint=restricted, applies_to_all_product_categories=False, product_categories=[self.category]) services.add_service_case_complaint(service_case=case, complaint_symptom=restricted) def test_removed_complaint_retains_unique_history(self): case = intake(self) record = services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) removed = services.remove_service_case_complaint(service_case=case, complaint=record) self.assertIsNotNone(removed.removed_at) self.assertEqual(case.complaints.count(), 1) with self.assertRaises(ValidationError): services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) other = intake(self) with self.assertRaises(ValidationError): services.remove_service_case_complaint(service_case=other, complaint=record) def test_master_deactivation_preserves_history_and_rejects_new_complaint(self): case = intake(self) record = services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) taxonomy.deactivate_complaint(complaint=self.symptom) catalog.deactivate_category(category=self.category) self.assertEqual(case.complaints.get(), record) with self.assertRaises(ValidationError): services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) def test_condition_accessory_types_trim_and_positive_quantity(self): case = intake(self) for kind in ServiceCaseIntakeCondition.Type.values: record = services.add_service_case_intake_condition(service_case=case, condition_type=kind, description=" Synthetic ") self.assertEqual(record.description, "Synthetic") for kind in ServiceCaseAccessory.Type.values: self.assertEqual(services.add_service_case_accessory(service_case=case, accessory_type=kind).quantity, 1) for value in [0, -1, True, 1.5]: with self.assertRaises(ValidationError): services.add_service_case_accessory(service_case=case, accessory_type="OTHER", quantity=value) with self.assertRaises(ValidationError): services.add_service_case_intake_condition(service_case=case, condition_type="OTHER", description=" ") def test_lifecycle_preserves_case_and_active_query_means_received_only(self): case = intake(self) customers.deactivate_customer(customer=self.customer) devices.deactivate_device(device=self.device) organization.deactivate_company(company=self.company) self.assertEqual(list(queries.active_service_cases_for_service_center(self.center)), [case]) self.assertEqual(case.warranty_snapshot.recorded_coverage, False) services.cancel_service_case(service_case=case) def test_warranty_snapshot_boundaries_and_immutability(self): current = warranty(self, coverage_start_date=date(2025, 1, 2), coverage_end_date=date(2025, 1, 4)) for day, expected in [(1, False), (2, True), (3, True), (4, True), (5, False)]: case = intake(self, received_at=datetime(2025, 1, day, tzinfo=tz.utc)) self.assertEqual(case.warranty_snapshot.recorded_coverage, expected) snapshot = ServiceCaseWarrantySnapshot.objects.filter(recorded_coverage=True).first() before = ServiceCaseWarrantySnapshot.objects.filter(pk=snapshot.pk).values().get() devices.clear_device_warranty_coverage(device=self.device) warranty(self, reference="SYNTHETIC-REPLACEMENT") self.assertEqual(ServiceCaseWarrantySnapshot.objects.filter(pk=snapshot.pk).values().get(), before) snapshot.coverage_reference = "Mutated" with self.assertRaises(ValidationError): snapshot.full_clean() def test_intake_never_mutates_evidence_or_coverage(self): devices.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-ONLY") warranty(self) before = (list(DevicePurchaseEvidence.objects.values()), list(DeviceWarrantyCoverage.objects.values())) intake(self) self.assertEqual(before, (list(DevicePurchaseEvidence.objects.values()), list(DeviceWarrantyCoverage.objects.values()))) def test_child_and_cancellation_rollback(self): case = intake(self) with self.assertRaises(RuntimeError), transaction.atomic(): services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) services.add_service_case_accessory(service_case=case, accessory_type="BOX") services.cancel_service_case(service_case=case) raise RuntimeError("Synthetic failure") case.refresh_from_db() self.assertEqual(case.status, "RECEIVED") self.assertFalse(case.complaints.exists()) self.assertFalse(case.accessories.exists()) def test_deletion_and_foreign_key_protection(self): case = intake(self) complaint = services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) for record in [case, case.warranty_snapshot, complaint]: with self.assertRaises(ValidationError): record.delete() with self.assertRaises(ValidationError): type(record).objects.all().delete() for record in [self.company, self.center, self.customer, self.device, self.user, self.symptom]: with self.assertRaises(ProtectedError): Collector(using="default").collect([record]) def test_database_case_checks_and_uniqueness(self): first = intake(self) for change in [dict(status="INVALID"), dict(status="CANCELLED"), dict(intake_channel="INVALID"), dict(job_number="bad")]: with self.subTest(change=change), self.assertRaises(IntegrityError), transaction.atomic(): ServiceCase.objects.filter(pk=first.pk).update(**change) second = intake(self) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCase.objects.filter(pk=second.pk).update(job_number=first.job_number) def test_database_child_and_snapshot_constraints(self): case = intake(self) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseAccessory.objects.bulk_create([ServiceCaseAccessory(service_case=case, accessory_type="BOX", quantity=0)]) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseIntakeCondition.objects.bulk_create([ServiceCaseIntakeCondition(service_case=case, condition_type="DIAGNOSIS", description="Synthetic")]) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseWarrantySnapshot.objects.filter(service_case=case).update(recorded_coverage=True) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseWarrantySnapshot.objects.bulk_create([ServiceCaseWarrantySnapshot(service_case=case)]) services.add_service_case_complaint(service_case=case, complaint_symptom=self.symptom) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseComplaint.objects.bulk_create([ServiceCaseComplaint(service_case=case, complaint_symptom=self.symptom)]) def test_queries_are_lazy_ordered_scoped_and_joined(self): first = intake(self); second = intake(self) foreign = intake(self, company=self.other_company, service_center=self.other_center, customer=self.outsider, device=self.device2) with self.assertNumQueries(0): rows = queries.service_cases_for_customer(self.customer) with self.assertNumQueries(1): self.assertEqual([row.pk for row in rows if str(row.device) and str(row.created_by) and str(row.warranty_snapshot)], [second.pk, first.pk]) self.assertEqual(list(queries.service_cases_for_device(self.device)), [second, first]) self.assertEqual(queries.service_case_by_job_number(service_center=self.other_center, job_number=first.job_number), foreign) self.assertEqual(set(queries.search_service_cases(company=self.company, query="Synthetic")), {first, second}) self.assertFalse(queries.service_cases_for_customer(None).exists()) def test_identifier_search_remains_scoped_after_correction(self): devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(70)) case = intake(self) devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(71)) self.assertEqual(list(queries.search_service_cases(company=self.company, query=synthetic_imei(70))), [case]) self.assertFalse(queries.search_service_cases(company=self.other_company, query=synthetic_imei(70)).exists()) self.assertEqual(case.device_id, self.device.pk) def test_no_authorization_adapter_or_creator_inference(self): case = intake(self) self.assertFalse(is_authorized(user=self.user, permission="service.view_servicecase", target=case)) self.assertEqual(case.company_id, self.company.pk) class IntakeAdminTests(TestCase): def setUp(self): setup(self) self.user.is_staff = self.user.is_superuser = True self.user.save(update_fields=["is_staff", "is_superuser"]) self.client.force_login(self.user) def url(self, case): return reverse("admin:service_servicecase_change", args=[case.pk]) def test_admin_create_uses_services_and_request_actor(self): response = self.client.post(reverse("admin:service_servicecase_add"), { "company": self.company.pk, "service_center": self.center.pk, "customer": self.customer.pk, "device": self.device.pk, "intake_channel": "WALK_IN", "received_at_0": "2025-01-01", "received_at_1": "12:00:00", "customer_reference": "Synthetic", "intake_note": "Synthetic note", "status": "CANCELLED", "job_number": "FORGED", "_save": "Save"}) self.assertEqual(response.status_code, 302) case = ServiceCase.objects.get() self.assertEqual(case.job_number, "JOB-00000001") self.assertEqual(case.status, "RECEIVED") self.assertEqual(case.created_by, self.user) self.assertIsNotNone(case.warranty_snapshot) def test_admin_stale_cancelled_page_cannot_restore_case(self): case = intake(self) token = self.client.get(self.url(case)).context["adminform"].form.initial["revision"] services.cancel_service_case(service_case=case) response = self.client.post(self.url(case), {"revision": token, "intake_note": "Stale", "status": "RECEIVED", "_save": "Save"}) self.assertEqual(response.status_code, 403) case.refresh_from_db(); self.assertEqual(case.status, "CANCELLED") def test_admin_stale_metadata_and_forged_identity_rejected(self): case = intake(self) token = self.client.get(self.url(case)).context["adminform"].form.initial["revision"] services.update_service_case_intake(service_case=case, intake_note="Newer") response = self.client.post(self.url(case), {"revision": token, "intake_note": "Old", "device": self.device2.pk, "_save": "Save"}) self.assertEqual(response.status_code, 200) case.refresh_from_db(); self.assertEqual(case.intake_note, "Newer"); self.assertEqual(case.device, self.device) def test_admin_current_metadata_edit_and_cancellation_action(self): case = intake(self) token = self.client.get(self.url(case)).context["adminform"].form.initial["revision"] response = self.client.post(self.url(case), {"revision": token, "customer_reference": "New reference", "intake_note": "Synthetic", "_save": "Save"}) self.assertEqual(response.status_code, 302) case.refresh_from_db(); self.assertEqual(case.customer_reference, "New reference") self.client.post(reverse("admin:service_servicecase_changelist"), {"action": "cancel_selected", "_selected_action": str(case.pk)}) case.refresh_from_db(); self.assertEqual(case.status, "CANCELLED") def test_admin_child_add_remove_and_cancelled_rejection(self): case = intake(self) response = self.client.post(reverse("admin:service_servicecasecomplaint_add"), {"service_case": case.pk, "complaint_symptom": self.symptom.pk, "customer_description": "Synthetic", "_save": "Save"}) self.assertEqual(response.status_code, 302) record = case.complaints.get() self.client.post(reverse("admin:service_servicecasecomplaint_changelist"), {"action": "remove_selected", "_selected_action": str(record.pk)}) record.refresh_from_db(); self.assertIsNotNone(record.removed_at) services.cancel_service_case(service_case=case) self.client.post(reverse("admin:service_servicecaseaccessory_add"), {"service_case": case.pk, "accessory_type": "BOX", "quantity": 1, "_save": "Save"}) self.assertFalse(case.accessories.exists()) def test_admin_snapshot_readonly_and_deletion_disabled(self): case = intake(self) self.assertEqual(self.client.post(reverse("admin:service_servicecase_delete", args=[case.pk]), {"post": "yes"}).status_code, 403) snapshot = case.warranty_snapshot url = reverse("admin:service_servicecasewarrantysnapshot_change", args=[snapshot.pk]) self.assertEqual(self.client.get(url).status_code, 200) self.assertEqual(self.client.post(url, {}).status_code, 403) def test_admin_csrf_and_view_only_permissions(self): client = Client(enforce_csrf_checks=True); client.force_login(self.user) self.assertEqual(client.post(reverse("admin:service_servicecase_add"), {}).status_code, 403) user = get_user_model().objects.create_user(username="synthetic-viewer", is_staff=True) user.user_permissions.add(Permission.objects.get(content_type__app_label="service", codename="view_servicecase")) self.client.force_login(user) case = intake(self) self.assertEqual(self.client.get(self.url(case)).status_code, 200) self.assertEqual(self.client.post(self.url(case), {}).status_code, 403)