79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
"""PostgreSQL contention checks for the checkout expiry worker and status API."""
|
|
|
|
import threading
|
|
import time
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.db import connection, connections, transaction
|
|
from django.test import Client
|
|
from django.utils import timezone
|
|
|
|
from apps.events.checkout import expire_stale_registration_checkouts
|
|
from apps.events.models import Event, Registration
|
|
from apps.users.models import User
|
|
from core.authentication import create_jwt_token
|
|
|
|
|
|
@pytest.mark.django_db(transaction=True)
|
|
@pytest.mark.integration
|
|
def test_status_reads_and_expiry_worker_progress_around_a_locked_checkout():
|
|
if connection.vendor != "postgresql":
|
|
pytest.skip("PostgreSQL row locks are required")
|
|
|
|
now = timezone.now()
|
|
event = Event.objects.create(
|
|
title="Checkout lock test", slug="checkout-lock-test", description="Lock test",
|
|
start_time=now + timedelta(days=2), end_time=now + timedelta(days=2, hours=1),
|
|
registration_start_date=now - timedelta(days=1),
|
|
registration_end_date=now + timedelta(days=1),
|
|
status=Event.StatusChoices.PUBLISHED, price=1000, capacity=2,
|
|
)
|
|
user = User.objects.create_user(username="checkout-lock-owner", mobile="09191112233", password="TestPass123!")
|
|
other = User.objects.create_user(username="checkout-lock-other", mobile="09191112234", password="TestPass123!")
|
|
locked = Registration.objects.create(
|
|
event=event, user=user, status=Registration.StatusChoices.PENDING,
|
|
checkout_expires_at=now - timedelta(minutes=2),
|
|
)
|
|
free = Registration.objects.create(
|
|
event=event, user=other, status=Registration.StatusChoices.PENDING,
|
|
checkout_expires_at=now - timedelta(minutes=1),
|
|
)
|
|
acquired = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def hold_row_lock():
|
|
try:
|
|
with transaction.atomic():
|
|
Registration.objects.select_for_update().get(pk=locked.pk)
|
|
acquired.set()
|
|
release.wait(10)
|
|
finally:
|
|
connections.close_all()
|
|
|
|
holder = threading.Thread(target=hold_row_lock)
|
|
holder.start()
|
|
try:
|
|
assert acquired.wait(5)
|
|
start = time.monotonic()
|
|
response = Client().get(
|
|
f"/api/events/{event.slug}/registration/status",
|
|
HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(user)}",
|
|
)
|
|
assert response.status_code == 200
|
|
assert time.monotonic() - start < 2
|
|
|
|
assert expire_stale_registration_checkouts(event_id=event.id) == 1
|
|
locked.refresh_from_db()
|
|
free.refresh_from_db()
|
|
assert locked.status == Registration.StatusChoices.PENDING
|
|
assert free.status == Registration.StatusChoices.CANCELLED
|
|
finally:
|
|
release.set()
|
|
holder.join(timeout=5)
|
|
|
|
assert not holder.is_alive()
|
|
assert expire_stale_registration_checkouts(event_id=event.id) == 1
|
|
locked.refresh_from_db()
|
|
assert locked.status == Registration.StatusChoices.CANCELLED
|