Amirhossein Khalili
ff99130938
Split backend tests into unit and integration suites, add isolated fixtures and coverage enforcement, and update CI and test documentation. Cover current authentication, content, notification, payment, media, and admin flows; fix behaviors revealed by the new tests. Validate both suites and the per-module gate against local PostgreSQL.
298 lines
18 KiB
Python
298 lines
18 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_and_cancel(self):
|
|
event = self._create_event(price=0)
|
|
registration = self._create_registration(event, self.user)
|
|
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 == 200
|
|
assert update.json()['status'] == Registration.StatusChoices.ATTENDED
|
|
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_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)
|