75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""PostgreSQL-only regression for simultaneous notice deletion/receipt submission."""
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from django.core.exceptions import ValidationError
|
|
from django.db import close_old_connections, connection
|
|
|
|
from apps.payments import services
|
|
from apps.payments.forms import ReceiptForm
|
|
from apps.payments.models import PaymentReceipt
|
|
from apps.payments.models import PaymentNotification
|
|
from apps.notifications.models import NotificationJob
|
|
from tests.test_payments import case, data, upload
|
|
|
|
pytestmark = [pytest.mark.django_db(transaction=True),
|
|
pytest.mark.skipif(connection.vendor != 'postgresql', reason='Requires PostgreSQL row locks')]
|
|
|
|
|
|
@pytest.mark.parametrize('same_key', [True, False])
|
|
def test_simultaneous_notice_sms_resend_creates_one_job(case, same_key):
|
|
original = PaymentNotification.objects.get(notice=case['notice'])
|
|
NotificationJob.objects.filter(pk=original.job_id).update(status='FAILED')
|
|
first_key = str(uuid4())
|
|
keys = [first_key, first_key if same_key else str(uuid4())]
|
|
barrier = Barrier(2)
|
|
def resend(key):
|
|
close_old_connections()
|
|
try:
|
|
barrier.wait(timeout=10)
|
|
try:
|
|
return services.resend_notice_sms(case['creator'], case['notice'], key).pk
|
|
except ValidationError:
|
|
return 'blocked'
|
|
finally:
|
|
connection.close()
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
outcomes = list(pool.map(resend, keys))
|
|
assert PaymentNotification.objects.filter(notice=case['notice']).count() == 2
|
|
assert len(set(outcomes)) == (1 if same_key else 2)
|
|
assert outcomes.count('blocked') == (0 if same_key else 1)
|
|
|
|
|
|
def test_simultaneous_delete_and_receipt_allow_exactly_one_operation(case):
|
|
form = ReceiptForm(data(case), {'receipt_image': upload()}, customer=case['customer'])
|
|
assert form.is_valid(), form.errors
|
|
barrier = Barrier(2)
|
|
|
|
def operate(operation):
|
|
close_old_connections()
|
|
try:
|
|
barrier.wait(timeout=10)
|
|
try:
|
|
if operation == 'delete':
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
else:
|
|
services.submit_receipt(case['customer'], case['customer'], form.cleaned_data)
|
|
return operation
|
|
except ValidationError:
|
|
return 'blocked'
|
|
finally:
|
|
connection.close()
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
outcomes = list(pool.map(operate, ['delete', 'receipt']))
|
|
case['notice'].refresh_from_db()
|
|
assert outcomes.count('blocked') == 1
|
|
if case['notice'].deleted_at:
|
|
assert not PaymentReceipt.objects.filter(notice=case['notice']).exists()
|
|
assert 'delete' in outcomes
|
|
else:
|
|
assert PaymentReceipt.objects.filter(notice=case['notice']).count() == 1
|
|
assert 'receipt' in outcomes
|