azionelab/backend/bookings/test_services.py

from datetime import timedelta
from unittest.mock import patch

from django.core import mail
from django.db import connection
from django.test import TestCase
from django.test.utils import CaptureQueriesContext
from django.test.utils import override_settings
from django.utils import timezone

from bookings.models import Reservation, ReservationToken
from bookings.qr import build_check_in_preview_url, generate_check_in_qr_base64
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)

    @override_settings(EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend")
    def test_create_pending_reservation_sends_confirmation_email(self):
        result = create_pending_reservation(
            performance_id=self.performance.id,
            name="Maria Rossi",
            email="maria@example.com",
            party_size=1,
        )

        self.assertEqual(len(mail.outbox), 1)
        self.assertEqual(mail.outbox[0].to, ["maria@example.com"])
        self.assertIn(result.raw_confirmation_token, mail.outbox[0].body)
        self.assertIn("/api/reservations/confirm/?token=", mail.outbox[0].body)

    @patch("bookings.emailing.send_mail", side_effect=RuntimeError("SMTP down"))
    def test_create_pending_reservation_logs_email_failure_without_crashing(self, mocked_send_mail):
        result = create_pending_reservation(
            performance_id=self.performance.id,
            name="Maria Rossi",
            email="maria@example.com",
            party_size=1,
        )

        self.assertEqual(result.reservation.status, Reservation.Status.PENDING)
        self.assertEqual(Reservation.objects.count(), 1)
        mocked_send_mail.assert_called_once()

    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)
        self.assertEqual(
            result.qr_code_url,
            build_check_in_preview_url(result.raw_check_in_token),
        )
        self.assertTrue(result.qr_code_image.startswith("data:image/png;base64,"))

    def test_qr_code_is_generated_for_confirmed_reservation(self):
        reservation = self.create_reservation(
            status=Reservation.Status.CONFIRMED,
            confirmed_at=timezone.now(),
        )
        raw_check_in_token = "opaque-check-in-token"

        qr_code_image = generate_check_in_qr_base64(
            reservation=reservation,
            raw_check_in_token=raw_check_in_token,
        )

        self.assertTrue(qr_code_image.startswith("data:image/png;base64,"))
        self.assertGreater(len(qr_code_image), len("data:image/png;base64,"))

    def test_qr_code_is_not_generated_for_pending_reservation(self):
        reservation = self.create_reservation()

        with self.assertRaisesMessage(
            ValueError,
            "QR codes are available only for confirmed reservations.",
        ):
            generate_check_in_qr_base64(
                reservation=reservation,
                raw_check_in_token="opaque-check-in-token",
            )

    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)