from django.contrib.auth.models import Permission from django.core.exceptions import PermissionDenied, ValidationError from django.db import transaction from django.test import TransactionTestCase, Client from django.urls import reverse from apps.organization import test_assignment_concurrency as concurrency from apps.service import test_handover as handover from apps.service.models import ServiceCase from apps.inventory.tests import grant from apps.access.services import set_role_permissions from .test_payment import PaymentFixture, setup_payment from . import payment_services as s, payment_queries as q from .models import ServicePayment, PaymentAllocation, ServicePaymentReceipt, PaymentReversal, ReceiptSequence, ServiceFinancialRelease class PaymentConcurrencyTests(PaymentFixture, TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): setup_payment(self) def other_invoice(self, *, center=None): from apps.service.tests import intake from apps.service.test_engineer_assignment import assign from apps.service import test_diagnosis as diagnosis, test_repair as repair from apps.organization.assignment_services import create_assignment from apps.access.services import create_role_assignment saved = {key:getattr(self, key) for key in ("case", "invoice", "execution", "action", "quote")} if center is not None: path = create_assignment(user=self.engineer, company=self.company, region=center.region, service_center=center) create_role_assignment(user=self.engineer, role=self.role, organization_assignment=path) self.case = intake(self, service_center=center or self.center) assign(self) assessment = diagnosis.begin(self) diagnosis.add(self, assessment) diagnosis.complete(self, assessment) self.execution, self.action = repair.prepared(self, performed=False) self.quote = self.decide(self.submit(self.lines(self.create()))) repair.perform(self, self.action) row = self.prepare() row = self.reconcile(row, confirmations=[dict(quotation_line=self.quote.lines.get(), quantity=1, reason="Confirmed synthetic labor")]) self.ready() row = self.finalize(row) for key, value in saved.items(): setattr(self, key, value) return row def test_two_payments_competing_for_remaining_balance(self): token = self.summary()["revision"] self.run_concurrent(lambda:self.pay(amount="70", expected_revision=token), lambda:self.pay(amount="70", expected_revision=token), expected="validation") self.assertEqual((ServicePayment.objects.count(), PaymentAllocation.objects.count(), self.summary()["balance"]), (1, 1, 30)) def test_two_full_allocations_cannot_overpay(self): token = self.summary()["revision"] self.run_concurrent(lambda:self.pay(expected_revision=token), lambda:self.pay(expected_revision=token), expected="validation") self.assertEqual(self.summary()["paid"], 100) self.assertEqual(PaymentAllocation.objects.count(), 1) def test_same_center_receipt_numbers(self): other = self.other_invoice() token = q.invoice_settlement_summary(actor=self.actor, invoice=other)["revision"] self.run_concurrent(self.pay, lambda:self.pay(invoice=other, expected_revision=token), expected="success") self.assertEqual(set(ServicePaymentReceipt.objects.values_list("number", flat=True)), {"RCT-00000001", "RCT-00000002"}) def test_different_center_receipt_numbers_do_not_block(self): other = self.other_invoice(center=self.center2) token = q.invoice_settlement_summary(actor=self.actor, invoice=other)["revision"] self.run_concurrent(self.pay, lambda:self.pay(invoice=other, expected_revision=token), expected="success", should_block=False) self.assertEqual(list(ServicePaymentReceipt.objects.values_list("number", flat=True)), ["RCT-00000001", "RCT-00000001"]) self.assertEqual(ReceiptSequence.objects.count(), 2) def test_posting_before_stale_reversal(self): payment = self.pay(amount="30") token = self.summary()["revision"] self.run_concurrent(lambda:self.pay(amount="20", expected_revision=token), lambda:self.reverse_payment(payment, expected_revision=token), expected="validation") self.assertEqual(self.summary()["paid"], 50) self.assertFalse(PaymentReversal.objects.exists()) def test_reversal_before_stale_posting(self): payment = self.pay(amount="30") token = self.summary()["revision"] self.run_concurrent(lambda:self.reverse_payment(payment, expected_revision=token), lambda:self.pay(amount="20", expected_revision=token), expected="validation") self.assertEqual(self.summary()["paid"], 0) self.assertEqual(ServicePayment.objects.count(), 1) def test_final_payment_before_delivery(self): handover.released(self) self.run_concurrent(self.pay, lambda:handover.handover(self), expected="success") self.case.refresh_from_db() self.assertEqual((self.case.status, self.summary()["state"]), ("DELIVERED", "PAID")) def test_unpaid_delivery_attempt_before_final_payment(self): handover.released(self) def rejected(): with self.assertRaises(ValidationError): handover.handover(self) ServiceCase.objects.select_for_update().get(pk=self.case.pk) self.run_concurrent(rejected, self.pay, expected="success") self.case.refresh_from_db() self.assertEqual((self.case.status, self.summary()["state"]), ("READY_FOR_DELIVERY", "PAID")) def test_final_payment_reversal_before_delivery(self): payment = self.pay() handover.released(self) self.run_concurrent(lambda:self.reverse_payment(payment), lambda:handover.handover(self), expected="validation") self.case.refresh_from_db() self.assertEqual((self.case.status, self.summary()["state"]), ("READY_FOR_DELIVERY", "UNPAID")) def test_delivery_before_unsecured_reversal(self): payment = self.pay() handover.released(self) self.run_concurrent(lambda:handover.handover(self), lambda:self.reverse_payment(payment), expected="validation") self.case.refresh_from_db() self.assertEqual((self.case.status, self.summary()["state"]), ("DELIVERED", "PAID")) self.assertFalse(PaymentReversal.objects.exists()) def test_due_release_before_delivery(self): handover.released(self) self.run_concurrent(self.release_due, lambda:handover.handover(self), expected="success") self.case.refresh_from_db() self.assertEqual((self.case.status, self.summary()["balance"]), ("DELIVERED", 100)) def test_release_limit_does_not_cover_racing_reversal(self): payment = self.pay(amount="60") token = self.summary()["revision"] self.run_concurrent(lambda:self.release_due(expected_revision=token), lambda:self.reverse_payment(payment, expected_revision=token), expected="validation") self.reverse_payment(payment) self.assertFalse(self.summary()["financially_clear"]) self.assertEqual(ServiceFinancialRelease.objects.get().outstanding_amount, 40) def test_two_reversals_preserve_one_evidence_record(self): payment = self.pay() token = self.summary()["revision"] self.run_concurrent(lambda:self.reverse_payment(payment, expected_revision=token), lambda:self.reverse_payment(payment, expected_revision=token), expected="validation") self.assertEqual(PaymentReversal.objects.count(), 1) self.assertEqual(self.summary()["balance"], 100) def test_duplicate_reference_across_invoices_serializes(self): other = self.other_invoice() token = q.invoice_settlement_summary(actor=self.actor, invoice=other)["revision"] self.run_concurrent(lambda:self.pay(method="CARD", reference="SYNTHETIC-DUPLICATE"), lambda:self.pay(invoice=other, expected_revision=token, method="CARD", reference="SYNTHETIC-DUPLICATE"), expected="validation") self.assertEqual(ServicePayment.objects.count(), 1) self.assertEqual(q.invoice_settlement_summary(actor=self.actor, invoice=other)["balance"], 100) def test_receipt_rollback_racing_payment(self): def rollback(): with self.assertRaises(RuntimeError), transaction.atomic(): self.pay() raise RuntimeError("Synthetic abort after receipt") ServiceCase.objects.select_for_update().get(pk=self.case.pk) self.run_concurrent(rollback, self.pay, expected="success") self.assertEqual(ServicePaymentReceipt.objects.get().number, "RCT-00000001") self.assertEqual(ReceiptSequence.objects.get().next_value, 2) def test_release_rollback_cannot_clear_delivery(self): handover.released(self) def rollback(): with self.assertRaises(RuntimeError), transaction.atomic(): self.release_due() raise RuntimeError("Synthetic release abort") ServiceCase.objects.select_for_update().get(pk=self.case.pk) self.run_concurrent(rollback, lambda:handover.handover(self), expected="validation") self.assertFalse(ServiceFinancialRelease.objects.exists()) self.case.refresh_from_db() self.assertEqual(self.case.status, "READY_FOR_DELIVERY") def test_permission_revocation_before_receiving(self): role = grant(self.user, self.company, center=self.center) set_role_permissions(role=role, permissions=Permission.objects.filter(codename="receive_servicepayment")) def rejected(): with self.assertRaises(PermissionDenied): self.pay(actor=self.user) self.run_concurrent(lambda:set_role_permissions(role=role, permissions=[]), rejected, expected="success") self.assertFalse(ServicePayment.objects.exists()) def test_permission_revocation_before_due_release(self): role = grant(self.user, self.company) set_role_permissions(role=role, permissions=Permission.objects.filter(codename="authorize_due_release")) def rejected(): with self.assertRaises(PermissionDenied): self.release_due(actor=self.user) self.run_concurrent(lambda:set_role_permissions(role=role, permissions=[]), rejected, expected="success") self.assertFalse(ServiceFinancialRelease.objects.exists()) def test_stale_admin_payment_after_competing_payment(self): client = Client() client.force_login(self.actor) url = reverse("admin:commercial_invoice_settlement", args=[self.invoice.pk]) token = client.get(url).context_data["revision_token"] def post(): self.assertEqual(client.post(url, dict(operation="receive", amount="20", method="CASH", revision_token=token)).status_code, 400) self.run_concurrent(lambda:self.pay(amount="20"), post, expected="success") self.assertEqual(self.summary()["paid"], 20)