Amirhossein Khalili
6c5fadef2f
Replace persisted event statuses with publication and cancellation flags, and compute public lifecycle labels from event timing and registration capacity. Keep filters, registration checks, and profile responses aligned with the derived status.
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),
|
|
is_published=True, 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
|