"""Separate-connection PostgreSQL review, replacement and lifecycle races.""" from django.contrib.auth import get_user_model from django.db import transaction from django.test import Client, TransactionTestCase from django.urls import reverse from apps.catalog import services as catalog_services from apps.catalog.models import ProductModel from apps.catalog.tests import make_catalog from apps.organization import test_assignment_concurrency as concurrency from . import services, queries from .evidence_admin import revision_token from .models import DevicePurchaseEvidence as Purchase, DeviceWarrantyCoverage as Coverage from .test_evidence import coverage from .tests import policy class EvidenceConcurrencyTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): _, _, self.model, _ = make_catalog("EVIDENCERACE") policy(self.model) self.device = services.register_device(product_model=self.model) self.user = get_user_model().objects.create_user(username="synthetic-reviewer") def evidence(self, **kwargs): return services.set_device_purchase_evidence(device=self.device, **kwargs) def test_simultaneous_first_evidence_writes_serialize_to_one_record(self): self.run_concurrent(lambda: self.evidence(seller_name="Synthetic Seller"), lambda: self.evidence(invoice_number="SYNTHETIC-INVOICE"), expected="success") record = Purchase.objects.get() self.assertEqual(record.seller_name, "Synthetic Seller") self.assertEqual(record.invoice_number, "SYNTHETIC-INVOICE") self.assertEqual(record.evidence_status, "UNVERIFIED") def test_verify_then_material_edit_resets_complete_review_metadata(self): evidence = self.evidence() self.run_concurrent(lambda: services.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user), lambda: self.evidence(invoice_number="SYNTHETIC-NEW"), expected="success") current = Purchase.objects.get() self.assertEqual(current.evidence_status, "UNVERIFIED") self.assertIsNone(current.reviewed_at) self.assertIsNone(current.reviewed_by_id) def test_material_edit_then_stale_verification_is_rejected(self): evidence = self.evidence() self.run_concurrent(lambda: self.evidence(invoice_number="SYNTHETIC-NEW"), lambda: services.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user), expected="validation") self.assertEqual(Purchase.objects.get().evidence_status, "UNVERIFIED") def test_verify_then_competing_stale_rejection_is_one_complete_decision(self): evidence = self.evidence() self.run_concurrent(lambda: services.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user, note="Accepted"), lambda: services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user, note="Rejected"), expected="validation") current = Purchase.objects.get() self.assertEqual((current.evidence_status, current.verification_note, current.reviewed_by_id), ("VERIFIED", "Accepted", self.user.pk)) self.assertIsNotNone(current.reviewed_at) def test_reviewer_deactivation_then_verification_rejected(self): evidence = self.evidence() self.run_concurrent(lambda: get_user_model().objects.filter(pk=self.user.pk).update(is_active=False), lambda: services.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user), expected="validation") self.assertEqual(Purchase.objects.get().evidence_status, "UNVERIFIED") def test_warranty_replacements_serialize_and_keep_history(self): self.run_concurrent(lambda: coverage(self.device, reference="SYNTHETIC-A"), lambda: coverage(self.device, reference="SYNTHETIC-B"), expected="success") self.assertEqual(Coverage.objects.count(), 2) self.assertEqual(Coverage.objects.get(is_active=True).reference, "SYNTHETIC-B") self.assertEqual(Coverage.objects.get(is_active=False).reference, "SYNTHETIC-A") def test_clear_then_replace_commits_new_active_coverage(self): coverage(self.device) self.run_concurrent(lambda: services.clear_device_warranty_coverage(device=self.device), lambda: coverage(self.device, reference="SYNTHETIC-NEW"), expected="success") self.assertEqual(Coverage.objects.get(is_active=True).reference, "SYNTHETIC-NEW") def test_replace_then_clear_leaves_only_history(self): coverage(self.device) self.run_concurrent(lambda: coverage(self.device), lambda: services.clear_device_warranty_coverage(device=self.device), expected="success") self.assertEqual(Coverage.objects.count(), 2) self.assertFalse(Coverage.objects.filter(is_active=True).exists()) def test_device_deactivation_then_evidence_edit_preserves_inactive_state(self): self.run_concurrent(lambda: services.deactivate_device(device=self.device), lambda: self.evidence(invoice_number="SYNTHETIC-HISTORY"), expected="success") self.device.refresh_from_db() self.assertFalse(self.device.is_active) self.assertEqual(Purchase.objects.get().invoice_number, "SYNTHETIC-HISTORY") def test_device_deactivation_then_coverage_replacement_rejects(self): current = coverage(self.device) self.run_concurrent(lambda: services.deactivate_device(device=self.device), lambda: coverage(self.device), expected="validation") self.assertEqual(queries.current_warranty_coverage(self.device).pk, current.pk) self.assertEqual(Coverage.objects.count(), 1) def test_coverage_then_device_deactivation_preserves_record_but_hides_operationally(self): self.run_concurrent(lambda: coverage(self.device), lambda: services.deactivate_device(device=self.device), expected="success") record = Coverage.objects.get() self.assertTrue(record.is_active) self.assertFalse(queries.device_has_recorded_warranty_coverage(device=self.device, on_date=record.coverage_start_date)) def test_evidence_edit_does_not_wait_for_unrelated_catalog_lock(self): def lock_model(): ProductModel.objects.select_for_update().get(pk=self.model.pk) self.run_concurrent(lock_model, lambda: self.evidence(invoice_number="SYNTHETIC"), expected="success", should_block=False) self.assertEqual(Purchase.objects.count(), 1) def test_purchase_coverage_then_material_edit_retires_coverage(self): evidence = self.evidence() services.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) self.run_concurrent(lambda: coverage(self.device, coverage_source="PURCHASE_EVIDENCE"), lambda: self.evidence(invoice_number="SYNTHETIC-CHANGED"), expected="success") self.assertEqual(Purchase.objects.get().evidence_status, "UNVERIFIED") self.assertFalse(Coverage.objects.get().is_active) def test_evidence_edit_then_purchase_coverage_creation_rejects(self): evidence = self.evidence() services.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) self.run_concurrent(lambda: self.evidence(invoice_number="SYNTHETIC-CHANGED"), lambda: coverage(self.device, coverage_source="PURCHASE_EVIDENCE"), expected="validation") self.assertFalse(Coverage.objects.exists()) def test_review_then_stale_admin_edit_waits_and_cannot_restore_old_facts(self): evidence = self.evidence(invoice_number="SYNTHETIC-ORIGINAL") user = get_user_model().objects.create_superuser(username="evidence-race-admin", password="test-only-password") def edit(): client = Client() client.force_login(user) response = client.post(reverse("admin:devices_devicepurchaseevidence_change", args=[evidence.pk]), { "purchase_date": "", "seller_name": "", "seller_reference": "", "invoice_number": "SYNTHETIC-STALE", "revision": revision_token(evidence), "_save": "Save", }) self.assertEqual(response.status_code, 302) self.run_concurrent(lambda: services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user), edit, expected="success") current = Purchase.objects.get() self.assertEqual(current.evidence_status, "REJECTED") self.assertEqual(current.invoice_number, "SYNTHETIC-ORIGINAL")