feat: add booking service layer

backend/bookings/test_services.py

@@ -0,0 +1,192 @@
+from datetime import timedelta
+
+from django.db import connection
+from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
+from django.utils import timezone
+
+from bookings.models import Reservation, ReservationToken
+from bookings.services import (
+    AlreadyConfirmedReservation,
+    ExpiredToken,
+    InvalidToken,
+    NotEnoughSeats,
+    calculate_available_seats,
+    confirm_reservation_from_token,
+    create_pending_reservation,
+    generate_confirmation_token,
+)
+from shows.models import Performance, Show, Venue
+
+
+class BookingServiceTests(TestCase):
+    def setUp(self):
+        self.show = Show.objects.create(
+            title="Open Stage",
+            slug="open-stage-services",
+            is_published=True,
+        )
+        self.venue = Venue.objects.create(
+            name="AzioneLab Theatre",
+            slug="azionelab-theatre-services",
+            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,
+        )
+
+    def test_create_pending_reservation_generates_confirmation_token(self):
+        result = create_pending_reservation(
+            performance_id=self.performance.id,
+            name="Maria Rossi",
+            email="maria@example.com",
+            party_size=2,
+        )
+
+        self.assertEqual(result.reservation.status, Reservation.Status.PENDING)
+        self.assertEqual(result.reservation.party_size, 2)
+        self.assertEqual(result.confirmation_token.purpose, ReservationToken.Purpose.CONFIRMATION)
+        self.assertEqual(
+            result.confirmation_token.token_hash,
+            ReservationToken.hash_token(result.raw_confirmation_token),
+        )
+        self.assertNotIn("maria@example.com", result.raw_confirmation_token)
+
+    def test_generate_confirmation_token_returns_raw_token_once(self):
+        reservation = self.create_reservation()
+
+        token, raw_token = generate_confirmation_token(reservation)
+
+        self.assertEqual(token.purpose, ReservationToken.Purpose.CONFIRMATION)
+        self.assertEqual(token.token_hash, ReservationToken.hash_token(raw_token))
+        self.assertGreater(token.expires_at, timezone.now())
+
+    def test_confirm_reservation_from_valid_token(self):
+        reservation = self.create_reservation(party_size=2)
+        token, raw_token = generate_confirmation_token(reservation)
+
+        result = confirm_reservation_from_token(raw_token)
+        reservation.refresh_from_db()
+        token.refresh_from_db()
+
+        self.assertEqual(reservation.status, Reservation.Status.CONFIRMED)
+        self.assertIsNotNone(reservation.confirmed_at)
+        self.assertIsNotNone(reservation.qr_code_generated_at)
+        self.assertIsNotNone(token.used_at)
+        self.assertEqual(result.check_in_token.purpose, ReservationToken.Purpose.CHECK_IN)
+        self.assertEqual(
+            result.check_in_token.token_hash,
+            ReservationToken.hash_token(result.raw_check_in_token),
+        )
+        self.assertEqual(result.available_seats, 1)
+
+    def test_confirmation_fails_when_capacity_is_exhausted(self):
+        Reservation.objects.create(
+            performance=self.performance,
+            name="Confirmed Guest",
+            email="confirmed@example.com",
+            party_size=3,
+            status=Reservation.Status.CONFIRMED,
+            confirmed_at=timezone.now(),
+        )
+        reservation = self.create_reservation(party_size=1)
+        token, raw_token = generate_confirmation_token(reservation)
+
+        with self.assertRaises(NotEnoughSeats):
+            confirm_reservation_from_token(raw_token)
+
+        reservation.refresh_from_db()
+        token.refresh_from_db()
+        self.assertEqual(reservation.status, Reservation.Status.PENDING)
+        self.assertIsNone(token.used_at)
+
+    def test_pending_reservation_does_not_reduce_confirmed_availability(self):
+        create_pending_reservation(
+            performance_id=self.performance.id,
+            name="Pending Guest",
+            email="pending@example.com",
+            party_size=3,
+        )
+
+        self.assertEqual(calculate_available_seats(self.performance), 3)
+
+    def test_invalid_token_fails(self):
+        with self.assertRaises(InvalidToken):
+            confirm_reservation_from_token("not-a-real-token")
+
+    def test_expired_token_marks_pending_reservation_expired(self):
+        reservation = self.create_reservation()
+        token, raw_token = generate_confirmation_token(
+            reservation,
+            expires_at=timezone.now() - timedelta(minutes=1),
+        )
+
+        with self.assertRaises(ExpiredToken):
+            confirm_reservation_from_token(raw_token)
+
+        reservation.refresh_from_db()
+        token.refresh_from_db()
+        self.assertEqual(reservation.status, Reservation.Status.EXPIRED)
+        self.assertIsNone(token.used_at)
+
+    def test_duplicate_confirmation_is_rejected_safely(self):
+        reservation = self.create_reservation()
+        token, raw_token = generate_confirmation_token(reservation)
+        confirm_reservation_from_token(raw_token)
+
+        with self.assertRaises(AlreadyConfirmedReservation):
+            confirm_reservation_from_token(raw_token)
+
+        reservation.refresh_from_db()
+        token.refresh_from_db()
+        self.assertEqual(reservation.status, Reservation.Status.CONFIRMED)
+        self.assertIsNotNone(token.used_at)
+        self.assertEqual(
+            ReservationToken.objects.filter(
+                reservation=reservation,
+                purpose=ReservationToken.Purpose.CHECK_IN,
+            ).count(),
+            1,
+        )
+
+    def test_confirmation_rechecks_capacity_after_another_confirmation(self):
+        self.performance.room_capacity = 1
+        self.performance.save(update_fields=["room_capacity", "updated_at"])
+        first = self.create_reservation(email="first@example.com", party_size=1)
+        second = self.create_reservation(email="second@example.com", party_size=1)
+        _, first_raw_token = generate_confirmation_token(first)
+        second_token, second_raw_token = generate_confirmation_token(second)
+
+        confirm_reservation_from_token(first_raw_token)
+
+        with self.assertRaises(NotEnoughSeats):
+            confirm_reservation_from_token(second_raw_token)
+
+        second.refresh_from_db()
+        second_token.refresh_from_db()
+        self.assertEqual(second.status, Reservation.Status.PENDING)
+        self.assertIsNone(second_token.used_at)
+        self.assertEqual(calculate_available_seats(self.performance), 0)
+
+    def test_confirmation_uses_row_level_locking(self):
+        reservation = self.create_reservation()
+        _, raw_token = generate_confirmation_token(reservation)
+
+        with CaptureQueriesContext(connection) as queries:
+            confirm_reservation_from_token(raw_token)
+
+        self.assertTrue(any("FOR UPDATE" in query["sql"] for query in queries))
+
+    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)