diff --git a/backend/checkins/services.py b/backend/checkins/services.py new file mode 100644 index 0000000..b28f5bc --- /dev/null +++ b/backend/checkins/services.py @@ -0,0 +1,130 @@ +from dataclasses import dataclass + +from django.db import IntegrityError, transaction + +from bookings.models import Reservation, ReservationToken + +from .models import CheckIn + + +class CheckInServiceError(Exception): + """Base class for check-in service domain errors.""" + + +class InvalidToken(CheckInServiceError): + pass + + +class ReservationNotConfirmed(CheckInServiceError): + pass + + +class AlreadyCheckedIn(CheckInServiceError): + pass + + +class MissingStaffUser(CheckInServiceError): + pass + + +@dataclass(frozen=True) +class CheckInPreview: + reservation_id: int + performance_id: int + show_title: str + venue_name: str + starts_at: object + party_size: int + checked_in: bool + + +@dataclass(frozen=True) +class CheckInResult: + check_in: CheckIn + preview: CheckInPreview + + +def preview_check_in_token(raw_token, *, staff_user): + _validate_staff_user(staff_user) + reservation = _get_reservation_for_check_in_token(raw_token) + _validate_reservation_for_check_in(reservation) + return _build_preview(reservation) + + +def confirm_check_in_from_token(raw_token, *, staff_user, source=CheckIn.Source.QR_SCAN): + _validate_staff_user(staff_user) + + with transaction.atomic(): + reservation = _get_reservation_for_check_in_token(raw_token, lock_token=True) + reservation = ( + Reservation.objects.select_for_update() + .select_related("performance__show", "performance__venue") + .get(pk=reservation.pk) + ) + _validate_reservation_for_check_in(reservation) + + try: + check_in = CheckIn.objects.create( + reservation=reservation, + checked_in_by=staff_user, + source=source, + ) + except IntegrityError as exc: + raise AlreadyCheckedIn("Reservation has already been checked in.") from exc + + return CheckInResult(check_in=check_in, preview=_build_preview(reservation)) + + +def _validate_staff_user(staff_user): + if staff_user is None: + raise MissingStaffUser("A staff user is required for check-in.") + if not getattr(staff_user, "is_authenticated", False): + raise MissingStaffUser("An authenticated staff user is required for check-in.") + if not getattr(staff_user, "is_staff", False): + raise MissingStaffUser("A staff user is required for check-in.") + + +def _get_reservation_for_check_in_token(raw_token, *, lock_token=False): + if not isinstance(raw_token, str) or not raw_token: + raise InvalidToken("Check-in token is invalid.") + + queryset = ReservationToken.objects.select_related( + "reservation__performance__show", + "reservation__performance__venue", + ) + if lock_token: + queryset = queryset.select_for_update() + + try: + token = queryset.get( + token_hash=ReservationToken.hash_token(raw_token), + purpose=ReservationToken.Purpose.CHECK_IN, + used_at__isnull=True, + ) + except ReservationToken.DoesNotExist as exc: + raise InvalidToken("Check-in token is invalid.") from exc + + if token.is_expired: + raise InvalidToken("Check-in token is invalid.") + + return token.reservation + + +def _validate_reservation_for_check_in(reservation): + if reservation.status != Reservation.Status.CONFIRMED: + raise ReservationNotConfirmed("Reservation must be confirmed before check-in.") + if hasattr(reservation, "check_in"): + raise AlreadyCheckedIn("Reservation has already been checked in.") + + +def _build_preview(reservation): + performance = reservation.performance + return CheckInPreview( + reservation_id=reservation.id, + performance_id=performance.id, + show_title=performance.show.title, + venue_name=performance.venue.name, + starts_at=performance.starts_at, + party_size=reservation.party_size, + checked_in=hasattr(reservation, "check_in"), + ) diff --git a/backend/checkins/test_services.py b/backend/checkins/test_services.py new file mode 100644 index 0000000..7f46277 --- /dev/null +++ b/backend/checkins/test_services.py @@ -0,0 +1,134 @@ +from datetime import timedelta + +from django.contrib.auth import get_user_model +from django.test import TestCase +from django.utils import timezone + +from bookings.models import Reservation, ReservationToken +from checkins.models import CheckIn +from checkins.services import ( + AlreadyCheckedIn, + InvalidToken, + MissingStaffUser, + ReservationNotConfirmed, + confirm_check_in_from_token, + preview_check_in_token, +) +from shows.models import Performance, Show, Venue + + +class CheckInServiceTests(TestCase): + def setUp(self): + self.show = Show.objects.create( + title="Open Stage", + slug="open-stage-checkins", + is_published=True, + ) + self.venue = Venue.objects.create( + name="AzioneLab Theatre", + slug="azionelab-theatre-checkins", + 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=20, + ) + self.staff_user = get_user_model().objects.create_user( + username="staff", + password="test", + is_staff=True, + ) + + def test_successful_preview_returns_minimum_admission_data(self): + reservation = self.create_reservation() + _, raw_token = self.create_check_in_token(reservation) + + preview = preview_check_in_token(raw_token, staff_user=self.staff_user) + + self.assertEqual(preview.reservation_id, reservation.id) + self.assertEqual(preview.performance_id, self.performance.id) + self.assertEqual(preview.show_title, self.show.title) + self.assertEqual(preview.venue_name, self.venue.name) + self.assertEqual(preview.party_size, reservation.party_size) + self.assertFalse(preview.checked_in) + self.assertFalse(hasattr(preview, "name")) + self.assertFalse(hasattr(preview, "email")) + self.assertFalse(hasattr(preview, "phone")) + + def test_preview_fails_for_invalid_token(self): + with self.assertRaises(InvalidToken): + preview_check_in_token("invalid-token", staff_user=self.staff_user) + + def test_check_in_succeeds_for_confirmed_reservation(self): + reservation = self.create_reservation() + _, raw_token = self.create_check_in_token(reservation) + + result = confirm_check_in_from_token(raw_token, staff_user=self.staff_user) + + self.assertEqual(result.check_in.reservation, reservation) + self.assertEqual(result.check_in.checked_in_by, self.staff_user) + self.assertEqual(result.check_in.source, CheckIn.Source.QR_SCAN) + self.assertTrue(result.preview.checked_in) + + def test_check_in_fails_for_pending_reservation(self): + reservation = self.create_reservation(status=Reservation.Status.PENDING, confirmed_at=None) + _, raw_token = self.create_check_in_token(reservation) + + with self.assertRaises(ReservationNotConfirmed): + confirm_check_in_from_token(raw_token, staff_user=self.staff_user) + + self.assertFalse(CheckIn.objects.filter(reservation=reservation).exists()) + + def test_duplicate_check_in_fails(self): + reservation = self.create_reservation() + _, raw_token = self.create_check_in_token(reservation) + confirm_check_in_from_token(raw_token, staff_user=self.staff_user) + + with self.assertRaises(AlreadyCheckedIn): + confirm_check_in_from_token(raw_token, staff_user=self.staff_user) + + self.assertEqual(CheckIn.objects.filter(reservation=reservation).count(), 1) + + def test_check_in_stores_timestamp_and_staff_user(self): + reservation = self.create_reservation() + _, raw_token = self.create_check_in_token(reservation) + + before_check_in = timezone.now() + result = confirm_check_in_from_token( + raw_token, + staff_user=self.staff_user, + source=CheckIn.Source.MANUAL, + ) + + self.assertGreaterEqual(result.check_in.checked_in_at, before_check_in) + self.assertEqual(result.check_in.checked_in_by, self.staff_user) + self.assertEqual(result.check_in.source, CheckIn.Source.MANUAL) + + def test_check_in_requires_staff_user(self): + reservation = self.create_reservation() + _, raw_token = self.create_check_in_token(reservation) + + with self.assertRaises(MissingStaffUser): + confirm_check_in_from_token(raw_token, staff_user=None) + + def create_reservation(self, **overrides): + data = { + "performance": self.performance, + "name": "Maria Rossi", + "email": "maria@example.com", + "party_size": 2, + "status": Reservation.Status.CONFIRMED, + "confirmed_at": timezone.now(), + } + data.update(overrides) + return Reservation.objects.create(**data) + + def create_check_in_token(self, reservation): + return ReservationToken.create_token( + reservation=reservation, + purpose=ReservationToken.Purpose.CHECK_IN, + expires_at=self.performance.starts_at + timedelta(days=1), + )