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.
60 lines
3.1 KiB
Python
60 lines
3.1 KiB
Python
import pytest
|
|
from types import SimpleNamespace
|
|
from datetime import timedelta
|
|
from django.contrib import admin
|
|
from django.utils import timezone
|
|
from apps.events.admin import EventAdmin, RegistrationAdmin
|
|
from apps.events.models import Event, Registration, RegistrationNotification
|
|
from apps.events.tasks import send_registration_cancellation_notification, send_registration_confirmation_notification
|
|
from apps.events.tasks import deliver_pending_registration_notifications
|
|
from apps.users.models import User
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.integration
|
|
class TestEventModelTests:
|
|
|
|
def setup_method(self):
|
|
now = timezone.now()
|
|
self.event = Event.objects.create(title='Workshop', description='**Important** details', start_time=now + timedelta(days=1), end_time=now + timedelta(days=1, hours=2), capacity=1, registration_start_date=now - timedelta(days=1), registration_end_date=now + timedelta(hours=1), is_published=True)
|
|
self.user = User.objects.create_user(username='attendee', mobile='09191110003', password='password')
|
|
|
|
def test_markdown_description_and_registration_window_remain_available(self):
|
|
assert '<strong>Important</strong>' in self.event.description_html
|
|
assert self.event.is_registration_open
|
|
self.event.registration_end_date = timezone.now() - timedelta(minutes=1)
|
|
assert not self.event.is_registration_open
|
|
|
|
def test_capacity_counts_confirmed_registration(self, mocker):
|
|
delay = mocker.patch('apps.events.tasks.send_registration_confirmation_notification.delay')
|
|
registration = Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED)
|
|
assert self.event.current_attendees_count == 1
|
|
assert not self.event.has_available_slots
|
|
assert RegistrationNotification.objects.filter(
|
|
registration=registration, kind=RegistrationNotification.Kind.CONFIRMED,
|
|
).exists()
|
|
delay.assert_not_called()
|
|
|
|
def test_notification_worker_retries_after_redis_failure(self, mocker):
|
|
registration = Registration.objects.create(
|
|
event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED,
|
|
)
|
|
entry = RegistrationNotification.objects.get(registration=registration)
|
|
notify = mocker.patch("apps.events.tasks.notify_user", side_effect=ConnectionError("redis unavailable"))
|
|
|
|
assert deliver_pending_registration_notifications() == 0
|
|
entry.refresh_from_db()
|
|
assert entry.delivered_at is None
|
|
|
|
notify.side_effect = None
|
|
assert deliver_pending_registration_notifications() == 1
|
|
entry.refresh_from_db()
|
|
registration.refresh_from_db()
|
|
assert entry.delivered_at is not None
|
|
assert registration.confirmation_notification_sent_at is not None
|
|
assert notify.call_args.args[1]["id"] == f"registration-{entry.pk}-confirmed"
|
|
assert deliver_pending_registration_notifications() == 0
|
|
|
|
def test_unlimited_capacity_allows_registration(self):
|
|
self.event.capacity = None
|
|
assert self.event.has_available_slots
|