"""Real PostgreSQL lock waits, both serialization orders and atomic rollback.""" from unittest.mock import patch from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError, PermissionDenied from django.test import TransactionTestCase, Client from apps.access import services as access from apps.organization import services as organization, assignment_services as postings, test_assignment_concurrency as concurrency from apps.customers import services as customers from apps.devices import services as devices from . import handover_services as services, handover_queries as queries from .models import ServiceCase, ServiceCaseHandover, ServiceCaseClosure, ServiceEngineerAssignment from .test_engineer_assignment import reassign from .test_handover import setup_handover, ready, released, handover, delivered, close, invariants from .test_handover_admin import staff, form_data class HandoverConcurrencyTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): setup_handover(self) def tearDown(self): invariants(self) def race(self, first, second, expected="validation"): def contender(): try: return second() except PermissionDenied as error: raise ValidationError("Business permission revoked") from error self.run_concurrent(first, contender, expected=expected) def reject_and_hold(self, action): with self.assertRaises((ValidationError, PermissionDenied)): action() ServiceCase.objects.select_for_update().get(pk=self.case.pk) def test_two_users_release_same_case(self): self.race(lambda: ready(self), lambda: ready(self, actor=self.actor2)) def test_two_users_handover_same_case(self): released(self) self.race(lambda: handover(self), lambda: handover(self, actor=self.actor2)) self.assertEqual(ServiceCaseHandover.objects.get().handed_over_by, self.actor) def test_duplicate_recipient_submission(self): released(self) self.race(lambda: handover(self), lambda: handover(self)) self.assertEqual(ServiceCaseHandover.objects.count(), 1) def test_accessory_shortage_then_full_return_cannot_overwrite(self): released(self) self.race(lambda: handover(self, accessories=[dict(intake_accessory=self.received[0], returned_quantity=1, condition_note="Synthetic shortage")]), lambda: handover(self)) self.assertEqual(ServiceCaseHandover.objects.get().accessories.get().returned_quantity, 1) def test_accessory_full_return_then_shortage_cannot_overwrite(self): released(self) self.race(lambda: handover(self), lambda: handover(self, accessories=[dict(intake_accessory=self.received[0], returned_quantity=1, condition_note="Synthetic shortage")])) self.assertEqual(ServiceCaseHandover.objects.get().accessories.get().returned_quantity, 2) def test_role_revocation_then_handover_denied(self): released(self) self.race(lambda: access.deactivate_role_assignment(assignment=self.delivery_role_assignment), lambda: handover(self)) def test_handover_then_role_revocation_preserves_evidence(self): released(self) self.race(lambda: handover(self), lambda: access.deactivate_role_assignment(assignment=self.delivery_role_assignment), "success") def test_posting_revocation_then_handover_denied(self): released(self) self.race(lambda: postings.deactivate_assignment(assignment=self.delivery_path), lambda: handover(self)) def test_handover_then_posting_revocation_preserves_evidence(self): released(self) self.race(lambda: handover(self), lambda: postings.deactivate_assignment(assignment=self.delivery_path), "success") def test_permission_removal_then_handover_denied(self): released(self) self.race(lambda: access.set_role_permissions(role=self.delivery_role, permissions=[]), lambda: handover(self)) def test_handover_then_permission_removal_preserves_evidence(self): released(self) self.race(lambda: handover(self), lambda: access.set_role_permissions(role=self.delivery_role, permissions=[]), "success") def test_company_deactivation_then_handover_denied(self): released(self) self.race(lambda: organization.deactivate_company(company=self.company), lambda: handover(self)) def test_handover_then_company_deactivation_preserves_evidence(self): released(self) self.race(lambda: handover(self), lambda: organization.deactivate_company(company=self.company), "success") def test_center_deactivation_then_handover_denied(self): released(self) self.race(lambda: organization.deactivate_service_center(service_center=self.center), lambda: handover(self)) def test_handover_then_center_deactivation_preserves_evidence(self): released(self) self.race(lambda: handover(self), lambda: organization.deactivate_service_center(service_center=self.center), "success") def test_customer_deactivation_then_handover_denied(self): released(self) self.race(lambda: customers.deactivate_customer(customer=self.customer), lambda: handover(self)) def test_handover_then_customer_deactivation_preserves_snapshot(self): released(self) self.race(lambda: handover(self), lambda: customers.deactivate_customer(customer=self.customer), "success") def test_customer_edit_then_old_confirmation_denied(self): released(self) version = self.customer.updated_at self.race(lambda: customers.update_customer(customer=self.customer, full_name="Changed synthetic customer"), lambda: handover(self, expected_customer_updated_at=version)) def test_handover_then_customer_edit_preserves_snapshot(self): released(self) original = self.customer.display_name self.race(lambda: handover(self), lambda: customers.update_customer(customer=self.customer, full_name="Changed synthetic customer"), "success") self.assertEqual(ServiceCaseHandover.objects.get().recipient_name, original) def test_device_deactivation_then_handover_denied(self): released(self) self.race(lambda: devices.deactivate_device(device=self.device), lambda: handover(self)) def test_handover_then_device_deactivation_preserves_evidence(self): released(self) self.race(lambda: handover(self), lambda: devices.deactivate_device(device=self.device), "success") def correct_identifier(self): return devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-CORRECTED") def test_identifier_change_then_handover_denied(self): released(self) fingerprint = queries.device_identity_fingerprint(self.device) self.race(self.correct_identifier, lambda: handover(self, expected_device_identity=fingerprint)) def test_handover_then_identifier_change_preserves_release(self): release = released(self) self.race(lambda: handover(self), self.correct_identifier, "success") release.refresh_from_db() self.assertNotEqual(release.device_identity, queries.device_identity_fingerprint(self.device)) def test_user_deactivation_then_handover_denied(self): released(self) def deactivate(): user = get_user_model().objects.get(pk=self.actor.pk) user.is_active = False user.save() self.race(deactivate, lambda: handover(self)) def test_handover_then_user_deactivation_preserves_evidence(self): released(self) def deactivate(): user = get_user_model().objects.get(pk=self.actor.pk) user.is_active = False user.save() self.race(lambda: handover(self), deactivate, "success") def test_handover_then_stale_closure_rejected(self): released(self) self.race(lambda: handover(self), lambda: close(self)) self.assertFalse(ServiceCaseClosure.objects.exists()) def test_premature_closure_denied_then_handover_succeeds(self): released(self) self.race(lambda: self.reject_and_hold(lambda: close(self)), lambda: handover(self), "success") def test_two_users_close_same_case(self): delivered(self) self.race(lambda: close(self), lambda: close(self, actor=self.actor2)) self.assertEqual(ServiceCaseClosure.objects.count(), 1) def test_closure_then_assignment_mutation_denied(self): delivered(self) self.race(lambda: close(self), lambda: reassign(self)) def test_assignment_mutation_denied_then_closure_succeeds(self): delivered(self) self.race(lambda: self.reject_and_hold(lambda: reassign(self)), lambda: close(self), "success") def test_closure_revocation_then_close_denied(self): delivered(self) self.race(lambda: access.deactivate_role_assignment(assignment=self.delivery_role_assignment), lambda: close(self)) def test_close_then_revocation_preserves_closure(self): delivered(self) self.race(lambda: close(self), lambda: access.deactivate_role_assignment(assignment=self.delivery_role_assignment), "success") def admin_form(self, operation): staff(self) client = Client() client.force_login(self.actor) token = client.get(self.url).context["form"].initial["revision"] data = form_data(self, operation, token) return client, data def test_concurrent_handover_rejects_stale_admin_post(self): released(self) client, data = self.admin_form("handover") def stale_post(): response = client.post(self.url, data) self.assertEqual(response.status_code, 200) self.assertContains(response, "changed; reload") self.race(lambda: handover(self), stale_post, "success") self.assertEqual(ServiceCaseHandover.objects.count(), 1) def test_concurrent_closure_rejects_stale_admin_post(self): delivered(self) client, data = self.admin_form("close") def stale_post(): response = client.post(self.url, data) self.assertEqual(response.status_code, 200) self.assertContains(response, "changed; reload") self.race(lambda: close(self), stale_post, "success") self.assertEqual(ServiceCaseClosure.objects.count(), 1) def rollback_then_hold(self, model, action): with patch.object(model, "_persist", side_effect=ValidationError("Synthetic late failure")), self.assertRaises(ValidationError): action() ServiceCase.objects.select_for_update().get(pk=self.case.pk) def test_failed_handover_rolls_back_before_competing_retry(self): released(self) self.race(lambda: self.rollback_then_hold(ServiceCase, lambda: handover(self)), lambda: handover(self), "success") self.assertEqual(ServiceCaseHandover.objects.count(), 1) def test_failed_assignment_termination_rolls_back_before_retry(self): released(self) self.race(lambda: self.rollback_then_hold(ServiceEngineerAssignment, lambda: handover(self)), lambda: handover(self), "success") self.assertEqual(ServiceCaseHandover.objects.count(), 1) def test_failed_closure_rolls_back_before_competing_retry(self): delivered(self) self.race(lambda: self.rollback_then_hold(ServiceCase, lambda: close(self)), lambda: close(self), "success") self.assertEqual(ServiceCaseClosure.objects.count(), 1)