init
Some checks failed
CI/CD / Backend & Frontend Checks (push) Has been cancelled
CI/CD / Deploy to Production (push) Has been cancelled

This commit is contained in:
2026-05-18 11:34:07 +03:30
commit 7a8ddeabed
279 changed files with 37390 additions and 0 deletions

View File

View File

@@ -0,0 +1,540 @@
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.test import TestCase, override_settings
from django.utils import timezone
from api.authentication import create_jwt_token
from api.schemas.events import (
EventSchema,
EventGallerySchema,
EventListSchema,
RegistrationSchema,
PaymentAdminSchema,
EventAdminDetailSchema,
)
from api.views.events import list_events
from events.models import Event, Registration
from gallery.models import Gallery
from payments.models import DiscountCode
from users.models import Major, University, User
MEDIA_ROOT = tempfile.mkdtemp()
@override_settings(MEDIA_ROOT=MEDIA_ROOT)
class EventsAPIIntegrationTests(TestCase):
password = "TestPass123!"
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username="event_user",
email="event.user@example.com",
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",
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(self):
super().setUp()
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)
# Basic event endpoints ------------------------------------------------
def test_list_events_filters_and_search(self):
# Act
response = self.client.get("/api/events/", {"status": "published", "search": "Searchable"})
data = response.json()
# Assert
self.assertEqual(response.status_code, 200)
self.assertTrue(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}")
self.assertEqual(response_id.status_code, 200)
self.assertEqual(response_slug.status_code, 200)
self.assertEqual(response_id.json()["id"], self.event.id)
self.assertEqual(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.assertEqual(created.status_code, 200)
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.assertEqual(updated.status_code, 200)
self.assertEqual(updated.json()["title"], "Updated Event")
deleted = self.client.delete(f"/api/events/{event_id}")
self.assertEqual(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)
# Non staff forbidden
list_resp = self.client.get(f"/api/events/{self.event.id}/admin-registrations", **user_headers)
self.assertEqual(list_resp.status_code, 403)
# Staff allowed
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)
self.assertEqual(list_resp.status_code, 200)
self.assertEqual(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",
},
)
self.assertEqual(response.status_code, 200)
self.assertTrue(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,
)
self.assertIn(event, 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",
)
body = response.json()
self.assertEqual(response.status_code, 200)
self.assertTrue(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.assertEqual(updated.status_code, 200)
self.assertEqual(updated.json()["slug"], "gallery-event-updated")
self.assertTrue(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,
)
self.assertEqual(response.status_code, 200)
self.assertEqual(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}",
)
self.assertEqual(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}",
)
self.assertEqual(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",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(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",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(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",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(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()
self.assertEqual(response.status_code, 200)
self.assertEqual(result["discount_code"], code.code)
self.assertEqual(result["discount_amount"], event.price // 2)
self.assertEqual(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",
)
self.assertEqual(response.status_code, 400)
def _create_event_user(self, username, email):
user = User.objects.create_user(username=username, email=email, 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",
)
self.assertEqual(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}",
)
self.assertEqual(update.status_code, 200)
self.assertEqual(update.json()["status"], Registration.StatusChoices.ATTENDED)
cancel = self.client.delete(
f"/api/events/registrations/{registration.id}",
HTTP_AUTHORIZATION=f"Bearer {self.token}",
)
self.assertEqual(cancel.status_code, 200)
self.assertEqual(cancel.json()["message"], "ثبت‌نام شما لغو شد :(")
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}",
)
self.assertEqual(verify.status_code, 200)
self.assertEqual(verify.json()["ticket_id"], str(registration.ticket_id))
my_regs = self.client.get(
"/api/events/my-registrations",
HTTP_AUTHORIZATION=f"Bearer {self.token}",
)
self.assertEqual(my_regs.status_code, 200)
self.assertGreater(len(my_regs.json()), 0)
status_resp = self.client.get(
f"/api/events/{event.id}/is-registered",
HTTP_AUTHORIZATION=f"Bearer {self.token}",
)
self.assertEqual(status_resp.status_code, 200)
self.assertTrue(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")
self.assertEqual(response.status_code, 200)
self.assertTrue(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,
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["count"], 1)
class EventSchemasIntegrationTests(TestCase):
password = "SchemaPass!123"
def setUp(self):
self.user = User.objects.create_user(
username="schema_user",
email="schema.user@example.com",
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()})
self.assertEqual(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)
self.assertEqual(EventSchema.resolve_absolute_featured_image_url(event_obj, context), "https://test/media/feat.png")
self.assertEqual(EventSchema.resolve_registration_count(self.event), 2)
self.assertIn("<p>", 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()}
self.assertEqual(EventListSchema.resolve_absolute_featured_image_url(obj, context), "https://test/media/feat.png")
self.assertEqual(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,
)
self.assertEqual(RegistrationSchema.resolve_discount_code(registration), discount.code)
def test_payment_admin_schema_normalizes_discount_code(self):
self.assertIsNone(PaymentAdminSchema.normalize_discount_code(None))
self.assertEqual(PaymentAdminSchema.normalize_discount_code("123"), "123")
self.assertEqual(PaymentAdminSchema.normalize_discount_code(SimpleNamespace(code="ABC")), "ABC")
def test_event_admin_detail_resolves_registrations(self):
registrations = EventAdminDetailSchema.resolve_registrations(self.event)
self.assertTrue(list(registrations))
# TODO registration-related tests

View File

@@ -0,0 +1,282 @@
import json
from datetime import timedelta
from unittest import mock
from django.test import TestCase, override_settings
from django.utils import timezone
from api.authentication import create_jwt_token
from events.models import Event, Registration
from payments.models import Payment, DiscountCode
from users.models import User
@override_settings(
ZARINPAL_MERCHANT_ID="MID",
ZARINPAL_REQUEST_URL="https://zarinpal/request",
ZARINPAL_STARTPAY="https://zarinpal/start/",
ZARINPAL_VERIFY_URL="https://zarinpal/verify",
ZARINPAL_CALLBACK_URL="https://frontend/callback",
)
class PaymentsAPIIntegrationTests(TestCase):
password = "PaymentPass!123"
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username="pay_user",
email="pay.user@example.com",
password=cls.password,
)
cls.user.is_email_verified = True
cls.user.save(update_fields=["is_email_verified"])
def setUp(self):
super().setUp()
self.event = Event.objects.create(
title="Pay Event",
description="Payment event",
start_time=timezone.now(),
end_time=timezone.now() + timedelta(hours=2),
registration_start_date=timezone.now() - timedelta(days=1),
registration_end_date=timezone.now() + timedelta(days=1),
slug="pay-event",
price=50000,
capacity=10,
status=Event.StatusChoices.PUBLISHED,
)
self.token = create_jwt_token(self.user)
def _headers(self):
return {"HTTP_AUTHORIZATION": f"Bearer {self.token}"}
def _create_paid_event(self):
return Event.objects.create(
title="Paid Event",
description="Paid",
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=2),
slug=f"paid-{timezone.now().timestamp()}",
price=20000,
capacity=5,
status=Event.StatusChoices.PUBLISHED,
)
def _create_discount_code(self, event):
code = DiscountCode.objects.create(
code="DISC50",
value=50,
is_active=True,
)
code.applicable_events.add(event)
return code
def test_create_payment_for_free_event(self):
free = Event.objects.create(
title="Free",
description="Zero",
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),
slug="free-event",
price=0,
capacity=10,
status=Event.StatusChoices.PUBLISHED,
)
response = self.client.post(
"/api/payments/create",
data=json.dumps(
{
"event_id": free.id,
"description": "Free registration",
}
),
content_type="application/json",
**self._headers(),
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["amount"], 0)
self.assertIsNone(data["start_pay_url"])
@mock.patch("api.views.payments.requests.post")
def test_create_payment_with_discount(self, mock_post):
mock_response = mock.Mock()
mock_response.json.return_value = {"data": {"code": 100, "authority": "AUTH"}}
mock_post.return_value = mock_response
code = self._create_discount_code(self.event)
response = self.client.post(
"/api/payments/create",
data=json.dumps(
{
"event_id": self.event.id,
"description": "Pay with discount",
"discount_code": code.code,
}
),
content_type="application/json",
**self._headers(),
)
self.assertEqual(response.status_code, 200)
payload = response.json()
self.assertEqual(payload["discount_amount"], self.event.price // 2)
self.assertEqual(payload["amount"], self.event.price // 2)
self.assertIn("start_pay_url", payload)
payment = Payment.objects.get(user=self.user, event=self.event)
self.assertEqual(payment.discount_code, code)
@mock.patch("api.views.payments.requests.post")
def test_callback_success_marks_paid(self, mock_post):
payment = Payment.objects.create(
user=self.user,
event=self.event,
base_amount=self.event.price,
amount=self.event.price,
status=Payment.OrderStatusChoices.PENDING,
authority="AUTH123",
)
mock_resp = mock.Mock()
mock_resp.json.return_value = {"data": {"code": 100, "ref_id": "REF", "card_pan": "123", "card_hash": "ABC"}}
mock_post.return_value = mock_resp
response = self.client.get(
"/api/payments/callback",
{"Authority": "AUTH123", "Status": "OK"},
)
payment.refresh_from_db()
self.assertEqual(payment.status, Payment.OrderStatusChoices.PAID)
self.assertTrue("status=success" in response.url)
@mock.patch("api.views.payments.requests.post")
def test_callback_failure_redirects_failed(self, mock_post):
payment = Payment.objects.create(
user=self.user,
event=self.event,
base_amount=self.event.price,
amount=self.event.price,
status=Payment.OrderStatusChoices.PENDING,
authority="AUTH456",
)
mock_resp = mock.Mock()
mock_resp.json.return_value = {"data": {"code": 101, "ref_id": "REF"}}
mock_post.return_value = mock_resp
response = self.client.get(
"/api/payments/callback",
{"Authority": "AUTH456", "Status": "OK"},
)
payment.refresh_from_db()
self.assertEqual(payment.status, Payment.OrderStatusChoices.PAID)
self.assertTrue("status=success" in response.url)
def test_callback_missing_authority_returns_error(self):
response = self.client.get("/api/payments/callback", {"Status": "OK"})
self.assertEqual(response.status_code, 400)
def test_callback_not_ok_cancels(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="AUTH789",
)
response = self.client.get(
"/api/payments/callback",
{"Authority": "AUTH789", "Status": "NOK"},
)
payment.refresh_from_db()
self.assertEqual(payment.status, Payment.OrderStatusChoices.CANCELED)
self.assertIn("status=failed", response.url)
@mock.patch("api.views.payments.requests.post", side_effect=RuntimeError("down"))
def test_create_payment_gateway_failure(self, mock_post):
response = self.client.post(
"/api/payments/create",
data=json.dumps(
{
"event_id": self.event.id,
"description": "Gateway fail",
}
),
content_type="application/json",
**self._headers(),
)
self.assertEqual(response.status_code, 502)
self.assertFalse(Payment.objects.filter(user=self.user).exists())
def test_create_payment_when_already_paid(self):
Payment.objects.create(
user=self.user,
event=self.event,
base_amount=self.event.price,
amount=self.event.price,
status=Payment.OrderStatusChoices.PAID,
)
response = self.client.post(
"/api/payments/create",
data=json.dumps({"event_id": self.event.id, "description": "Duplicate"}),
content_type="application/json",
**self._headers(),
)
self.assertEqual(response.status_code, 400)
@mock.patch("api.views.payments.requests.post")
def test_registration_final_price_none_updates(self, mock_post):
registration = Registration.objects.create(
event=self.event,
user=self.user,
status=Registration.StatusChoices.PENDING,
final_price=None,
)
mock_response = mock.Mock()
mock_response.json.return_value = {"data": {"code": 100, "authority": "AUTH"}}
mock_post.return_value = mock_response
response = self.client.post(
"/api/payments/create",
data=json.dumps({"event_id": self.event.id, "description": "Update"}),
content_type="application/json",
**self._headers(),
)
self.assertEqual(response.status_code, 200)
registration.refresh_from_db()
if registration.final_price is None:
self.fail("final_price should be populated")
def test_coupon_check_success_and_errors(self):
code = DiscountCode.objects.create(code="PAYCO", value=20, is_active=True, type=DiscountCode.Type.PERCENT)
code.applicable_events.add(self.event)
# missing code
missing = self.client.post(
"/api/payments/coupon/check",
data=json.dumps({"event_id": self.event.id}),
content_type="application/json",
**self._headers(),
)
self.assertEqual(missing.status_code, 422)
# invalid code
invalid = self.client.post(
"/api/payments/coupon/check",
data=json.dumps({"event_id": self.event.id, "code": "INVALID"}),
content_type="application/json",
**self._headers(),
)
self.assertEqual(invalid.status_code, 404)
success = self.client.post(
"/api/payments/coupon/check",
data=json.dumps({"event_id": self.event.id, "code": code.code}),
content_type="application/json",
**self._headers(),
)
self.assertEqual(success.status_code, 200)
self.assertIn("final_price", success.json())

View File

@@ -0,0 +1,724 @@
import json
import shutil
import tempfile
import uuid
from datetime import timedelta
from unittest import mock
from django.conf import settings
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase, override_settings
from django.utils import timezone
import jwt
from users.models import User, Major, University
class UsersAPIIntegrationTests(TestCase):
password = "Sup3rSecure!123"
@classmethod
def setUpTestData(cls):
super().setUpTestData()
cls.major_cs, _ = Major.objects.get_or_create(
code="CS", defaults={"name": "Computer Science"}
)
cls.major_gil, _ = Major.objects.get_or_create(
code="GIL_CS", defaults={"name": "Gilan Computer Science"}
)
cls.university_ut, _ = University.objects.get_or_create(
code="UT", defaults={"name": "University of Tehran"}
)
cls.university_gilan, _ = University.objects.get_or_create(
code="GILAN", defaults={"name": "Gilan University"}
)
def setUp(self):
super().setUp()
patchers = [
mock.patch("users.tasks.send_verification_email.delay"),
mock.patch("users.signals.send_verification_email.delay"),
mock.patch("users.tasks.send_password_reset_email.delay"),
]
(
self.mock_send_verification_task,
self.mock_signal_verification_task,
self.mock_password_reset_task,
) = [patcher.start() for patcher in patchers]
for patcher in patchers:
self.addCleanup(patcher.stop)
# Helper utilities -----------------------------------------------------
def _numeric_student_id(self) -> str:
return str(uuid.uuid4().int)[-10:]
def _resolve_major(self, value):
if value is None:
return None
if isinstance(value, Major):
return value
return Major.objects.filter(code=value).first()
def _resolve_university(self, value):
if value is None:
return None
if isinstance(value, University):
return value
return University.objects.filter(code=value).first()
def _create_user(self, **overrides) -> User:
unique = uuid.uuid4().hex[:8]
defaults = {
"username": f"user_{unique}",
"email": f"{unique}@example.com",
"student_id": self._numeric_student_id(),
"first_name": "Test",
"last_name": "User",
"year_of_study": 2,
"major": self.major_cs,
"university": self.university_ut,
}
defaults.update(overrides)
if isinstance(defaults.get("major"), str):
defaults["major"] = self._resolve_major(defaults["major"])
if isinstance(defaults.get("university"), str):
defaults["university"] = self._resolve_university(defaults["university"])
password = defaults.pop("password", self.password)
return User.objects.create_user(password=password, **defaults)
def _auth_headers(self, token: str) -> dict:
return {"HTTP_AUTHORIZATION": f"Bearer {token}"}
def _login_and_get_tokens(self, user: User, password: str | None = None) -> dict:
response = self.client.post(
"/api/auth/login",
data=json.dumps({"email": user.email, "password": password or self.password}),
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
return response.json()
def _refresh_token_value(self, user: User | None = None, **overrides) -> str:
now = timezone.now()
payload = {
"type": "refresh",
"exp": now + timedelta(minutes=5),
"iat": now,
}
if user is not None:
payload["user_id"] = user.id
payload.update(overrides)
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
# Registration ---------------------------------------------------------
def test_register_creates_user_and_enqueues_signal(self):
# Arrange
payload = {
"username": "integration_user",
"email": "integration@example.com",
"password": "RegisterPass!9",
"student_id": "2023123456",
"first_name": "Integration",
"last_name": "Tester",
"university": self.university_ut.code,
"major": self.major_cs.code,
"year_of_study": 3,
}
# Act
response = self.client.post(
"/api/auth/register", data=json.dumps(payload), content_type="application/json"
)
# Assert
self.assertEqual(response.status_code, 201)
self.assertTrue(User.objects.filter(email=payload["email"]).exists())
self.assertTrue(self.mock_signal_verification_task.called)
def test_register_rejects_short_student_id(self):
# Arrange
payload = {
"username": "short_id",
"email": "short@example.com",
"password": "RegisterPass!9",
"student_id": "123456789", # 9 digits
}
# Act
response = self.client.post(
"/api/auth/register", data=json.dumps(payload), content_type="application/json"
)
# Assert
self.assertEqual(response.status_code, 400)
def test_register_rejects_duplicate_username(self):
# Arrange
existing = self._create_user(username="duplicate")
payload = {
"username": existing.username,
"email": "someone@example.com",
"password": "RegisterPass!9",
}
# Act
response = self.client.post(
"/api/auth/register", data=json.dumps(payload), content_type="application/json"
)
# Assert
self.assertEqual(response.status_code, 400)
def test_register_rejects_duplicate_email(self):
# Arrange
existing = self._create_user(email="duplicate@example.com")
payload = {
"username": "newuser",
"email": existing.email,
"password": "RegisterPass!9",
}
# Act
response = self.client.post(
"/api/auth/register", data=json.dumps(payload), content_type="application/json"
)
# Assert
self.assertEqual(response.status_code, 400)
def test_register_rejects_duplicate_student_id_in_same_university(self):
# Arrange
student_id = "2023012345"
self._create_user(student_id=student_id, university=self.university_gilan)
payload = {
"username": "dupstudent",
"email": "dupstudent@example.com",
"password": "RegisterPass!9",
"student_id": student_id,
"university": self.university_gilan.code,
}
# Act
response = self.client.post(
"/api/auth/register", data=json.dumps(payload), content_type="application/json"
)
# Assert
self.assertEqual(response.status_code, 400)
# Login & Refresh ------------------------------------------------------
def test_login_returns_tokens_for_verified_user(self):
# Arrange
user = self._create_user()
user.is_email_verified = True
user.save(update_fields=["is_email_verified"])
# Act
response = self.client.post(
"/api/auth/login",
data=json.dumps({"email": user.email, "password": self.password}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 200)
body = response.json()
self.assertIn("access_token", body)
self.assertIn("refresh_token", body)
def test_login_rejects_unverified_user(self):
# Arrange
user = self._create_user()
# Act
response = self.client.post(
"/api/auth/login",
data=json.dumps({"email": user.email, "password": self.password}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_login_rejects_inactive_user(self):
# Arrange
user = self._create_user(is_email_verified=True, is_active=False)
# Act
response = self.client.post(
"/api/auth/login",
data=json.dumps({"email": user.email, "password": self.password}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_refresh_returns_tokens(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": tokens["refresh_token"]}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 200)
refreshed = response.json()
self.assertIn("access_token", refreshed)
self.assertIn("refresh_token", refreshed)
def test_refresh_rejects_non_refresh_token(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": tokens["access_token"]}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_refresh_rejects_missing_user_id(self):
# Arrange
token = self._refresh_token_value()
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": token}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_refresh_rejects_unverified_user(self):
# Arrange
user = self._create_user()
token = self._refresh_token_value(user=user)
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": token}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_refresh_rejects_inactive_user(self):
# Arrange
user = self._create_user(is_email_verified=True, is_active=False)
token = self._refresh_token_value(user=user)
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": token}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_refresh_rejects_expired_token(self):
# Arrange
user = self._create_user(is_email_verified=True)
token = self._refresh_token_value(
user=user,
exp=timezone.now() - timedelta(minutes=1),
)
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": token}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
def test_refresh_rejects_invalid_token_string(self):
# Arrange
token = "not-a-valid-token"
# Act
response = self.client.post(
"/api/auth/refresh",
data=json.dumps({"refresh_token": token}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 401)
# Email verification ---------------------------------------------------
def test_verify_email_marks_user_verified(self):
# Arrange
user = self._create_user()
token = str(user.email_verification_token)
# Act
response = self.client.get(f"/api/auth/verify-email/{token}")
# Assert
self.assertEqual(response.status_code, 200)
user.refresh_from_db()
self.assertTrue(user.is_email_verified)
def test_verify_email_rejects_unknown_token(self):
# Arrange
token = uuid.uuid4()
# Act
response = self.client.get(f"/api/auth/verify-email/{token}")
# Assert
self.assertEqual(response.status_code, 404)
def test_resend_verification_rejects_unknown_email(self):
# Arrange
payload = {"email": "missing@example.com"}
# Act
response = self.client.post(f"/api/auth/resend-verification?email={payload['email']}")
# Assert
self.assertEqual(response.status_code, 404)
# Profiles -------------------------------------------------------------
def test_get_profile_returns_schema_fields(self):
# Arrange
user = self._create_user(major=self.major_cs, university=self.university_gilan)
user.is_email_verified = True
user.save(update_fields=["is_email_verified"])
tokens = self._login_and_get_tokens(user)
# Act
response = self.client.get("/api/auth/profile", **self._auth_headers(tokens["access_token"]))
# Assert
self.assertEqual(response.status_code, 200)
profile = response.json()
self.assertEqual(profile["major"], user.get_major_display())
self.assertEqual(profile["university"], user.get_university_display())
def test_get_profile_requires_authentication(self):
# Arrange
# No token supplied.
# Act
response = self.client.get("/api/auth/profile")
# Assert
self.assertEqual(response.status_code, 401)
def test_update_profile_persists_changes(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
payload = {"bio": "Updated bio", "year_of_study": 4}
# Act
response = self.client.put(
"/api/auth/profile",
data=json.dumps(payload),
content_type="application/json",
**self._auth_headers(tokens["access_token"]),
)
# Assert
self.assertEqual(response.status_code, 200)
user.refresh_from_db()
self.assertEqual(user.bio, payload["bio"])
self.assertEqual(user.year_of_study, payload["year_of_study"])
@override_settings(MEDIA_URL="/media/", MEDIA_ROOT=tempfile.gettempdir())
def test_upload_profile_picture_succeeds(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
image = SimpleUploadedFile(
"avatar.png", b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR", content_type="image/png"
)
# Act
response = self.client.post(
"/api/auth/profile/picture", {"file": image}, **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 200)
profile = self.client.get(
"/api/auth/profile", **self._auth_headers(tokens["access_token"])
).json()
self.assertIn("profile_pictures", profile["profile_picture"])
def test_upload_profile_picture_requires_file(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
# Act
response = self.client.post(
"/api/auth/profile/picture", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 400)
def test_upload_profile_picture_rejects_invalid_type(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
text_file = SimpleUploadedFile("doc.txt", b"text", content_type="text/plain")
# Act
response = self.client.post(
"/api/auth/profile/picture",
{"file": text_file},
**self._auth_headers(tokens["access_token"]),
)
# Assert
self.assertEqual(response.status_code, 400)
def test_upload_profile_picture_rejects_large_files(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
large_content = b"x" * (5 * 1024 * 1024 + 1)
large_file = SimpleUploadedFile("large.png", large_content, content_type="image/png")
# Act
response = self.client.post(
"/api/auth/profile/picture",
{"file": large_file},
**self._auth_headers(tokens["access_token"]),
)
# Assert
self.assertEqual(response.status_code, 400)
def test_delete_profile_picture_removes_file(self):
# Arrange
temp_media = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(temp_media, ignore_errors=True))
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
with override_settings(MEDIA_ROOT=temp_media, MEDIA_URL="/media/"):
image = SimpleUploadedFile("avatar.png", b"data", content_type="image/png")
self.client.post(
"/api/auth/profile/picture",
{"file": image},
**self._auth_headers(tokens["access_token"]),
)
# Act
response = self.client.delete(
"/api/auth/profile/picture", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 200)
user.refresh_from_db()
self.assertFalse(bool(user.profile_picture))
# Password reset ------------------------------------------------------
def test_request_password_reset_enqueues_email(self):
# Arrange
user = self._create_user()
# Act
response = self.client.post(
"/api/auth/request-password-reset",
data=json.dumps({"email": user.email}),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 200)
user.refresh_from_db()
self.assertIsNotNone(user.password_reset_token)
self.mock_password_reset_task.assert_called_once()
def test_request_password_reset_unknown_email_returns_error(self):
# Arrange
payload = {"email": "missing@example.com"}
# Act
response = self.client.post(
"/api/auth/request-password-reset",
data=json.dumps(payload),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 400)
def test_reset_password_confirm_updates_credentials(self):
# Arrange
user = self._create_user()
user.set_password_reset_token()
payload = {"token": str(user.password_reset_token), "new_password": "BrandNewPass!9"}
# Act
response = self.client.post(
"/api/auth/reset-password-confirm",
data=json.dumps(payload),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 200)
user.refresh_from_db()
self.assertIsNone(user.password_reset_token)
self.assertTrue(user.check_password(payload["new_password"]))
def test_reset_password_confirm_rejects_expired_token(self):
# Arrange
user = self._create_user()
user.set_password_reset_token()
user.password_reset_token_expires_at = timezone.now() - timedelta(minutes=1)
user.save(update_fields=["password_reset_token_expires_at"])
payload = {"token": str(user.password_reset_token), "new_password": "New!!!Pass"}
# Act
response = self.client.post(
"/api/auth/reset-password-confirm",
data=json.dumps(payload),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 400)
def test_reset_password_confirm_rejects_unknown_token(self):
# Arrange
payload = {"token": str(uuid.uuid4()), "new_password": "AnotherPass!9"}
# Act
response = self.client.post(
"/api/auth/reset-password-confirm",
data=json.dumps(payload),
content_type="application/json",
)
# Assert
self.assertEqual(response.status_code, 400)
# Admin utilities -----------------------------------------------------
def test_list_deleted_users_requires_privileged_user(self):
# Arrange
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
# Act
response = self.client.get(
"/api/auth/users/deleted", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 403)
def test_list_deleted_users_returns_payload_for_staff(self):
# Arrange
deleted = self._create_user(is_deleted=True, deleted_at=timezone.now())
staff = self._create_user(is_email_verified=True, is_staff=True)
tokens = self._login_and_get_tokens(staff)
# Act
response = self.client.get(
"/api/auth/users/deleted", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 200)
payload = response.json()
self.assertTrue(any(item["id"] == deleted.id for item in payload))
def test_restore_user_requires_privileged_user(self):
# Arrange
target = self._create_user(is_deleted=True, deleted_at=timezone.now())
user = self._create_user(is_email_verified=True)
tokens = self._login_and_get_tokens(user)
# Act
response = self.client.post(
f"/api/auth/users/{target.id}/restore", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 403)
def test_restore_user_restores_record_for_staff(self):
# Arrange
target = self._create_user(is_deleted=True, deleted_at=timezone.now())
staff = self._create_user(is_email_verified=True, is_staff=True)
tokens = self._login_and_get_tokens(staff)
# Act
response = self.client.post(
f"/api/auth/users/{target.id}/restore", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 200)
target.refresh_from_db()
self.assertFalse(target.is_deleted)
def test_restore_user_missing_returns_error(self):
# Arrange
staff = self._create_user(is_email_verified=True, is_staff=True)
tokens = self._login_and_get_tokens(staff)
# Act
response = self.client.post(
"/api/auth/users/999/restore", **self._auth_headers(tokens["access_token"])
)
# Assert
self.assertEqual(response.status_code, 400)
# Username checks ------------------------------------------------------
def test_check_username_reports_existing(self):
# Arrange
user = self._create_user()
# Act
response = self.client.get("/api/auth/check-username", {"username": user.username})
# Assert
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["exists"])
def test_check_username_reports_availability(self):
# Arrange
username = "available_user"
# Act
response = self.client.get("/api/auth/check-username", {"username": username})
# Assert
self.assertEqual(response.status_code, 200)
self.assertFalse(response.json()["exists"])