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,
    ReservationNotConfirmed,
    calculate_available_seats,
    confirm_reservation_manually,
    confirm_reservation_from_token,
    create_pending_reservation,
    generate_confirmation_token,
    issue_check_in_access_for_reservation,
    retrieve_reservation_qr_from_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",
        SITE_BASE_URL="https://tickets.azionelab.example",
    )
    def test_create_pending_reservation_sends_confirmation_email_after_commit(self):
        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            result = create_pending_reservation(
                performance_id=self.performance.id,
                name="Maria Rossi",
                email="maria@example.com",
                party_size=1,
            )

        self.assertEqual(len(callbacks), 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(
            "https://tickets.azionelab.example/api/reservations/confirm/?token=",
            mail.outbox[0].body,
        )

    @override_settings(EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend")
    def test_create_pending_reservation_defers_email_until_commit(self):
        with self.captureOnCommitCallbacks(execute=False) as callbacks:
            create_pending_reservation(
                performance_id=self.performance.id,
                name="Maria Rossi",
                email="maria@example.com",
                party_size=1,
            )

        self.assertEqual(len(callbacks), 1)
        self.assertEqual(len(mail.outbox), 0)

    @override_settings(
        EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
        LOG_RESERVATION_CONFIRMATION_URLS=True,
        SITE_BASE_URL="https://tickets.azionelab.example",
    )
    def test_create_pending_reservation_logs_confirmation_link_in_local_mode(self):
        with self.assertLogs("bookings.emailing", level="WARNING") as captured_logs:
            with self.captureOnCommitCallbacks(execute=True):
                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.assertTrue(
            any(
                "LOCAL DEV confirmation URL:" in log_entry
                and result.raw_confirmation_token in log_entry
                for log_entry in captured_logs.output
            )
        )

    @override_settings(LOG_RESERVATION_CONFIRMATION_URLS=False)
    @patch("bookings.emailing.send_mail", side_effect=RuntimeError("SMTP down"))
    def test_create_pending_reservation_logs_email_failure_without_crashing(self, mocked_send_mail):
        with self.assertLogs("bookings.emailing", level="ERROR") as captured_logs:
            with self.captureOnCommitCallbacks(execute=True):
                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()
        self.assertTrue(
            any(
                "Failed to send confirmation email for reservation" in log_entry
                for log_entry in captured_logs.output
            )
        )
        self.assertFalse(
            any(result.raw_confirmation_token in log_entry for log_entry in captured_logs.output)
        )

    @override_settings(
        LOG_RESERVATION_CONFIRMATION_URLS=True,
        SITE_BASE_URL="https://tickets.azionelab.example",
    )
    @patch("bookings.emailing.send_mail", side_effect=RuntimeError("SMTP down"))
    def test_create_pending_reservation_logs_confirmation_link_before_email_failure_in_local_mode(
        self,
        mocked_send_mail,
    ):
        with self.assertLogs("bookings.emailing", level="WARNING") as captured_logs:
            with self.captureOnCommitCallbacks(execute=True):
                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)
        mocked_send_mail.assert_called_once()
        confirmation_log_index = next(
            index
            for index, log_entry in enumerate(captured_logs.output)
            if "LOCAL DEV confirmation URL:" in log_entry
            and result.raw_confirmation_token in log_entry
        )
        failure_log_index = next(
            index
            for index, log_entry in enumerate(captured_logs.output)
            if "Local/debug email delivery failed for reservation" in log_entry
        )

        self.assertLess(confirmation_log_index, failure_log_index)
        self.assertEqual(
            sum(
                result.raw_confirmation_token in log_entry
                for log_entry in captured_logs.output
            ),
            1,
        )
        self.assertFalse(
            any("Traceback" in log_entry for log_entry in captured_logs.output)
        )

    @override_settings(
        LOG_RESERVATION_CONFIRMATION_URLS=False,
        SITE_BASE_URL="https://tickets.azionelab.example",
    )
    @patch("bookings.emailing.send_mail")
    def test_create_pending_reservation_does_not_log_confirmation_link_outside_local_mode(
        self,
        mocked_send_mail,
    ):
        with self.assertNoLogs("bookings.emailing", level="WARNING"):
            with self.captureOnCommitCallbacks(execute=True):
                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)
        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())

    @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
    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_url.startswith(
                "https://tickets.azionelab.example/api/check-ins/preview/?token="
            )
        )
        self.assertTrue(result.qr_code_image.startswith("data:image/png;base64,"))

    @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
    def test_confirmation_token_cannot_be_reused_as_qr_or_check_in_token(self):
        reservation = self.create_reservation()
        _, raw_token = generate_confirmation_token(reservation)

        result = confirm_reservation_from_token(raw_token)

        self.assertNotEqual(raw_token, result.raw_check_in_token)
        self.assertNotEqual(
            build_check_in_preview_url(raw_token),
            result.qr_code_url,
        )

    @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
    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,"))
        self.assertEqual(
            build_check_in_preview_url(raw_check_in_token),
            "https://tickets.azionelab.example/api/check-ins/preview/?token=opaque-check-in-token",
        )

    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_qr_retrieval_rejects_confirmation_token(self):
        reservation = self.create_reservation()
        _, raw_confirmation_token = generate_confirmation_token(reservation)
        confirm_reservation_from_token(raw_confirmation_token)

        with self.assertRaises(InvalidToken):
            retrieve_reservation_qr_from_token(raw_confirmation_token)

    def test_qr_retrieval_accepts_check_in_token(self):
        reservation = self.create_reservation()
        _, raw_confirmation_token = generate_confirmation_token(reservation)
        result = confirm_reservation_from_token(raw_confirmation_token)

        qr_result = retrieve_reservation_qr_from_token(result.raw_check_in_token)

        self.assertEqual(qr_result.reservation, reservation)
        self.assertEqual(
            qr_result.qr_code_url,
            build_check_in_preview_url(result.raw_check_in_token),
        )
        self.assertTrue(qr_result.qr_code_image.startswith("data:image/png;base64,"))

    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))

    @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
    def test_manual_confirmation_reuses_capacity_rules_and_generates_check_in_access(self):
        reservation = self.create_reservation()
        confirmation_token, _ = generate_confirmation_token(reservation)

        result = confirm_reservation_manually(reservation_id=reservation.id)

        reservation.refresh_from_db()
        confirmation_token.refresh_from_db()
        self.assertEqual(reservation.status, Reservation.Status.CONFIRMED)
        self.assertIsNotNone(confirmation_token.used_at)
        self.assertEqual(result.check_in_token.purpose, ReservationToken.Purpose.CHECK_IN)
        self.assertTrue(
            result.qr_code_url.startswith(
                "https://tickets.azionelab.example/api/check-ins/preview/?token="
            )
        )

    def test_manual_confirmation_rejects_non_pending_reservation(self):
        reservation = self.create_reservation(
            status=Reservation.Status.CONFIRMED,
            confirmed_at=timezone.now(),
        )

        with self.assertRaises(AlreadyConfirmedReservation):
            confirm_reservation_manually(reservation_id=reservation.id)

    @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
    def test_issue_check_in_access_requires_confirmed_reservation(self):
        reservation = self.create_reservation()

        with self.assertRaises(ReservationNotConfirmed):
            issue_check_in_access_for_reservation(reservation_id=reservation.id)

    @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
    def test_issue_check_in_access_generates_qr_for_confirmed_reservation(self):
        reservation = self.create_reservation(
            status=Reservation.Status.CONFIRMED,
            confirmed_at=timezone.now(),
        )

        result = issue_check_in_access_for_reservation(reservation_id=reservation.id)

        self.assertEqual(result.reservation, reservation)
        self.assertEqual(result.check_in_token.purpose, ReservationToken.Purpose.CHECK_IN)
        self.assertTrue(result.qr_code_image.startswith("data:image/png;base64,"))
        self.assertIn("/api/check-ins/preview/?token=", result.qr_code_url)

    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)