"""Real PostgreSQL waits observed through the established blocking helper.""" from django.contrib.auth import get_user_model from django.test import Client, TransactionTestCase from django.urls import reverse from apps.catalog import services as catalog_services from apps.catalog.models import ProductCategory from apps.catalog.tests import make_catalog from apps.organization import test_assignment_concurrency as concurrency from . import queries, services from .models import Device, DeviceIdentifier from .tests import policy, synthetic_imei class DeviceConcurrencyTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): self.brand, self.category, self.model, self.variant = make_catalog("DEVICERACE") policy(self.model) def register(self, seed=1, **kwargs): return services.register_device(product_model=self.model, imei1=synthetic_imei(seed), **kwargs) def test_same_imei_registration_has_one_winner_and_no_orphan_device(self): self.run_concurrent(self.register, self.register, expected="validation") self.assertEqual(Device.objects.count(), 1) self.assertEqual(DeviceIdentifier.objects.count(), 1) # Subsequent writes still work after the rejected transaction. self.register(2) self.assertEqual(Device.objects.count(), 2) def test_cross_model_cross_slot_imei_registration_has_one_winner(self): _, _, other_model, _ = make_catalog("OTHER") policy(other_model) self.run_concurrent(self.register, lambda: services.register_device(product_model=other_model, imei2=synthetic_imei()), expected="validation") self.assertEqual(Device.objects.count(), 1) self.assertEqual(DeviceIdentifier.objects.count(), 1) def test_same_serial_registration_has_one_winner(self): def register(): services.register_device(product_model=self.model, serial="SYNTHETIC-RACE") self.run_concurrent(register, register, expected="validation") self.assertEqual(Device.objects.count(), 1) def test_distinct_registrations_share_catalog_without_blocking(self): self.run_concurrent(lambda: self.register(1), lambda: self.register(2), expected="success", should_block=False) self.assertEqual(Device.objects.count(), 2) def test_two_corrections_serialize_and_preserve_complete_history(self): device = self.register() self.run_concurrent( lambda: services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)), lambda: services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(3)), expected="success") self.assertEqual(device.identifiers.count(), 3) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei(3)) for seed in (1, 2, 3): self.assertEqual(queries.find_device_by_identifier(synthetic_imei(seed)), device) def test_corrections_of_different_devices_cannot_claim_same_new_identifier(self): first, second = self.register(1), self.register(2) self.run_concurrent( lambda: services.replace_device_identifier(device=first, identifier_type="IMEI1", new_value=synthetic_imei(3)), lambda: services.replace_device_identifier(device=second, identifier_type="IMEI1", new_value=synthetic_imei(3)), expected="validation") self.assertEqual(second.identifiers.get(is_active=True).normalized_value, synthetic_imei(2)) self.assertEqual(DeviceIdentifier.objects.count(), 3) def test_deactivation_then_correction_preserves_inactive_device(self): device = self.register() self.run_concurrent(lambda: services.deactivate_device(device=device), lambda: services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)), expected="success") device.refresh_from_db() self.assertFalse(device.is_active) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei(2)) def test_correction_then_deactivation_preserves_corrected_identifier(self): device = self.register() self.run_concurrent( lambda: services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)), lambda: services.deactivate_device(device=device), expected="success") device.refresh_from_db() self.assertFalse(device.is_active) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei(2)) def test_brand_deactivation_blocks_then_rejects_registration(self): self.run_concurrent(lambda: catalog_services.deactivate_brand(brand=self.brand), self.register, expected="validation") self.assertFalse(Device.objects.exists()) def test_category_deactivation_blocks_then_rejects_registration(self): self.run_concurrent(lambda: catalog_services.deactivate_category(category=self.category), self.register, expected="validation") self.assertFalse(Device.objects.exists()) def test_model_deactivation_blocks_then_rejects_registration(self): self.run_concurrent(lambda: catalog_services.deactivate_product_model(product_model=self.model), self.register, expected="validation") self.assertFalse(Device.objects.exists()) def test_variant_deactivation_blocks_then_rejects_registration(self): self.run_concurrent(lambda: catalog_services.deactivate_variant(variant=self.variant), lambda: self.register(product_variant=self.variant), expected="validation") self.assertFalse(Device.objects.exists()) def test_registration_then_model_deactivation_preserves_device_but_hides_operationally(self): self.run_concurrent(self.register, lambda: catalog_services.deactivate_product_model(product_model=self.model), expected="success") self.assertTrue(Device.objects.get().is_active) self.assertFalse(queries.active_devices_for_product_model(self.model).exists()) def test_model_deactivation_blocks_then_rejects_device_reactivation(self): device = services.deactivate_device(device=self.register()) self.run_concurrent(lambda: catalog_services.deactivate_product_model(product_model=self.model), lambda: services.reactivate_device(device=device), expected="validation") device.refresh_from_db() self.assertFalse(device.is_active) def test_policy_update_before_registration_uses_new_requirements(self): self.run_concurrent(lambda: policy(self.model, serial="REQUIRED"), self.register, expected="validation") self.assertFalse(Device.objects.exists()) def test_registration_before_policy_update_preserves_device_and_identifiers(self): self.run_concurrent(self.register, lambda: policy(self.model, serial="REQUIRED"), expected="success") self.assertEqual(Device.objects.count(), 1) self.assertEqual(DeviceIdentifier.objects.count(), 1) def test_category_correction_during_lock_acquisition_fails_safely_without_new_ancestor_lock(self): destination = ProductCategory.objects.create(code="DEST", name="Synthetic destination") def move(): self.model.category = destination self.model.save(update_fields=["category"]) self.run_concurrent(move, self.register, expected="validation") self.assertFalse(Device.objects.exists()) self.register() self.assertEqual(Device.objects.count(), 1) def test_stale_admin_edit_after_correction_and_deactivation_does_not_revert_state(self): device = self.register() user = get_user_model().objects.create_superuser(username="device-race-admin", password="test-only-password") def first(): services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) services.deactivate_device(device=device) def edit(): client = Client() client.force_login(user) response = client.post(reverse("admin:devices_device_change", args=[device.pk]), {"imei1": synthetic_imei(), "is_active": "on", "_save": "Save"}) self.assertEqual(response.status_code, 302) # A no-op edit need not lock or wait; it must never write its stale snapshot. self.run_concurrent(first, edit, expected="success", should_block=False) device.refresh_from_db() self.assertFalse(device.is_active) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei(2))