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.
528 lines
21 KiB
Python
528 lines
21 KiB
Python
import json
|
|
import uuid
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.test import Client
|
|
from django.utils import timezone
|
|
|
|
from apps.events.models import Event, Registration, RegistrationNotification
|
|
from apps.events.checkout import cancel_event_checkouts, expire_stale_registration_checkouts
|
|
from apps.payments.models import Payment
|
|
from apps.users.models import User
|
|
from core.authentication import create_jwt_token
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.integration
|
|
class TestRegistrationCheckout:
|
|
def setup_method(self):
|
|
self.client = Client()
|
|
suffix = uuid.uuid4().hex[:10]
|
|
self.user = User.objects.create_user(
|
|
username=f"checkout-{suffix}",
|
|
email=f"checkout-{suffix}@example.com",
|
|
mobile=f"0912{suffix[:7]}",
|
|
password="CheckoutPass!123",
|
|
)
|
|
self.token = create_jwt_token(self.user)
|
|
now = timezone.now()
|
|
self.event = Event.objects.create(
|
|
title="Checkout event",
|
|
description="Paid checkout behavior 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),
|
|
slug=f"checkout-{suffix}",
|
|
price=1250000,
|
|
capacity=1,
|
|
is_published=True,
|
|
)
|
|
|
|
def auth_headers(self):
|
|
return {"HTTP_AUTHORIZATION": f"Bearer {self.token}"}
|
|
|
|
def create_pending_checkout(self, *, expires_at=None):
|
|
now = timezone.now()
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.PENDING,
|
|
final_price=self.event.price,
|
|
checkout_started_at=now,
|
|
checkout_expires_at=expires_at or now + timedelta(minutes=15),
|
|
)
|
|
payment = Payment.objects.create(
|
|
user=self.user,
|
|
event=self.event,
|
|
registration=registration,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PENDING,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
idempotency_key=uuid.uuid4(),
|
|
)
|
|
return registration, payment
|
|
|
|
def test_active_checkout_reserves_capacity_but_can_be_resumed_by_its_owner(self):
|
|
registration, _ = self.create_pending_checkout()
|
|
|
|
assert Event.objects.get(pk=self.event.pk).current_attendees_count == 1
|
|
resume_response = self.client.post(
|
|
f"/api/events/{self.event.id}/register",
|
|
data=json.dumps({}),
|
|
content_type="application/json",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
other = User.objects.create_user(
|
|
username=f"other-{uuid.uuid4().hex[:8]}",
|
|
email=f"other-{uuid.uuid4().hex[:8]}@example.com",
|
|
mobile=f"0913{uuid.uuid4().int % 10**7:07d}",
|
|
password="CheckoutPass!123",
|
|
)
|
|
other_token = create_jwt_token(other)
|
|
other_response = self.client.post(
|
|
f"/api/events/{self.event.id}/register",
|
|
data=json.dumps({}),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION=f"Bearer {other_token}",
|
|
)
|
|
|
|
assert resume_response.status_code == 200
|
|
assert resume_response.json()["id"] == registration.id
|
|
assert other_response.status_code == 400
|
|
assert "ظرفیت" in other_response.json().get("detail", "")
|
|
|
|
def test_active_checkout_remains_resumable_after_registration_closes(self):
|
|
registration, _ = self.create_pending_checkout()
|
|
self.event.registration_end_date = timezone.now() - timedelta(minutes=1)
|
|
self.event.save(update_fields=["registration_end_date"])
|
|
|
|
response = self.client.post(
|
|
f"/api/events/{self.event.id}/register",
|
|
data=json.dumps({}),
|
|
content_type="application/json",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["id"] == registration.id
|
|
|
|
def test_registration_status_returns_resumable_payment_without_changing_legacy_flag(self):
|
|
registration, payment = self.create_pending_checkout()
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.id}/is-registered",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["is_registered"] is False
|
|
assert payload["checkout_state"] == "active"
|
|
assert payload["can_resume"] is True
|
|
assert payload["registration_id"] == registration.id
|
|
assert payload["payment_id"] == payment.id
|
|
expected_expiry = registration.checkout_expires_at.isoformat(timespec="milliseconds").replace("+00:00", "Z")
|
|
assert payload["checkout_expires_at"] == expected_expiry
|
|
|
|
def test_registration_hub_returns_ready_state_for_user_without_registration(self):
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["state"] == "ready"
|
|
assert payload["event"]["slug"] == self.event.slug
|
|
assert payload["registration"] is None
|
|
assert payload["payment"] is None
|
|
assert payload["can_register"] is True
|
|
|
|
def test_status_reads_do_not_cancel_an_expired_checkout_when_notifications_are_unavailable(self, mocker):
|
|
registration, _ = self.create_pending_checkout(expires_at=timezone.now() - timedelta(seconds=1))
|
|
mocker.patch(
|
|
"apps.events.tasks.send_registration_cancellation_notification.delay",
|
|
side_effect=RuntimeError("broker unavailable"),
|
|
)
|
|
|
|
hub = self.client.get(f"/api/events/{self.event.slug}/registration/status", **self.auth_headers())
|
|
legacy = self.client.get(f"/api/events/{self.event.id}/is-registered", **self.auth_headers())
|
|
|
|
registration.refresh_from_db()
|
|
assert hub.status_code == 200
|
|
assert hub.json()["state"] == "verification_pending"
|
|
assert legacy.status_code == 200
|
|
assert registration.status == Registration.StatusChoices.PENDING
|
|
assert not RegistrationNotification.objects.filter(registration=registration).exists()
|
|
|
|
def test_registration_hub_requires_authentication(self):
|
|
response = self.client.get(f"/api/events/{self.event.slug}/registration/status")
|
|
assert response.status_code == 401
|
|
|
|
def test_registration_hub_does_not_trust_a_status_query_parameter(self):
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.CONFIRMED,
|
|
final_price=self.event.price,
|
|
)
|
|
Payment.objects.create(
|
|
user=self.user,
|
|
event=self.event,
|
|
registration=registration,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PAID,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status?status=failed",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] == "paid_success"
|
|
|
|
def test_registration_hub_shows_confirmed_free_registration(self):
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.CONFIRMED,
|
|
final_price=0,
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
{"registration_id": str(registration.ticket_id)},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] == "free_success"
|
|
|
|
def test_registration_hub_returns_404_for_foreign_payment(self):
|
|
other = User.objects.create_user(
|
|
username=f"other-{uuid.uuid4().hex[:8]}",
|
|
email=f"other-{uuid.uuid4().hex[:8]}@example.com",
|
|
mobile=f"0913{uuid.uuid4().int % 10**7:07d}",
|
|
password="CheckoutPass!123",
|
|
)
|
|
payment = Payment.objects.create(
|
|
user=other,
|
|
event=self.event,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PENDING,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
{"payment_id": payment.id},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_registration_hub_uses_registration_ticket_uuid_and_exposes_active_payment(self):
|
|
registration, payment = self.create_pending_checkout()
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
{"registration_id": str(registration.ticket_id)},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["state"] == "active_checkout"
|
|
assert payload["registration"]["ticket_id"] == str(registration.ticket_id)
|
|
assert payload["payment"]["id"] == payment.id
|
|
assert payload["can_resume"] is True
|
|
assert payload["can_cancel"] is True
|
|
|
|
def test_registration_hub_hides_another_users_registration(self):
|
|
other = User.objects.create_user(
|
|
username=f"other-{uuid.uuid4().hex[:8]}",
|
|
email=f"other-{uuid.uuid4().hex[:8]}@example.com",
|
|
mobile=f"0913{uuid.uuid4().int % 10**7:07d}",
|
|
password="CheckoutPass!123",
|
|
)
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=other,
|
|
status=Registration.StatusChoices.CONFIRMED,
|
|
final_price=self.event.price,
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
{"registration_id": str(registration.ticket_id)},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_registration_hub_accepts_legacy_payment_without_linked_registration(self):
|
|
payment = Payment.objects.create(
|
|
user=self.user,
|
|
event=self.event,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PENDING,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
{"payment_id": payment.id},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["state"] == "verification_pending"
|
|
assert payload["registration"] is None
|
|
assert payload["payment"]["id"] == payment.id
|
|
assert payload["can_register"] is False
|
|
|
|
def test_registration_hub_uses_event_id_for_legacy_events_without_a_slug(self):
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.CONFIRMED,
|
|
final_price=self.event.price,
|
|
confirmation_notification_sent_at=timezone.now(),
|
|
)
|
|
payment = Payment.objects.create(
|
|
user=self.user,
|
|
event=self.event,
|
|
registration=registration,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PAID,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
)
|
|
Event.objects.filter(pk=self.event.pk).update(slug="")
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.pk}/registration/status",
|
|
{"payment_id": payment.id},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["state"] == "paid_success"
|
|
assert payload["event"]["slug"] == ""
|
|
assert payload["payment"]["id"] == payment.id
|
|
|
|
def test_registration_hub_prefers_active_attempt_over_newer_settled_attempt(self):
|
|
confirmed = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.CONFIRMED,
|
|
final_price=self.event.price,
|
|
)
|
|
Payment.objects.create(
|
|
user=self.user,
|
|
event=self.event,
|
|
registration=confirmed,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PAID,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
)
|
|
active, active_payment = self.create_pending_checkout()
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["state"] == "active_checkout"
|
|
assert payload["registration"]["ticket_id"] == str(active.ticket_id)
|
|
assert payload["payment"]["id"] == active_payment.id
|
|
|
|
def test_registration_hub_returns_manual_review_and_refund_state_from_saved_records(self):
|
|
registration, payment = self.create_pending_checkout()
|
|
payment.verification_manual_review_at = timezone.now()
|
|
payment.refund_status = Payment.RefundStatusChoices.MANUAL_REVIEW
|
|
payment.save(update_fields=["verification_manual_review_at", "refund_status"])
|
|
|
|
response = self.client.get(
|
|
f"/api/events/{self.event.slug}/registration/status",
|
|
{"payment_id": payment.id},
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["state"] == "refund_manual_review"
|
|
assert payload["registration"]["ticket_id"] == str(registration.ticket_id)
|
|
assert payload["payment"]["verification_state"] == "manual_review"
|
|
assert payload["payment"]["refund_status"] == "manual_review"
|
|
|
|
def test_expired_unresolved_checkout_releases_registration_and_blocks_duplicate_payment(self):
|
|
expired_at = timezone.now() - timedelta(seconds=1)
|
|
registration, payment = self.create_pending_checkout(expires_at=expired_at)
|
|
|
|
response = self.client.post(
|
|
"/api/payments/create",
|
|
data=json.dumps({
|
|
"event_id": self.event.id,
|
|
"description": "Retry after abandoned checkout",
|
|
"idempotency_key": str(uuid.uuid4()),
|
|
}),
|
|
content_type="application/json",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
registration.refresh_from_db()
|
|
payment.refresh_from_db()
|
|
assert response.status_code == 409
|
|
assert registration.status == Registration.StatusChoices.CANCELLED
|
|
assert registration.cancellation_reason == Registration.CancellationReasonChoices.CHECKOUT_EXPIRED
|
|
assert payment.status == Payment.OrderStatusChoices.PENDING
|
|
assert Payment.objects.filter(event=self.event, user=self.user).count() == 1
|
|
|
|
def test_payment_creation_rechecks_window_after_expiring_checkout_without_payment(self):
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.PENDING,
|
|
checkout_expires_at=timezone.now() - timedelta(seconds=1),
|
|
final_price=self.event.price,
|
|
)
|
|
self.event.registration_end_date = timezone.now() - timedelta(minutes=1)
|
|
self.event.save(update_fields=["registration_end_date"])
|
|
|
|
response = self.client.post(
|
|
"/api/payments/create",
|
|
data=json.dumps({"event_id": self.event.id, "description": "Expired checkout",
|
|
"idempotency_key": str(uuid.uuid4())}),
|
|
content_type="application/json",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
registration.refresh_from_db()
|
|
assert response.status_code == 400
|
|
assert registration.status == Registration.StatusChoices.CANCELLED
|
|
assert not Payment.objects.filter(registration=registration).exists()
|
|
|
|
def test_expiry_sweep_cancels_only_new_expired_checkouts_and_releases_the_seat(self):
|
|
expired_at = timezone.now() - timedelta(seconds=1)
|
|
registration, payment = self.create_pending_checkout(expires_at=expired_at)
|
|
legacy = Registration.objects.create(
|
|
event=self.event,
|
|
user=User.objects.create_user(
|
|
username=f"legacy-{uuid.uuid4().hex[:8]}",
|
|
email=f"legacy-{uuid.uuid4().hex[:8]}@example.com",
|
|
mobile=f"0914{uuid.uuid4().int % 10**7:07d}",
|
|
password="CheckoutPass!123",
|
|
),
|
|
status=Registration.StatusChoices.PENDING,
|
|
final_price=self.event.price,
|
|
)
|
|
|
|
count = expire_stale_registration_checkouts()
|
|
|
|
registration.refresh_from_db()
|
|
payment.refresh_from_db()
|
|
legacy.refresh_from_db()
|
|
assert count == 1
|
|
assert registration.status == Registration.StatusChoices.CANCELLED
|
|
assert registration.cancellation_reason == Registration.CancellationReasonChoices.CHECKOUT_EXPIRED
|
|
assert payment.status == Payment.OrderStatusChoices.PENDING
|
|
assert Event.objects.get(pk=self.event.pk).current_attendees_count == 0
|
|
assert legacy.status == Registration.StatusChoices.PENDING
|
|
|
|
def test_user_can_cancel_active_checkout_without_deleting_registration(self, mocker):
|
|
registration, payment = self.create_pending_checkout()
|
|
mocker.patch("apps.payments.tasks.reconcile_payment.apply_async")
|
|
|
|
response = self.client.post(
|
|
f"/api/events/{self.event.id}/checkout/cancel",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
registration.refresh_from_db()
|
|
payment.refresh_from_db()
|
|
assert response.status_code == 200
|
|
assert registration.status == Registration.StatusChoices.CANCELLED
|
|
assert registration.cancellation_reason == Registration.CancellationReasonChoices.USER_CANCELLED
|
|
assert Registration.objects.filter(pk=registration.pk).exists()
|
|
assert payment.status == Payment.OrderStatusChoices.PENDING
|
|
assert Event.objects.get(pk=self.event.pk).current_attendees_count == 0
|
|
|
|
def test_new_checkout_is_blocked_while_late_payment_refund_is_unresolved(self):
|
|
registration = Registration.objects.create(
|
|
event=self.event,
|
|
user=self.user,
|
|
status=Registration.StatusChoices.CANCELLED,
|
|
cancellation_reason=Registration.CancellationReasonChoices.CHECKOUT_EXPIRED,
|
|
checkout_started_at=timezone.now() - timedelta(minutes=20),
|
|
checkout_expires_at=timezone.now() - timedelta(minutes=5),
|
|
final_price=self.event.price,
|
|
)
|
|
Payment.objects.create(
|
|
user=self.user,
|
|
event=self.event,
|
|
registration=registration,
|
|
base_amount=self.event.price,
|
|
amount=self.event.price,
|
|
status=Payment.OrderStatusChoices.PAID,
|
|
refund_status=Payment.RefundStatusChoices.PENDING,
|
|
authority=f"AUTH-{uuid.uuid4().hex}",
|
|
)
|
|
|
|
register_response = self.client.post(
|
|
f"/api/events/{self.event.id}/register",
|
|
data=json.dumps({}),
|
|
content_type="application/json",
|
|
**self.auth_headers(),
|
|
)
|
|
payment_response = self.client.post(
|
|
"/api/payments/create",
|
|
data=json.dumps({
|
|
"event_id": self.event.id,
|
|
"description": "Do not charge twice while refund is unresolved",
|
|
"idempotency_key": str(uuid.uuid4()),
|
|
}),
|
|
content_type="application/json",
|
|
**self.auth_headers(),
|
|
)
|
|
|
|
assert register_response.status_code == 409
|
|
assert payment_response.status_code == 409
|
|
assert Registration.objects.filter(event=self.event, user=self.user).count() == 1
|
|
|
|
def test_event_cancellation_ends_checkouts_and_recovers_payment_reconciliation(self, mocker):
|
|
registration, payment = self.create_pending_checkout()
|
|
from apps.payments.tasks import sweep_stale_payment_verifications
|
|
|
|
enqueue = mocker.patch("apps.payments.tasks.reconcile_payment.apply_async")
|
|
self.event.is_cancelled = True
|
|
self.event.save(update_fields=["is_cancelled"])
|
|
|
|
cancelled = cancel_event_checkouts(self.event.id)
|
|
|
|
registration.refresh_from_db()
|
|
assert cancelled == 1
|
|
assert registration.status == Registration.StatusChoices.CANCELLED
|
|
assert registration.cancellation_reason == Registration.CancellationReasonChoices.EVENT_CANCELLED
|
|
assert Event.objects.get(pk=self.event.pk).current_attendees_count == 0
|
|
assert RegistrationNotification.objects.filter(
|
|
registration=registration,
|
|
kind=RegistrationNotification.Kind.CANCELLED,
|
|
).exists()
|
|
assert sweep_stale_payment_verifications() == 1
|
|
enqueue.assert_called_once_with(args=(payment.id,))
|