generated from bisco/codex-bootstrap
360 lines
14 KiB
Python
360 lines
14 KiB
Python
from datetime import timedelta
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.cache import cache
|
|
from django.core import mail
|
|
from django.urls import reverse
|
|
from django.test.utils import override_settings
|
|
from django.utils import timezone
|
|
from rest_framework import status
|
|
from rest_framework.test import APIClient, APITestCase
|
|
|
|
from bookings.models import Reservation
|
|
from bookings.services import generate_confirmation_token
|
|
from bookings.views import ReservationConfirmThrottle, ReservationCreateThrottle
|
|
from shows.models import Performance, Show, Venue
|
|
|
|
|
|
class BookingApiTests(APITestCase):
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.show = Show.objects.create(
|
|
title="Open Stage",
|
|
slug="open-stage-api",
|
|
summary="A contemporary theatre performance.",
|
|
description="Full public show description.",
|
|
is_published=True,
|
|
)
|
|
self.hidden_show = Show.objects.create(
|
|
title="Hidden Stage",
|
|
slug="hidden-stage-api",
|
|
is_published=False,
|
|
)
|
|
self.venue = Venue.objects.create(
|
|
name="AzioneLab Theatre",
|
|
slug="azionelab-theatre-api",
|
|
address="Via Example 1",
|
|
city="Rome",
|
|
)
|
|
self.performance = Performance.objects.create(
|
|
show=self.show,
|
|
venue=self.venue,
|
|
starts_at=timezone.now() + timedelta(days=7),
|
|
room_capacity=3,
|
|
)
|
|
Performance.objects.create(
|
|
show=self.hidden_show,
|
|
venue=self.venue,
|
|
starts_at=timezone.now() + timedelta(days=8),
|
|
room_capacity=5,
|
|
)
|
|
|
|
def test_show_list_returns_published_shows(self):
|
|
response = self.client.get(reverse("api-show-list"))
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(len(response.data["results"]), 1)
|
|
self.assertEqual(response.data["results"][0]["slug"], self.show.slug)
|
|
|
|
def test_show_detail_returns_public_performances(self):
|
|
response = self.client.get(reverse("api-show-detail", kwargs={"slug": self.show.slug}))
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["slug"], self.show.slug)
|
|
self.assertEqual(len(response.data["performances"]), 1)
|
|
self.assertEqual(response.data["performances"][0]["id"], self.performance.id)
|
|
self.assertEqual(response.data["performances"][0]["available_seats"], 3)
|
|
|
|
@override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
|
|
def test_reservation_creation_success(self):
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
response = self.client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"phone": "+390600000000",
|
|
"party_size": 2,
|
|
"notes": "Front row if possible.",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data["status"], Reservation.Status.PENDING)
|
|
self.assertEqual(response.data["performance"], self.performance.id)
|
|
self.assertNotIn("token", response.data)
|
|
self.assertNotIn("email", response.data)
|
|
self.assertEqual(Reservation.objects.count(), 1)
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
self.assertIn(
|
|
"https://tickets.azionelab.example/api/reservations/confirm/?token=",
|
|
mail.outbox[0].body,
|
|
)
|
|
|
|
@override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
|
|
def test_reservation_creation_allows_anonymous_post_without_csrf(self):
|
|
csrf_client = APIClient(enforce_csrf_checks=True)
|
|
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
response = csrf_client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 2,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data["status"], Reservation.Status.PENDING)
|
|
|
|
@override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
|
|
def test_reservation_creation_ignores_session_csrf_for_public_endpoint(self):
|
|
csrf_client = APIClient(enforce_csrf_checks=True)
|
|
user = get_user_model().objects.create_user(
|
|
username="box-office",
|
|
email="staff@example.com",
|
|
password="test-pass-123",
|
|
)
|
|
csrf_client.force_login(user)
|
|
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
response = csrf_client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 2,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data["status"], Reservation.Status.PENDING)
|
|
|
|
def test_reservation_creation_schedules_email_after_commit(self):
|
|
with self.captureOnCommitCallbacks(execute=False) as callbacks:
|
|
response = self.client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 2,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(len(callbacks), 1)
|
|
self.assertEqual(len(mail.outbox), 0)
|
|
|
|
def test_reservation_creation_is_throttled(self):
|
|
with patch.dict(ReservationCreateThrottle.THROTTLE_RATES, {"reservation_create": "1/minute"}, clear=False):
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
first_response = self.client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 1,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
second_response = self.client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 1,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(first_response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(second_response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
|
|
|
|
def test_reservation_creation_with_insufficient_seats(self):
|
|
response = self.client.post(
|
|
reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
|
|
{
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 4,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_409_CONFLICT)
|
|
self.assertEqual(response.data["status"], "booking_unavailable")
|
|
self.assertEqual(Reservation.objects.count(), 0)
|
|
|
|
@override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
|
|
def test_confirmation_success(self):
|
|
reservation = self.create_reservation()
|
|
_, raw_token = generate_confirmation_token(reservation)
|
|
|
|
response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": raw_token},
|
|
format="json",
|
|
)
|
|
|
|
reservation.refresh_from_db()
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["reservation_id"], reservation.id)
|
|
self.assertEqual(response.data["status"], Reservation.Status.CONFIRMED)
|
|
self.assertEqual(response.data["party_size"], reservation.party_size)
|
|
self.assertTrue(
|
|
response.data["qr_code_url"].startswith(
|
|
"https://tickets.azionelab.example/check-in?token="
|
|
)
|
|
)
|
|
self.assertNotIn(raw_token, response.data["qr_code_url"])
|
|
self.assertNotIn("token", response.data)
|
|
self.assertTrue(response.data["qr_code_image"].startswith("data:image/png;base64,"))
|
|
self.assertEqual(reservation.status, Reservation.Status.CONFIRMED)
|
|
|
|
def test_confirmation_success_via_email_link_get_request(self):
|
|
reservation = self.create_reservation(email="get@example.com")
|
|
_, raw_token = generate_confirmation_token(reservation)
|
|
|
|
response = self.client.get(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": raw_token},
|
|
)
|
|
|
|
reservation.refresh_from_db()
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["reservation_id"], reservation.id)
|
|
self.assertEqual(reservation.status, Reservation.Status.CONFIRMED)
|
|
|
|
def test_confirmation_with_invalid_token(self):
|
|
response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": "invalid-token"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
|
self.assertEqual(response.data["status"], "invalid_token")
|
|
|
|
def test_duplicate_confirmation_behavior(self):
|
|
reservation = self.create_reservation()
|
|
_, raw_token = generate_confirmation_token(reservation)
|
|
first_response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": raw_token},
|
|
format="json",
|
|
)
|
|
|
|
second_response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": raw_token},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(first_response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(second_response.status_code, status.HTTP_409_CONFLICT)
|
|
self.assertEqual(second_response.data["status"], "already_confirmed")
|
|
|
|
def test_confirmation_is_throttled(self):
|
|
with patch.dict(ReservationConfirmThrottle.THROTTLE_RATES, {"reservation_confirm": "1/minute"}, clear=False):
|
|
first_reservation = self.create_reservation(email="first@example.com")
|
|
_, first_raw_token = generate_confirmation_token(first_reservation)
|
|
second_reservation = self.create_reservation(email="second@example.com")
|
|
_, second_raw_token = generate_confirmation_token(second_reservation)
|
|
|
|
first_response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": first_raw_token},
|
|
format="json",
|
|
)
|
|
second_response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": second_raw_token},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(first_response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(second_response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
|
|
|
|
@override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
|
|
def test_qr_retrieval_success_for_confirmed_reservation(self):
|
|
reservation = self.create_reservation()
|
|
_, raw_token = generate_confirmation_token(reservation)
|
|
confirmation_response = self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": raw_token},
|
|
format="json",
|
|
)
|
|
check_in_token = confirmation_response.data["qr_code_url"].split("token=", 1)[1]
|
|
|
|
response = self.client.get(
|
|
reverse("api-reservation-qr"),
|
|
{"token": check_in_token},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["reservation_id"], reservation.id)
|
|
self.assertTrue(
|
|
response.data["qr_code_url"].startswith(
|
|
"https://tickets.azionelab.example/check-in?token="
|
|
)
|
|
)
|
|
self.assertTrue(response.data["qr_code_image"].startswith("data:image/png;base64,"))
|
|
self.assertNotIn("email", response.data)
|
|
self.assertNotIn("name", response.data)
|
|
|
|
def test_qr_retrieval_rejects_confirmation_token(self):
|
|
reservation = self.create_reservation()
|
|
_, raw_token = generate_confirmation_token(reservation)
|
|
self.client.post(
|
|
reverse("api-reservation-confirm"),
|
|
{"token": raw_token},
|
|
format="json",
|
|
)
|
|
|
|
response = self.client.get(
|
|
reverse("api-reservation-qr"),
|
|
{"token": raw_token},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
|
self.assertEqual(response.data["status"], "invalid_token")
|
|
|
|
def test_qr_retrieval_fails_for_invalid_token(self):
|
|
response = self.client.get(
|
|
reverse("api-reservation-qr"),
|
|
{"token": "invalid-token"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
|
self.assertEqual(response.data["status"], "invalid_token")
|
|
|
|
def test_qr_retrieval_fails_for_pending_reservation(self):
|
|
reservation = self.create_reservation()
|
|
_, raw_token = generate_confirmation_token(reservation)
|
|
|
|
response = self.client.get(
|
|
reverse("api-reservation-qr"),
|
|
{"token": raw_token},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
|
self.assertEqual(response.data["status"], "invalid_token")
|
|
self.assertEqual(reservation.status, Reservation.Status.PENDING)
|
|
|
|
def create_reservation(self, **overrides):
|
|
data = {
|
|
"performance": self.performance,
|
|
"name": "Maria Rossi",
|
|
"email": "maria@example.com",
|
|
"party_size": 1,
|
|
}
|
|
data.update(overrides)
|
|
return Reservation.objects.create(**data)
|