"""Synthetic custody, authorization, immutable evidence and query contracts.""" from datetime import timedelta from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission, Group from django.core.exceptions import ValidationError, PermissionDenied from django.db import IntegrityError, transaction from django.test import TestCase from django.utils import timezone from apps.access import services as access from apps.access.models import Role from apps.organization import assignment_services as postings, services as organization from apps.customers import services as customers from apps.devices import services as devices from apps.devices.models import CustomerDeviceRelationship from . import test_quality_control as qc, test_repair as repair, test_diagnosis as diagnosis from . import handover_services as services, handover_queries as queries, engineer_queries from .test_engineer_assignment import assign, reassign, unassign from .tests import intake from .services import add_service_case_accessory, add_service_case_complaint, cancel_service_case from .models import (ServiceCase, ServiceEngineerAssignment, ServiceCaseDeliveryRelease, ServiceCaseHandover, ServiceCaseHandoverAccessory, ServiceCaseClosure, ServiceQualityControl, ServiceRepairExecution) def setup_handover(test, *, accessories=True): original = qc.assign def with_accessory(obj): test.received = [] if accessories: test.received.append(add_service_case_accessory(service_case=obj.case, accessory_type="CABLE", quantity=2, description="Synthetic cable")) return original(obj) with patch.object(qc, "assign", with_accessory): qc.setup_qc(test) test.passed = qc.pass_qc(test, qc.prepared(test, filled=True)) test.case.refresh_from_db() test.handover_permission = Permission.objects.get(content_type__app_label="service", codename="handover_servicecase") test.close_permission = Permission.objects.get(content_type__app_label="service", codename="close_servicecase") test.delivery_role = Role.objects.create(code="SYNTHETIC-DELIVERY", name="Synthetic delivery") access.set_role_permissions(role=test.delivery_role, permissions=[test.handover_permission, test.close_permission]) test.actor = get_user_model().objects.create_user(username="synthetic-delivery") test.delivery_path = postings.create_assignment(user=test.actor, company=test.company, region=test.region, service_center=test.center) test.delivery_role_assignment = access.create_role_assignment(user=test.actor, role=test.delivery_role, organization_assignment=test.delivery_path) test.actor2 = get_user_model().objects.create_user(username="synthetic-delivery-2") test.delivery_path2 = postings.create_assignment(user=test.actor2, company=test.company) access.create_role_assignment(user=test.actor2, role=test.delivery_role, organization_assignment=test.delivery_path2) def ready(test, **kwargs): return services.mark_service_case_ready_for_delivery(**dict(dict(service_case=test.case, actor=test.actor, expected_device=test.device, expected_device_identity=queries.device_identity_fingerprint(test.device)), **kwargs)) def released(test): result = ready(test) test.case.refresh_from_db() return result def handover(test, **kwargs): return services.handover_service_case(**dict(dict(service_case=test.case, actor=test.actor, expected_device=test.device, expected_device_identity=queries.device_identity_fingerprint(test.device), recipient_type="CUSTOMER", verification_method="CUSTOMER_RECORD", accessories=[dict(intake_accessory=a, returned_quantity=a.quantity) for a in test.received]), **kwargs)) def delivered(test): released(test) row = handover(test) test.case.refresh_from_db() return row def close(test, **kwargs): return services.close_service_case(**dict(dict(service_case=test.case, actor=test.actor), **kwargs)) def invariants(test): for case in ServiceCase.objects.all(): test.assertEqual(ServiceEngineerAssignment.objects.filter(service_case=case, ended_at=None).count(), int(case.status not in ("RECEIVED", "CANCELLED", "DELIVERED", "CLOSED"))) 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 HandoverTests(TestCase): @classmethod def setUpTestData(cls): setup_handover(cls) def tearDown(self): invariants(self) def test_qc_pass_is_neither_release_nor_delivery(self): self.assertEqual(self.case.status, "QC_PASSED") self.assertFalse(ServiceCaseDeliveryRelease.objects.exists()) self.assertFalse(ServiceCaseHandover.objects.exists()) with self.assertRaises(ValidationError): handover(self) def test_explicit_release_captures_latest_qc_identity_actor_time(self): row = released(self) self.assertEqual(self.case.status, "READY_FOR_DELIVERY") self.assertEqual(row.quality_control, self.passed) self.assertEqual(row.readied_by, self.actor) self.assertEqual(row.device, self.device) self.assertTrue(timezone.is_aware(row.readied_at)) self.assertFalse(ServiceCaseHandover.objects.exists()) self.assertTrue(engineer_queries.current_engineer_assignment(self.case)) def test_release_requires_qc_pass(self): with transaction.atomic(), self.assertRaises(ValidationError): ServiceCase.objects.filter(pk=self.case.pk).update(status="REPAIRED") ready(self) # Restore the fixture-only corruption; supported transitions cannot create it. ServiceCase.objects.filter(pk=self.case.pk).update(status="QC_PASSED") def test_duplicate_and_stale_release(self): old = self.case.updated_at released(self) for kwargs in ({}, {"expected_updated_at": old}): with self.assertRaises(ValidationError): ready(self, **kwargs) self.assertEqual(ServiceCaseDeliveryRelease.objects.count(), 1) def test_customer_delivery_snapshots_and_assignment_end_are_atomic(self): row = delivered(self) self.assertEqual(self.case.status, "DELIVERED") self.assertEqual(row.recipient_customer, self.customer) self.assertEqual(row.recipient_name, self.customer.display_name) self.assertEqual(row.recipient_mobile, self.customer.primary_mobile) self.assignment.refresh_from_db() self.assertEqual(self.assignment.ended_by, self.actor) self.assertEqual(self.assignment.ended_at, row.delivered_at) self.assertIn("delivered", self.assignment.end_reason) self.assertFalse(ServiceCaseClosure.objects.exists()) def test_representative_normalization_no_customer_or_ownership_creation(self): from apps.customers.models import Customer count, ownership = Customer.objects.count(), list(CustomerDeviceRelationship.objects.values()) released(self) row = handover(self, recipient_type="AUTHORIZED_REPRESENTATIVE", recipient_name=" Synthetic Collector ", recipient_mobile="+1 (202) 555-0123", recipient_relationship=" Authorized colleague ", verification_method="AUTHORIZED_REPRESENTATIVE", verification_reference=" SYNTHETIC-REF ") self.assertIsNone(row.recipient_customer) self.assertEqual(row.recipient_name, "Synthetic Collector") self.assertEqual(row.recipient_mobile, "+12025550123") self.assertEqual(row.verification_reference, "SYNTHETIC-REF") self.assertEqual(Customer.objects.count(), count) self.assertEqual(list(CustomerDeviceRelationship.objects.values()), ownership) def test_representative_requires_each_explicit_fact(self): released(self) values = dict(recipient_type="AUTHORIZED_REPRESENTATIVE", recipient_name="Synthetic Collector", recipient_mobile="2025550123", recipient_relationship="Colleague", verification_method="AUTHORIZED_REPRESENTATIVE", verification_reference="SYNTHETIC") for field in ("recipient_name", "recipient_mobile", "recipient_relationship", "verification_method", "verification_reference"): with self.subTest(field=field), self.assertRaises(ValidationError): handover(self, **dict(values, **{field: " "})) def test_unknown_recipient_and_verification_rejected(self): released(self) for values in ({"recipient_type": "OTHER"}, {"verification_method": "OTP"}, {"recipient_mobile": None}, {"recipient_name": "Different"}): with self.subTest(values=values), self.assertRaises(ValidationError): handover(self, **values) def test_manual_document_check_requires_bounded_minimal_references(self): released(self) values = dict(verification_method="MANUAL_DOCUMENT_CHECK", recipient_identity_type="PASSPORT", recipient_identity_reference="SYNTHETIC-REVIEW", verification_reference="SYNTHETIC-DESK") for field in ("recipient_identity_type", "recipient_identity_reference", "verification_reference"): with self.subTest(field=field), self.assertRaises(ValidationError): handover(self, **dict(values, **{field: ""})) row = handover(self, **values) self.assertEqual(row.recipient_identity_reference, "SYNTHETIC-REVIEW") def test_reference_and_note_lengths_are_bounded(self): released(self) for field, length in (("verification_reference", 121), ("recipient_identity_reference", 65), ("note", 2001)): with self.subTest(field=field), self.assertRaises(ValidationError): handover(self, **{field: "x" * length}) def test_document_reference_not_accepted_for_non_document_method(self): released(self) with self.assertRaises(ValidationError): handover(self, recipient_identity_reference="SYNTHETIC-REF") def test_customer_edit_does_not_rewrite_delivered_snapshot(self): row = delivered(self) original = row.recipient_name customers.update_customer(customer=self.customer, full_name="Changed synthetic customer") row.refresh_from_db() self.assertEqual(row.recipient_name, original) close(self) def test_customer_change_rejects_stale_confirmation(self): released(self) old = self.customer.updated_at customers.update_customer(customer=self.customer, full_name="Changed synthetic customer") with self.assertRaises(ValidationError): handover(self, expected_customer_updated_at=old) def test_complete_accessory_return_preserves_intake(self): original = self.received[0] row = delivered(self) acknowledgement = row.accessories.get() self.assertEqual(acknowledgement.returned_quantity, 2) original.refresh_from_db() self.assertEqual(original.quantity, 2) self.assertEqual(original.description, "Synthetic cable") def test_shortage_with_reason_and_zero_return_allowed(self): released(self) row = handover(self, accessories=[dict(intake_accessory=self.received[0], returned_quantity=0, condition_note="Synthetic discrepancy acknowledged")]) self.assertEqual(row.accessories.get().returned_quantity, 0) def test_invalid_quantities_and_unexplained_shortage_roll_back(self): released(self) for quantity in (-1, 3, 1, True, "2"): with self.subTest(quantity=quantity), self.assertRaises(ValidationError): handover(self, accessories=[dict(intake_accessory=self.received[0], returned_quantity=quantity)]) self.assertFalse(ServiceCaseHandover.objects.exists()) self.assertTrue(ServiceEngineerAssignment.objects.filter(pk=self.assignment.pk, ended_at=None).exists()) def test_missing_duplicate_and_foreign_accessories_rejected(self): other_case = intake(self) foreign = add_service_case_accessory(service_case=other_case, accessory_type="CASE") released(self) item = dict(intake_accessory=self.received[0], returned_quantity=2) for items in ([], None, [item, item], [dict(intake_accessory=foreign, returned_quantity=1)]): with self.subTest(items=items), self.assertRaises(ValidationError): handover(self, accessories=items) def test_stale_accessory_revision_rejected(self): released(self) self.received[0].updated_at -= timedelta(seconds=1) with self.assertRaises(ValidationError): handover(self) def test_wrong_device_or_identity_digest_rejected(self): for values in ({"expected_device": self.device2}, {"expected_device_identity": "0" * 64}): with self.subTest(values=values), self.assertRaises(ValidationError): ready(self, **values) def test_identity_change_after_release_requires_explicit_future_recovery(self): released(self) devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-NEW") # Even a freshly confirmed digest cannot silently replace the release evidence. with self.assertRaises(ValidationError): handover(self) def test_superseding_repair_invalidates_release(self): with self.assertRaises(ValidationError), transaction.atomic(): import uuid newer = ServiceRepairExecution.objects.get(pk=self.execution.pk) newer.pk = uuid.uuid4() newer.started_at += timedelta(microseconds=1) ServiceRepairExecution.objects.bulk_create([newer]) ready(self) def test_open_repair_prevents_handover(self): released(self) with self.assertRaises(ValidationError), transaction.atomic(): ServiceRepairExecution.objects.filter(pk=self.execution.pk).update(status="OPEN", completed_at=None, completed_by=None, outcome=None) handover(self) def test_latest_failed_qc_prevents_handover(self): released(self) with self.assertRaises(ValidationError), transaction.atomic(): ServiceQualityControl.objects.filter(pk=self.passed.pk).update(outcome="FAILED", failure_note="Synthetic failure") handover(self) def test_closure_rejects_remaining_assignment(self): delivered(self) with self.assertRaises(ValidationError), transaction.atomic(): ServiceEngineerAssignment.objects.filter(pk=self.assignment.pk).update(ended_at=None, ended_by=None, end_reason="") close(self) def test_closure_rejects_missing_acknowledgement(self): row = delivered(self) with self.assertRaises(ValidationError), transaction.atomic(): row.accessories.all()._raw_delete("default") close(self) def test_closure_rejects_corrupted_qc(self): delivered(self) with self.assertRaises(ValidationError), transaction.atomic(): ServiceQualityControl.objects.filter(pk=self.passed.pk).update(outcome="FAILED", failure_note="Synthetic failure") close(self) def test_naive_and_future_evidence_timestamps_rejected(self): import uuid row = delivered(self) row.pk = uuid.uuid4() for value in (timezone.now().replace(tzinfo=None), timezone.now() + timedelta(days=1)): row.delivered_at = value with self.assertRaisesMessage(ValidationError, "timestamps must be aware"): row.clean() def test_inactive_customer_prevents_delivery(self): released(self) customers.deactivate_customer(customer=self.customer) with self.assertRaises(ValidationError): handover(self) def test_inactive_device_prevents_delivery(self): released(self) devices.deactivate_device(device=self.device) with self.assertRaises(ValidationError): handover(self) def test_failed_or_superseded_qc_prevents_release(self): with self.assertRaises(ValidationError), transaction.atomic(): ServiceQualityControl.objects.filter(pk=self.passed.pk).update(outcome="FAILED", failure_note="Synthetic failure") ready(self) def test_closure_is_separate_attributed_and_terminal(self): row = delivered(self) closure = close(self, note=" Synthetic closure ") self.case.refresh_from_db() self.assertEqual(self.case.status, "CLOSED") self.assertEqual(closure.handover, row) self.assertEqual(closure.closed_by, self.actor) self.assertGreaterEqual(closure.closed_at, row.delivered_at) self.assertEqual(closure.note, "Synthetic closure") for operation in (lambda: ready(self), lambda: handover(self), lambda: close(self), lambda: cancel_service_case(service_case=self.case, cancelled_by=self.user), lambda: assign(self), lambda: reassign(self), lambda: unassign(self), lambda: diagnosis.begin(self), lambda: repair.begin(self), lambda: qc.pending(self), lambda: add_service_case_complaint(service_case=self.case, complaint_symptom=self.symptom)): with self.assertRaises(ValidationError): operation() def test_closure_requires_delivered(self): with self.assertRaises(ValidationError): close(self) released(self) with self.assertRaises(ValidationError): close(self) def test_all_post_qc_states_reject_cancellation(self): for next_step in (lambda: None, lambda: released(self), lambda: handover(self), lambda: close(self)): next_step() self.case.refresh_from_db() with self.assertRaises(ValidationError): cancel_service_case(service_case=self.case, cancelled_by=self.user) def test_duplicate_and_stale_handover_rejected(self): released(self) old = self.case.updated_at handover(self) self.case.refresh_from_db() for values in ({}, {"expected_updated_at": old}): with self.assertRaises(ValidationError): handover(self, **values) def test_duplicate_and_stale_closure_rejected(self): delivered(self) old = self.case.updated_at close(self) self.case.refresh_from_db() for values in ({}, {"expected_updated_at": old}): with self.assertRaises(ValidationError): close(self, **values) def test_all_evidence_immutable_and_undeletable(self): row = delivered(self) closure = close(self) for obj in (row.delivery_release, row, row.accessories.get(), closure): for operation in (obj.save, obj.delete, lambda obj=obj: type(obj).objects.filter(pk=obj.pk).delete()): with self.assertRaises(ValidationError): operation() obj.created_at -= timedelta(seconds=1) with self.assertRaises(ValidationError): obj.full_clean() def test_handover_rolls_back_late_case_write_failure(self): released(self) with patch.object(ServiceCase, "_persist", side_effect=ValidationError("Synthetic late failure")), self.assertRaises(ValidationError): handover(self) self.assertFalse(ServiceCaseHandover.objects.exists()) self.assertFalse(ServiceCaseHandoverAccessory.objects.exists()) self.assignment.refresh_from_db() self.assertIsNone(self.assignment.ended_at) def test_assignment_termination_failure_rolls_back_handover(self): released(self) with patch.object(services, "_close", side_effect=ValidationError("Synthetic assignment failure")), self.assertRaises(ValidationError): handover(self) self.assertFalse(ServiceCaseHandover.objects.exists()) self.assertFalse(ServiceCaseHandoverAccessory.objects.exists()) def test_closure_failure_rolls_back_record(self): delivered(self) with patch.object(ServiceCase, "_persist", side_effect=ValidationError("Synthetic late failure")), self.assertRaises(ValidationError): close(self) self.assertFalse(ServiceCaseClosure.objects.exists()) self.case.refresh_from_db() self.assertEqual(self.case.status, "DELIVERED") def test_database_recipient_and_verification_constraints(self): row = delivered(self) for values in ({"recipient_type": "OTHER"}, {"verification_method": "OTP"}, {"recipient_customer": None}, {"recipient_name": ""}): with self.subTest(values=values), self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseHandover.objects.filter(pk=row.pk).update(**values) def test_database_duplicate_accessory_and_negative_quantity(self): row = delivered(self).accessories.get() with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseHandoverAccessory.objects.bulk_create([ServiceCaseHandoverAccessory(handover=row.handover, intake_accessory=row.intake_accessory, returned_quantity=1)]) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseHandoverAccessory.objects.filter(pk=row.pk).update(returned_quantity=-1) def test_database_one_handover_and_one_closure(self): row = delivered(self) closure = close(self) for obj in (row, closure): import uuid obj.pk = uuid.uuid4() with self.assertRaises(IntegrityError), transaction.atomic(): type(obj).objects.bulk_create([obj]) def test_database_closure_timestamp_coherence(self): delivered(self) row = close(self) with self.assertRaises(IntegrityError), transaction.atomic(): ServiceCaseClosure.objects.filter(pk=row.pk).update(closed_at=timezone.now() + timedelta(days=1)) class HandoverAuthorizationTests(TestCase): @classmethod def setUpTestData(cls): setup_handover(cls) def test_company_scope_allows_all_three_operations(self): released(self) handover(self, actor=self.actor2) self.case.refresh_from_db() close(self, actor=self.actor2) def test_inactive_actor_rejected(self): self.actor.is_active = False self.actor.save() with self.assertRaises(ValidationError): ready(self) def test_revoked_organization_path_denied(self): postings.deactivate_assignment(assignment=self.delivery_path) with self.assertRaises(PermissionDenied): ready(self) def test_revoked_role_assignment_denied(self): access.deactivate_role_assignment(assignment=self.delivery_role_assignment) with self.assertRaises(PermissionDenied): ready(self) def test_wrong_center_and_wrong_company_denied(self): access.deactivate_role_assignment(assignment=self.delivery_role_assignment) for company, center in ((self.company, self.center2), (self.other_company, self.other_center)): path = postings.create_assignment(user=self.actor, company=company, region=center.region, service_center=center) access.create_role_assignment(user=self.actor, role=self.delivery_role, organization_assignment=path) with self.assertRaises(PermissionDenied): ready(self) def test_permission_scope_cannot_mix_paths(self): access.deactivate_role_assignment(assignment=self.delivery_role_assignment) path = postings.create_assignment(user=self.actor, company=self.other_company) access.create_role_assignment(user=self.actor, role=self.delivery_role, organization_assignment=path) self.assertTrue(self.delivery_path.is_active) with self.assertRaises(PermissionDenied): ready(self) def test_staff_group_direct_permissions_do_not_grant_business_access(self): access.deactivate_role_assignment(assignment=self.delivery_role_assignment) self.actor.is_staff = True self.actor.save() self.actor.user_permissions.add(self.handover_permission, self.close_permission) group = Group.objects.create(name="Synthetic delivery") group.permissions.add(self.handover_permission, self.close_permission) self.actor.groups.add(group) with self.assertRaises(PermissionDenied): ready(self) def test_established_active_superuser_bypass_without_posting(self): self.user.is_staff = self.user.is_superuser = True self.user.save() ready(self, actor=self.user) self.case.refresh_from_db() handover(self, actor=self.user) self.case.refresh_from_db() close(self, actor=self.user) def test_handover_permission_does_not_grant_closure(self): delivered(self) access.set_role_permissions(role=self.delivery_role, permissions=[self.handover_permission]) with self.assertRaises(PermissionDenied): close(self) def test_closure_permission_does_not_grant_handover(self): access.set_role_permissions(role=self.delivery_role, permissions=[self.close_permission]) with self.assertRaises(PermissionDenied): ready(self) def test_inactive_company_denied(self): organization.deactivate_company(company=self.company) with self.assertRaises((PermissionDenied, ValidationError)): ready(self) class HandoverQueryTests(TestCase): @classmethod def setUpTestData(cls): setup_handover(cls) def test_ready_queue_lazy_joined_one_query(self): released(self) with self.assertNumQueries(0): rows = queries.ready_for_delivery_cases_for_center(self.center) with self.assertNumQueries(1): for row in rows: self.assertEqual(row.customer.pk, self.customer.pk) str(row.device); str(row.service_center); str(row.delivery_release.readied_by) def test_delivered_queue_joined_one_query(self): delivered(self) with self.assertNumQueries(1): for row in queries.delivered_cases_for_center(self.center): str(row.customer); str(row.device); str(row.service_center); str(row.handover.handed_over_by) def test_handover_detail_joined_one_query(self): delivered(self) with self.assertNumQueries(1): row = queries.get_service_case_handover(self.case) str(row.recipient_customer); str(row.handed_over_by); str(row.service_case.device); str(row.service_case.service_center) def test_accessory_display_joined_one_query(self): row = delivered(self) with self.assertNumQueries(1): for item in queries.handover_accessories(row): str(item.handover); str(item.intake_accessory); self.assertEqual(item.returned_quantity, 2) def test_closure_detail_joined_one_query(self): delivered(self) close(self) with self.assertNumQueries(1): row = queries.get_service_case_closure(self.case) str(row.closed_by); str(row.handover.handed_over_by); str(row.service_case.customer); str(row.service_case.service_center); str(row.service_case.device) def test_sql_authorization_and_center_isolation(self): released(self) self.assertEqual(list(queries.ready_for_delivery_cases_for_center(self.center, actor=self.actor)), [self.case]) self.assertFalse(queries.ready_for_delivery_cases_for_center(self.other_center, actor=self.actor).exists()) self.assertFalse(queries.ready_for_delivery_cases_for_center(self.center2, actor=self.actor).exists()) access.deactivate_role_assignment(assignment=self.delivery_role_assignment) self.assertFalse(queries.ready_for_delivery_cases_for_center(self.center, actor=self.actor).exists()) def test_customer_history_and_closed_queue(self): row = delivered(self) close(self) self.assertEqual(list(queries.handover_history_for_customer(self.customer)), [row]) self.assertFalse(queries.handover_history_for_customer(self.outsider).exists()) self.assertEqual(list(queries.closed_cases_for_center(self.center)), [self.case]) self.assertFalse(queries.delivered_cases_for_center(self.center).exists()) self.assertFalse(engineer_queries.assigned_service_cases_for_engineer(self.engineer).exists()) def test_scoped_ready_queue_is_lazy_and_one_sql_query(self): released(self) with self.assertNumQueries(0): rows = queries.ready_for_delivery_cases_for_center(self.center, actor=self.actor) with self.assertNumQueries(1): self.assertEqual([row.customer.pk for row in rows], [self.customer.pk]) def test_representative_history_belongs_to_presenting_customer(self): released(self) row = handover(self, recipient_type="AUTHORIZED_REPRESENTATIVE", recipient_name="Synthetic Collector", recipient_mobile="2025550123", recipient_relationship="Colleague", verification_method="AUTHORIZED_REPRESENTATIVE", verification_reference="SYNTHETIC") self.assertEqual(list(queries.handover_history_for_customer(self.customer)), [row]) class NoAccessoryHandoverTests(TestCase): @classmethod def setUpTestData(cls): setup_handover(cls, accessories=False) def test_explicit_empty_reconciliation_can_deliver_and_close(self): row = delivered(self) self.assertFalse(row.accessories.exists()) close(self) invariants(self)