azionelab/backend/bookings/test_services.py

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)