Files
guilan-ace-backend/apps/events/tests/integration/test_events.py

394 lines
24 KiB
Python

import pytest
import io
import json
import tempfile
import uuid
from datetime import timedelta
from types import SimpleNamespace
from PIL import Image
from django.core.files.uploadedfile import SimpleUploadedFile
from django.utils import timezone
from core.authentication import create_jwt_token
from apps.events.api.schemas import EventSchema, EventGallerySchema, EventListSchema, RegistrationSchema, PaymentAdminSchema, EventAdminDetailSchema
from apps.events.api.views import list_events
from apps.events.models import Event, Registration
from apps.gallery.models import Gallery
from apps.payments.models import DiscountCode
from apps.users.models import Major, University, User
MEDIA_ROOT = tempfile.mkdtemp()
@pytest.mark.django_db
@pytest.mark.integration
class TestEventsAPIIntegrationTests:
password = 'TestPass123!'
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(username='event_user', email='event.user@example.com', mobile='09198000001', password=cls.password)
cls.user.is_email_verified = True
cls.user.save(update_fields=['is_email_verified'])
cls.staff = User.objects.create_user(username='event_staff', email='event.staff@example.com', mobile='09198000002', password=cls.password, is_staff=True)
cls.staff.is_email_verified = True
cls.staff.save(update_fields=['is_email_verified'])
cls.major, _ = Major.objects.get_or_create(code='CS', defaults={'name': 'Computer Science'})
cls.university, _ = University.objects.get_or_create(code='UT', defaults={'name': 'University of Tehran'})
cls.user.major = cls.major
cls.user.university = cls.university
cls.user.save(update_fields=['major', 'university'])
cls.staff.major = cls.major
cls.staff.university = cls.university
cls.staff.save(update_fields=['major', 'university'])
def setup_method(self):
type(self).setUpTestData()
self.token = create_jwt_token(self.user)
self.staff_token = create_jwt_token(self.staff)
self.event = self._create_event(title='Integration Event', description='Integration description.', status=Event.StatusChoices.PUBLISHED, price=0)
self.other_event = self._create_event(title='Other Published', description='Searchable', status=Event.StatusChoices.PUBLISHED, price=0)
def _auth_headers(self, token):
return {'HTTP_AUTHORIZATION': f'Bearer {token}'}
def _create_event(self, **overrides):
now = timezone.now()
defaults = {'title': 'Event Title', 'description': 'Description', 'start_time': now, 'end_time': now + timedelta(hours=2), 'registration_start_date': now - timedelta(days=1), 'registration_end_date': now + timedelta(days=5), 'slug': f'event-{uuid.uuid4().hex[:6]}', 'location': 'Campus', 'online_link': 'https://meet.example.com', 'price': 0, 'capacity': 10, 'status': Event.StatusChoices.PUBLISHED}
defaults.update(overrides)
return Event.objects.create(**defaults)
def _create_gallery_image(self):
buffer = io.BytesIO()
Image.new('RGB', (10, 10), color='blue').save(buffer, format='PNG')
buffer.seek(0)
file = SimpleUploadedFile('gallery.png', buffer.read(), content_type='image/png')
return Gallery.objects.create(title='Gallery image', description='desc', image=file, uploaded_by=self.user)
def _create_paid_event(self):
return self._create_event(price=30000, capacity=5)
def _create_registration(self, event, user, status=Registration.StatusChoices.PENDING):
return Registration.objects.create(event=event, user=user, status=status, final_price=event.price)
def test_list_events_filters_and_search(self):
response = self.client.get('/api/events/', {'status': 'published', 'search': 'Searchable'})
data = response.json()
assert response.status_code == 200
assert any((item['id'] == self.other_event.id for item in data))
def test_get_event_by_id_and_slug(self):
response_id = self.client.get(f'/api/events/{self.event.id}')
response_slug = self.client.get(f'/api/events/slug/{self.event.slug}')
assert response_id.status_code == 200
assert response_slug.status_code == 200
assert response_id.json()['id'] == self.event.id
assert response_slug.json()['slug'] == self.event.slug
def test_create_update_and_delete_event(self):
payload = {'title': 'New Event', 'description': 'Desc', 'start_time': (timezone.now() + timedelta(days=1)).isoformat(), 'end_time': (timezone.now() + timedelta(days=1, hours=1)).isoformat(), 'event_type': Event.TypeChoices.ON_SITE, 'status': Event.StatusChoices.DRAFT, 'price': 5000}
created = self.client.post('/api/events/', data=json.dumps(payload), content_type='application/json', **self._auth_headers(self.staff_token))
assert created.status_code == 201
event_id = created.json()['id']
updated = self.client.put(f'/api/events/{event_id}', data=json.dumps({'title': 'Updated Event'}), content_type='application/json', **self._auth_headers(self.staff_token))
assert updated.status_code == 200
assert updated.json()['title'] == 'Updated Event'
deleted = self.client.delete(f'/api/events/{event_id}', **self._auth_headers(self.staff_token))
assert deleted.status_code == 200
def test_admin_detail_and_registration_list_requires_staff(self):
staff_headers = self._auth_headers(self.staff_token)
user_headers = self._auth_headers(self.token)
_ = self._create_registration(self.event, self.user, status=Registration.StatusChoices.CONFIRMED)
list_resp = self.client.get(f'/api/events/{self.event.id}/admin-registrations', **user_headers)
assert list_resp.status_code == 403
list_resp = self.client.get(f'/api/events/{self.event.id}/admin-registrations', **staff_headers)
detail_resp = self.client.get(f'/api/events/{self.event.id}/admin-detail', **staff_headers)
assert list_resp.status_code == 200
assert detail_resp.status_code == 200
def test_list_events_filters_by_event_type_and_search(self):
event = self._create_event(title='Special Search', description='Unique discovery', event_type=Event.TypeChoices.ONLINE, status=Event.StatusChoices.PUBLISHED)
response = self.client.get('/api/events/', {'event_type': Event.TypeChoices.ONLINE, 'search': 'Unique discovery'})
assert response.status_code == 200
assert any((item['id'] == event.id for item in response.json()))
def test_list_events_handles_comma_status_parameter(self):
event = self._create_event(title='Comma Event', status=Event.StatusChoices.PUBLISHED)
results = list_events(None, status=f'{Event.StatusChoices.PUBLISHED},{Event.StatusChoices.DRAFT}', event_type=None, search=None, limit=10, offset=0)
assert event in list(results)
def test_create_event_attaches_gallery_images(self):
gallery = self._create_gallery_image()
payload = {'title': 'Gallery Event', 'description': 'Gallery desc', 'start_time': (timezone.now() + timedelta(days=1)).isoformat(), 'end_time': (timezone.now() + timedelta(days=1, hours=1)).isoformat(), 'event_type': Event.TypeChoices.ON_SITE, 'status': Event.StatusChoices.DRAFT, 'price': 5000, 'gallery_image_ids': [gallery.id]}
response = self.client.post('/api/events/', data=json.dumps(payload), content_type='application/json', **self._auth_headers(self.staff_token))
body = response.json()
assert response.status_code == 201
assert body['gallery_images']
updated = self.client.put(f'/api/events/{body['id']}', data=json.dumps({'title': 'Gallery Event Updated', 'gallery_image_ids': [gallery.id]}), content_type='application/json', **self._auth_headers(self.staff_token))
assert updated.status_code == 200
assert updated.json()['slug'] == 'gallery-event-updated'
assert updated.json()['gallery_images']
def test_admin_registration_filters_include_university_major_and_search(self):
event = self.event
self._create_registration(event, self.user, status=Registration.StatusChoices.CONFIRMED)
headers = self._auth_headers(self.staff_token)
response = self.client.get(f'/api/events/{event.id}/admin-registrations', {'university': self.user.university.code, 'major': self.user.major.code, 'search': self.user.username, 'status': [Registration.StatusChoices.CONFIRMED, Registration.StatusChoices.PENDING]}, **headers)
assert response.status_code == 200
assert response.json()['count'] == 1
def test_register_before_start_and_after_end_dates_fail(self):
future_event = self._create_event(registration_start_date=timezone.now() + timedelta(days=1))
future_response = self.client.post(f'/api/events/{future_event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert future_response.status_code == 400
closed_event = self._create_event(registration_end_date=timezone.now() - timedelta(hours=1))
closed_response = self.client.post(f'/api/events/{closed_event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert closed_response.status_code == 400
def test_register_recreates_after_cancelled_registration(self):
event = self._create_event(price=0)
Registration.objects.create(event=event, user=self.user, status=Registration.StatusChoices.CANCELLED, final_price=0)
response = self.client.post(f'/api/events/{event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}', content_type='application/json')
assert response.status_code == 200
assert response.json()['status'] == Registration.StatusChoices.CONFIRMED
def test_register_updates_final_price_when_none(self):
event = self._create_paid_event()
registration = Registration.objects.create(event=event, user=self.user, status=Registration.StatusChoices.PENDING, final_price=None)
response = self.client.post(f'/api/events/{event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}', content_type='application/json')
assert response.status_code == 200
assert response.json()['final_price'] == event.price
def _create_discount_code(self, event):
code = DiscountCode.objects.create(code=f'CODE-{uuid.uuid4().hex[:4]}', value=50, type=DiscountCode.Type.PERCENT, is_active=True)
code.applicable_events.add(event)
return code
def test_register_for_event_with_free_price_confirms(self):
event = self._create_event(price=0)
response = self.client.post(f'/api/events/{event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}', content_type='application/json')
assert response.status_code == 200
assert response.json()['status'] == Registration.StatusChoices.CONFIRMED
def test_register_for_event_with_discount_updates_final_price(self):
event = self._create_paid_event()
code = self._create_discount_code(event)
response = self.client.post(f'/api/events/{event.id}/register', data=json.dumps({'discount_code': code.code}), content_type='application/json', HTTP_AUTHORIZATION=f'Bearer {self.token}')
result = response.json()
assert response.status_code == 200
assert result['discount_code'] == code.code
assert result['discount_amount'] == event.price // 2
assert result['final_price'] == event.price // 2
def test_register_fails_when_capacity_full(self):
event = self._create_event(capacity=1)
other = self._create_event_user('other_user', 'other@example.com')
Registration.objects.create(event=event, user=other, status=Registration.StatusChoices.CONFIRMED, final_price=0)
response = self.client.post(f'/api/events/{event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}', content_type='application/json')
assert response.status_code == 400
def _create_event_user(self, username, email):
suffix = str(abs(hash(username)) % 1000000).zfill(6)
user = User.objects.create_user(username=username, email=email, mobile=f'09190{suffix}', password=self.password)
user.is_email_verified = True
user.save(update_fields=['is_email_verified'])
user.major = self.user.major
user.university = self.user.university
user.save(update_fields=['major', 'university'])
return user
def test_register_rejects_duplicate_confirmed(self):
event = self._create_event(price=0)
Registration.objects.create(event=event, user=self.user, status=Registration.StatusChoices.CONFIRMED, final_price=0)
response = self.client.post(f'/api/events/{event.id}/register', HTTP_AUTHORIZATION=f'Bearer {self.token}', content_type='application/json')
assert response.status_code == 400
def test_registration_status_update_rejects_self_attendance_and_allows_cancel(self):
event = self._create_event(price=0)
registration = self._create_registration(event, self.user)
original_status = registration.status
update = self.client.put(f'/api/events/registrations/{registration.id}', data=json.dumps({'status': Registration.StatusChoices.ATTENDED}), content_type='application/json', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert update.status_code == 403
registration.refresh_from_db()
assert registration.status == original_status
cancel = self.client.delete(f'/api/events/registrations/{registration.id}', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert cancel.status_code == 200
assert cancel.json()['message'] == 'ثبت\u200cنام شما لغو شد :('
def test_verify_registration_and_my_registrations(self):
event = self._create_event(price=0)
registration = self._create_registration(event, self.user, status=Registration.StatusChoices.CONFIRMED)
verify = self.client.get(f'/api/events/registerations/verify/{registration.ticket_id}', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert verify.status_code == 200
assert verify.json()['ticket_id'] == str(registration.ticket_id)
my_regs = self.client.get('/api/events/my-registrations', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert my_regs.status_code == 200
assert len(my_regs.json()) > 0
status_resp = self.client.get(f'/api/events/{event.id}/is-registered', HTTP_AUTHORIZATION=f'Bearer {self.token}')
assert status_resp.status_code == 200
assert status_resp.json()['is_registered']
def test_my_registrations_pagination_is_scoped_by_status(self):
for index in range(12):
event = self._create_event(title=f'Confirmed {index}')
self._create_registration(event, self.user, status=Registration.StatusChoices.CONFIRMED)
cancelled_event = self._create_event(title='Cancelled entry')
self._create_registration(cancelled_event, self.user, status=Registration.StatusChoices.CANCELLED)
first_page = self.client.get(
'/api/events/my-registrations/paginated',
{'status': Registration.StatusChoices.CONFIRMED, 'page': 1},
**self._auth_headers(self.token),
)
second_page = self.client.get(
'/api/events/my-registrations/paginated',
{'status': Registration.StatusChoices.CONFIRMED, 'page': 2},
**self._auth_headers(self.token),
)
cancelled_page = self.client.get(
'/api/events/my-registrations/paginated',
{'status': Registration.StatusChoices.CANCELLED, 'page': 1},
**self._auth_headers(self.token),
)
assert first_page.status_code == 200, first_page.content
assert first_page.json()['count'] == 12
assert first_page.json()['page_size'] == 10
assert first_page.json()['total_pages'] == 2
assert len(first_page.json()['results']) == 10
assert len(second_page.json()['results']) == 2
assert all(item['status'] == Registration.StatusChoices.CONFIRMED for item in first_page.json()['results'])
assert cancelled_page.json()['count'] == 1
assert cancelled_page.json()['results'][0]['status'] == Registration.StatusChoices.CANCELLED
all_first_page = self.client.get(
'/api/events/my-registrations/paginated',
{'status': 'all', 'page': 1},
**self._auth_headers(self.token),
)
all_second_page = self.client.get(
'/api/events/my-registrations/paginated',
{'status': 'all', 'page': 2},
**self._auth_headers(self.token),
)
assert all_first_page.status_code == 200, all_first_page.content
assert all_first_page.json()['count'] == 13
assert all_first_page.json()['total_pages'] == 2
assert len(all_first_page.json()['results']) == 10
assert len(all_second_page.json()['results']) == 3
assert {item['status'] for item in all_first_page.json()['results'] + all_second_page.json()['results']} == {
Registration.StatusChoices.CONFIRMED,
Registration.StatusChoices.CANCELLED,
}
def test_my_registrations_supports_status_filter_and_event_ordering(self):
now = timezone.now()
older = self._create_event(title='Bravo Event', slug='bravo-event', start_time=now - timedelta(days=10))
middle = self._create_event(title='Alpha Event', slug='alpha-event', start_time=now - timedelta(days=5))
newest = self._create_event(title='Charlie Event', slug='charlie-event', start_time=now - timedelta(days=1))
self._create_registration(older, self.user, status=Registration.StatusChoices.CANCELLED)
self._create_registration(middle, self.user, status=Registration.StatusChoices.CONFIRMED)
self._create_registration(newest, self.user, status=Registration.StatusChoices.PENDING)
endpoint = '/api/events/my-registrations/paginated'
headers = self._auth_headers(self.token)
all_status_default_order = self.client.get(endpoint, {'status': 'all'}, **headers)
legacy_default = self.client.get(endpoint, **headers)
confirmed_only = self.client.get(endpoint, {'status': 'confirmed'}, **headers)
start_time_ascending = self.client.get(endpoint, {'status': 'all', 'sort_by': 'start_time', 'sort_order': 'asc'}, **headers)
title_ascending = self.client.get(endpoint, {'status': 'all', 'sort_by': 'title', 'sort_order': 'asc'}, **headers)
title_descending = self.client.get(endpoint, {'status': 'all', 'sort_by': 'title', 'sort_order': 'desc'}, **headers)
invalid_sort = self.client.get(endpoint, {'sort_by': 'created_at'}, **headers)
invalid_direction = self.client.get(endpoint, {'sort_order': 'sideways'}, **headers)
assert all_status_default_order.status_code == 200, all_status_default_order.content
assert all_status_default_order.json()['count'] == 3
assert [row['event']['slug'] for row in all_status_default_order.json()['results']] == [
'charlie-event', 'alpha-event', 'bravo-event',
]
assert legacy_default.json()['count'] == 1
assert legacy_default.json()['results'][0]['status'] == Registration.StatusChoices.CONFIRMED
assert confirmed_only.json()['count'] == 1
assert confirmed_only.json()['results'][0]['status'] == Registration.StatusChoices.CONFIRMED
assert [row['event']['slug'] for row in start_time_ascending.json()['results']] == [
'bravo-event', 'alpha-event', 'charlie-event',
]
assert [row['event']['title'] for row in title_ascending.json()['results']] == [
'Alpha Event', 'Bravo Event', 'Charlie Event',
]
assert [row['event']['title'] for row in title_descending.json()['results']] == [
'Charlie Event', 'Bravo Event', 'Alpha Event',
]
assert invalid_sort.status_code == 422
assert invalid_direction.status_code == 422
def test_list_event_registrations(self):
event = self.event
self._create_registration(event, self.user)
response = self.client.get(f'/api/events/{event.id}/registrations')
assert response.status_code == 200
assert response.json()
def test_list_event_registrations_admin_filters(self):
event = self.event
self._create_registration(event, self.user, status=Registration.StatusChoices.PENDING)
headers = self._auth_headers(self.staff_token)
response = self.client.get(f'/api/events/{event.id}/admin-registrations', {'status': [Registration.StatusChoices.PENDING]}, **headers)
assert response.status_code == 200
assert response.json()['count'] == 1
@pytest.fixture(autouse=True)
def _pytest_client(self, client):
self.client = client
@pytest.fixture(autouse=True)
def _pytest_settings(self, settings):
settings.MEDIA_ROOT = MEDIA_ROOT
@pytest.mark.django_db
@pytest.mark.integration
class TestEventSchemasIntegrationTests:
password = 'SchemaPass!123'
def setup_method(self):
self.user = User.objects.create_user(username='schema_user', email='schema.user@example.com', mobile='09198000003', password=self.password)
self.user.is_email_verified = True
self.user.save(update_fields=['is_email_verified'])
self.event = Event.objects.create(title='Schema Event', description='**bold**', start_time=timezone.now(), end_time=timezone.now() + timedelta(hours=1), registration_start_date=timezone.now() - timedelta(days=1), registration_end_date=timezone.now() + timedelta(days=1), price=1000, slug='schema-event')
Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED, final_price=0)
Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.ATTENDED, final_price=0)
def _mock_request(self):
return SimpleNamespace(build_absolute_uri=lambda path: f'https://test{path}')
def test_gallery_schema_returns_full_url(self):
obj = SimpleNamespace(image=SimpleNamespace(url='/media/gallery.png'))
result = EventGallerySchema.resolve_absolute_image_url(obj, {'request': self._mock_request()})
assert result == 'https://test/media/gallery.png'
def test_event_schema_resolvers(self):
context = {'request': self._mock_request()}
event_obj = SimpleNamespace(featured_image=SimpleNamespace(url='/media/feat.png'), registrations=self.event.registrations)
assert EventSchema.resolve_absolute_featured_image_url(event_obj, context) == 'https://test/media/feat.png'
assert EventSchema.resolve_registration_count(self.event) == 2
assert '<p>' in EventSchema.resolve_description_html(self.event)
def test_event_list_schema_resolvers(self):
obj = SimpleNamespace(featured_image=SimpleNamespace(url='/media/feat.png'), registrations=self.event.registrations)
context = {'request': self._mock_request()}
assert EventListSchema.resolve_absolute_featured_image_url(obj, context) == 'https://test/media/feat.png'
assert EventListSchema.resolve_registration_count(self.event) == 2
def test_registration_schema_resolves_discount_code(self):
discount = DiscountCode.objects.create(code='SCHEMA', type=DiscountCode.Type.FIXED, value=100, is_active=True)
discount.applicable_events.add(self.event)
registration = Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED, final_price=900, discount_code=discount)
assert RegistrationSchema.resolve_discount_code(registration) == discount.code
def test_payment_admin_schema_normalizes_discount_code(self):
assert PaymentAdminSchema.normalize_discount_code(None) is None
assert PaymentAdminSchema.normalize_discount_code('123') == '123'
assert PaymentAdminSchema.normalize_discount_code(SimpleNamespace(code='ABC')) == 'ABC'
def test_event_admin_detail_resolves_registrations(self):
registrations = EventAdminDetailSchema.resolve_registrations(self.event)
assert list(registrations)