"""Real PostgreSQL diagnostic serialization, including taxonomy and rollback.""" from django.core.exceptions import ValidationError from django.db import transaction from django.test import TransactionTestCase from apps.access import services as access from apps.catalog import services as catalog from apps.devices import services as devices from apps.organization import test_assignment_concurrency as concurrency from apps.service_catalog import services as taxonomy from . import diagnostic_services as services, diagnostic_queries as queries from .models import ServiceCase, ServiceDiagnosticFinding from .services import cancel_service_case from .test_diagnosis import setup_diagnosis, begin, add, complete, abandon, assert_diagnostic_invariants from .test_engineer_assignment import reassign, unassign class DiagnosisConcurrencyTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): setup_diagnosis(self) def race(self, first, second, expected="validation"): self.run_concurrent(first, second, expected=expected) assert_diagnostic_invariants(self) def prepared(self, **kwargs): assessment = begin(self) finding = add(self, assessment, **kwargs) return assessment, finding def cancel(self, **kwargs): return cancel_service_case(service_case=self.case, cancelled_by=self.user, **kwargs) def test_begin_vs_begin(self): self.race(lambda: begin(self), lambda: begin(self)) def test_begin_then_reassign(self): self.race(lambda: begin(self), lambda: reassign(self)) def test_reassign_then_begin_rejects_old_engineer(self): self.race(lambda: reassign(self), lambda: begin(self)) def test_begin_then_unassign(self): self.race(lambda: begin(self), lambda: unassign(self)) def test_unassign_then_begin(self): self.race(lambda: unassign(self), lambda: begin(self)) def test_begin_then_cancel(self): self.race(lambda: begin(self), self.cancel, "success") def test_cancel_then_begin(self): self.race(self.cancel, lambda: begin(self)) def test_add_then_complete_rejects_changed_finding_set(self): assessment, _ = self.prepared() self.race(lambda: add(self, assessment, fault_diagnosis=self.fault2), lambda: complete(self, assessment)) def test_complete_then_add(self): assessment, _ = self.prepared() self.race(lambda: complete(self, assessment), lambda: add(self, assessment, fault_diagnosis=self.fault2)) def test_update_then_complete_rejects_changed_finding_set(self): assessment, finding = self.prepared() self.race(lambda: services.update_diagnostic_finding(finding=finding, actor=self.engineer, note="Synthetic update"), lambda: complete(self, assessment)) def test_complete_then_update(self): assessment, finding = self.prepared() self.race(lambda: complete(self, assessment), lambda: services.update_diagnostic_finding(finding=finding, actor=self.engineer, note="Synthetic update")) def test_remove_then_complete(self): assessment, finding = self.prepared() self.race(lambda: services.remove_diagnostic_finding(finding=finding, actor=self.engineer), lambda: complete(self, assessment)) def test_complete_then_remove(self): assessment, finding = self.prepared() self.race(lambda: complete(self, assessment), lambda: services.remove_diagnostic_finding(finding=finding, actor=self.engineer)) def test_complete_vs_complete(self): assessment, _ = self.prepared() self.race(lambda: complete(self, assessment), lambda: complete(self, assessment)) def test_complete_then_cancel(self): assessment, _ = self.prepared() self.race(lambda: complete(self, assessment), self.cancel, "success") def test_cancel_then_complete(self): assessment, _ = self.prepared() self.race(self.cancel, lambda: complete(self, assessment)) def test_abandon_then_complete(self): assessment, _ = self.prepared() self.race(lambda: abandon(self, assessment), lambda: complete(self, assessment)) def test_complete_then_abandon(self): assessment, _ = self.prepared() self.race(lambda: complete(self, assessment), lambda: abandon(self, assessment)) def test_abandon_then_cancel(self): assessment, _ = self.prepared() self.race(lambda: abandon(self, assessment), self.cancel, "success") def test_cancel_then_abandon(self): assessment, _ = self.prepared() self.race(self.cancel, lambda: abandon(self, assessment)) def test_fault_deactivation_then_add(self): assessment = begin(self) self.race(lambda: taxonomy.deactivate_fault_diagnosis(diagnosis=self.fault), lambda: add(self, assessment)) def test_root_deactivation_then_add(self): assessment = begin(self) self.race(lambda: taxonomy.deactivate_root_cause(root_cause=self.root), lambda: add(self, assessment, root_cause=self.root)) def test_fault_deactivation_then_complete(self): assessment, _ = self.prepared() self.race(lambda: taxonomy.deactivate_fault_diagnosis(diagnosis=self.fault), lambda: complete(self, assessment)) def test_root_deactivation_then_complete(self): assessment, _ = self.prepared(root_cause=self.root) self.race(lambda: taxonomy.deactivate_root_cause(root_cause=self.root), lambda: complete(self, assessment)) def test_fault_applicability_change_then_add(self): assessment = begin(self) self.race(lambda: taxonomy.set_fault_diagnosis_applicability(diagnosis=self.fault, applies_to_all_product_categories=False, product_categories=[]), lambda: add(self, assessment)) def test_root_applicability_change_then_complete(self): assessment, _ = self.prepared(root_cause=self.root) self.race(lambda: taxonomy.set_root_cause_applicability(root_cause=self.root, applies_to_all_product_categories=False, product_categories=[]), lambda: complete(self, assessment)) def test_complete_then_applicability_replacement_preserves_history(self): assessment, _ = self.prepared(root_cause=self.root) self.race(lambda: complete(self, assessment), lambda: taxonomy.set_root_cause_applicability(root_cause=self.root, applies_to_all_product_categories=False, product_categories=[self.category]), "success") def test_eligibility_revocation_then_begin(self): self.race(lambda: access.deactivate_role_assignment(assignment=self.role_assignment), lambda: begin(self)) def test_permission_revocation_then_complete(self): assessment, _ = self.prepared() self.race(lambda: access.set_role_permissions(role=self.role, permissions=[]), lambda: complete(self, assessment)) def test_device_deactivation_then_begin(self): self.race(lambda: devices.deactivate_device(device=self.device), lambda: begin(self)) def test_category_deactivation_then_complete(self): assessment, _ = self.prepared() self.race(lambda: catalog.deactivate_category(category=self.category), lambda: complete(self, assessment)) def test_competing_stale_finding_edit(self): _, finding = self.prepared() self.race(lambda: services.update_diagnostic_finding(finding=finding, actor=self.engineer, note="first"), lambda: services.update_diagnostic_finding(finding=finding, actor=self.engineer, note="stale", expected_updated_at=finding.updated_at)) def test_stale_assessment_completion(self): assessment, _ = self.prepared() assessment.refresh_from_db() self.race(lambda: services.update_diagnostic_assessment(assessment=assessment, actor=self.engineer, technical_note="newer"), lambda: complete(self, assessment, expected_updated_at=assessment.updated_at)) def test_aggregate_revision_change_then_finding_edit(self): assessment, finding = self.prepared() assessment.refresh_from_db() self.race(lambda: services.update_diagnostic_assessment(assessment=assessment, actor=self.engineer, technical_note="newer"), lambda: services.update_diagnostic_finding(finding=finding, actor=self.engineer, note="stale", expected_updated_at=finding.updated_at, expected_assessment_updated_at=assessment.updated_at)) def test_duplicate_null_findings_serialize(self): assessment = begin(self) self.race(lambda: add(self, assessment), lambda: add(self, assessment)) self.assertEqual(ServiceDiagnosticFinding.objects.count(), 1) def test_rollback_after_assessment_creation_then_begin(self): def failed_begin(): with self.assertRaises(ValidationError): with transaction.atomic(): begin(self) raise ValidationError("Synthetic outer transaction failure") ServiceCase.objects.select_for_update().get(pk=self.case.pk) self.race(failed_begin, lambda: begin(self), "success") def test_completion_failure_then_valid_completion(self): assessment, _ = self.prepared() def failed_complete(): with self.assertRaises(ValidationError): complete(self, assessment, technical_note="x" * 4001) ServiceCase.objects.select_for_update().get(pk=self.case.pk) self.race(failed_complete, lambda: complete(self, assessment), "success") def test_cancellation_failure_preserves_diagnosis_then_completion(self): assessment, _ = self.prepared() def failed_cancel(): # Assessment abandonment is written before assignment reason validation. with self.assertRaises(ValidationError): self.cancel(reason="x" * 2001) self.assertIsNotNone(queries.current_diagnostic_assessment(self.case)) ServiceCase.objects.select_for_update().get(pk=self.case.pk) self.race(failed_cancel, lambda: complete(self, assessment), "success")