import pytest import io import json import tempfile import uuid from datetime import datetime, time as datetime_time, 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_browse_events_returns_numbered_pages_and_global_price_ceiling(self): self._create_event(title='Price ceiling 650k toman', price=6500000) self._create_event(title='Price ceiling 4.3m toman', price=43000000) for index in range(13): self._create_event(title=f'Browse page event {index}', price=100000) first_page = self.client.get('/api/events/browse/', {'page': 1}) second_page = self.client.get('/api/events/browse/', {'page': 2}) assert first_page.status_code == 200 assert first_page.json()['count'] == 17 assert first_page.json()['page'] == 1 assert first_page.json()['page_size'] == 12 assert first_page.json()['total_pages'] == 2 assert len(first_page.json()['results']) == 12 assert second_page.json()['page'] == 2 assert len(second_page.json()['results']) == 5 assert first_page.json()['price_ceiling'] == 5000000 assert 'address' in first_page.json()['results'][0] def test_browse_events_upcoming_includes_future_events_that_cannot_register(self): now = timezone.now() open_event = self._create_event( title='Browse open event', start_time=now + timedelta(days=4), end_time=now + timedelta(days=4, hours=2), registration_start_date=now - timedelta(days=1), registration_end_date=now + timedelta(days=2), ) not_started = self._create_event( title='Browse not started event', start_time=now + timedelta(days=5), end_time=now + timedelta(days=5, hours=2), registration_start_date=now + timedelta(days=1), ) closed = self._create_event( title='Browse closed event', start_time=now + timedelta(days=6), end_time=now + timedelta(days=6, hours=2), registration_end_date=now - timedelta(minutes=1), ) full = self._create_event( title='Browse full event', start_time=now + timedelta(days=7), end_time=now + timedelta(days=7, hours=2), capacity=1, ) self._create_registration(full, self.user, status=Registration.StatusChoices.CONFIRMED) upcoming = self.client.get('/api/events/browse/', {'state': 'upcoming', 'search': 'Browse'}).json() open_results = self.client.get('/api/events/browse/', {'state': 'open', 'search': 'Browse'}).json() upcoming_ids = {item['id'] for item in upcoming['results']} open_ids = {item['id'] for item in open_results['results']} assert {not_started.id, closed.id, full.id}.issubset(upcoming_ids) assert open_event.id not in upcoming_ids assert open_ids == {open_event.id} def test_browse_events_filters_start_dates_inclusive_and_combines_type_price_and_free(self): selected_day = timezone.localdate() local_timezone = timezone.get_current_timezone() def at(day, hour, minute): return timezone.make_aware(datetime.combine(day, datetime_time(hour, minute)), local_timezone) start_day = self._create_event( title='Date range probe free', event_type=Event.TypeChoices.ONLINE, price=0, start_time=at(selected_day, 0, 0), end_time=at(selected_day, 2, 0), ) end_day = self._create_event( title='Date range probe paid online', event_type=Event.TypeChoices.ONLINE, price=300000, start_time=at(selected_day, 23, 59), end_time=at(selected_day + timedelta(days=1), 1, 0), ) self._create_event( title='Date range probe paid onsite', event_type=Event.TypeChoices.ON_SITE, price=300000, start_time=at(selected_day, 12, 0), end_time=at(selected_day, 14, 0), ) outside_day = self._create_event( title='Date range probe outside', event_type=Event.TypeChoices.ONLINE, price=300000, start_time=at(selected_day + timedelta(days=1), 0, 0), end_time=at(selected_day + timedelta(days=1), 2, 0), ) params = { 'search': 'Date range probe', 'date_from': selected_day.isoformat(), 'date_to': selected_day.isoformat(), 'event_type': Event.TypeChoices.ONLINE, 'price_min': 10000, 'price_max': 40000, } paid_results = self.client.get('/api/events/browse/', params).json() free_results = self.client.get('/api/events/browse/', {**params, 'free_only': 'true'}).json() assert {item['id'] for item in paid_results['results']} == {end_day.id} assert {item['id'] for item in free_results['results']} == {start_day.id} assert outside_day.id not in {item['id'] for item in paid_results['results']} 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 '

' 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)