diff --git a/src/django_program/manage/templates/django_program/manage/base.html b/src/django_program/manage/templates/django_program/manage/base.html index d79e2bb..31872e6 100644 --- a/src/django_program/manage/templates/django_program/manage/base.html +++ b/src/django_program/manage/templates/django_program/manage/base.html @@ -1113,6 +1113,11 @@ Check-in +
  • + + Terminal POS + +
    • diff --git a/src/django_program/manage/templates/django_program/manage/terminal_pos.html b/src/django_program/manage/templates/django_program/manage/terminal_pos.html new file mode 100644 index 0000000..4057902 --- /dev/null +++ b/src/django_program/manage/templates/django_program/manage/terminal_pos.html @@ -0,0 +1,1558 @@ +{% extends "django_program/manage/base.html" %} + +{% block title %}Terminal POS{% endblock %} + +{% block extra_head %} + + +{% endblock %} + +{% block breadcrumb %} + +{% endblock %} + +{% block page_title %} +

    Terminal POS

    +

    Process in-person payments with Stripe Terminal readers.

    +{% endblock %} + +{% block page_actions %} +Check-in +{% endblock %} + +{% block content %} +
    {% csrf_token %}
    +
    + +
    +
    +

    Reader

    +
    +
    +
    + + Disconnected +
    + +
    +
    + + +
    + +
    + + +
    +

    Attendee Lookup

    +
    + + +
    +
    +
    +
    +
    + +
    +
    +
    +
    + + +
    +
    +

    Products

    +
    +
    +
    Loading inventory...
    +
    +
    +
    + + +
    +
    +

    Cart

    +
    +
    +
    No items in cart.
    +
      + + +
      + + +
      + + + +
      +
      + + + +
      +
      + Card + -- +
      +
      + Amount + -- +
      +
      + Status + -- +
      +
      + +
      +
      +
      +
      +{% endblock %} + +{% block extra_js %} + +{% endblock %} diff --git a/src/django_program/manage/urls.py b/src/django_program/manage/urls.py index 7c14dfd..a39512b 100644 --- a/src/django_program/manage/urls.py +++ b/src/django_program/manage/urls.py @@ -86,6 +86,7 @@ VoucherListView, ) from django_program.manage.views_checkin import CheckInDashboardView, CheckInScannerView +from django_program.manage.views_terminal import TerminalPOSView app_name = "manage" @@ -276,6 +277,8 @@ # --- Check-in --- path("/checkin/", CheckInDashboardView.as_view(), name="checkin-dashboard"), path("/checkin/scanner/", CheckInScannerView.as_view(), name="checkin-scanner"), + # --- Terminal POS --- + path("/terminal/", TerminalPOSView.as_view(), name="terminal-pos"), # --- Bulk Purchases --- path("/bulk-purchases/", include("django_program.manage.urls_bulk_purchases")), # --- Voucher Bulk Generation --- diff --git a/src/django_program/manage/views_terminal.py b/src/django_program/manage/views_terminal.py new file mode 100644 index 0000000..151b289 --- /dev/null +++ b/src/django_program/manage/views_terminal.py @@ -0,0 +1,33 @@ +"""Stripe Terminal POS views for the manage app. + +Provides the staff-facing Point of Sale interface for processing in-person +payments via Stripe Terminal hardware readers during on-site registration. +""" + +from django.views.generic import TemplateView + +from django_program.manage.views import ManagePermissionMixin + + +class TerminalPOSView(ManagePermissionMixin, TemplateView): + """Staff-facing Point of Sale interface for Stripe Terminal payments. + + Renders the POS template with conference context. All payment processing, + reader management, and cart logic is handled client-side via the terminal + API endpoints and the Stripe Terminal JS SDK. + """ + + template_name = "django_program/manage/terminal_pos.html" + + def get_context_data(self, **kwargs: object) -> dict[str, object]: + """Add active navigation state to the template context. + + Args: + **kwargs: Additional context data. + + Returns: + Template context with conference and active navigation state. + """ + context = super().get_context_data(**kwargs) + context["active_nav"] = "terminal" + return context diff --git a/src/django_program/registration/admin.py b/src/django_program/registration/admin.py index f0bb1de..eae5f51 100644 --- a/src/django_program/registration/admin.py +++ b/src/django_program/registration/admin.py @@ -28,6 +28,7 @@ TicketType, Voucher, ) +from django_program.registration.terminal import TerminalPayment @admin.register(Attendee) @@ -437,3 +438,47 @@ def has_change_permission(self, request: HttpRequest, obj: Badge | None = None) def has_delete_permission(self, request: HttpRequest, obj: Badge | None = None) -> bool: # noqa: ARG002, D102 return False + + +# -- Terminal payment admin --------------------------------------------------- + + +@admin.register(TerminalPayment) +class TerminalPaymentAdmin(admin.ModelAdmin): + """Read-only admin for Stripe Terminal payment records.""" + + list_display = ( + "payment_intent_id", + "conference", + "capture_status", + "card_brand", + "card_last4", + "reader_id", + "created_at", + ) + list_filter = ("conference", "capture_status", "card_brand") + search_fields = ("payment_intent_id", "reader_id", "terminal_id", "card_last4") + readonly_fields = ( + "payment", + "conference", + "terminal_id", + "reader_id", + "payment_intent_id", + "capture_status", + "captured_at", + "cancelled_at", + "card_brand", + "card_last4", + "receipt_url", + "created_at", + "updated_at", + ) + + def has_add_permission(self, request: HttpRequest) -> bool: # noqa: ARG002, D102 + return False + + def has_change_permission(self, request: HttpRequest, obj: TerminalPayment | None = None) -> bool: # noqa: ARG002, D102 + return False + + def has_delete_permission(self, request: HttpRequest, obj: TerminalPayment | None = None) -> bool: # noqa: ARG002, D102 + return False diff --git a/src/django_program/registration/migrations/0018_alter_payment_method_terminalpayment.py b/src/django_program/registration/migrations/0018_alter_payment_method_terminalpayment.py new file mode 100644 index 0000000..604af25 --- /dev/null +++ b/src/django_program/registration/migrations/0018_alter_payment_method_terminalpayment.py @@ -0,0 +1,93 @@ +# Generated by Django 5.2.11 on 2026-03-19 03:24 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("program_conference", "0008_kpitargets"), + ("program_registration", "0017_alter_productredemption_unique_together"), + ] + + operations = [ + migrations.AlterField( + model_name="payment", + name="method", + field=models.CharField( + choices=[ + ("stripe", "Stripe"), + ("terminal", "Terminal"), + ("comp", "Complimentary"), + ("credit", "Credit"), + ("manual", "Manual"), + ], + default="stripe", + max_length=20, + ), + ), + migrations.CreateModel( + name="TerminalPayment", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ( + "terminal_id", + models.CharField(blank=True, default="", help_text="Stripe Terminal Location ID.", max_length=200), + ), + ( + "reader_id", + models.CharField( + blank=True, + default="", + help_text="Stripe Terminal Reader ID that processed the payment.", + max_length=200, + ), + ), + ( + "payment_intent_id", + models.CharField( + help_text="Stripe PaymentIntent ID for this terminal transaction.", max_length=200 + ), + ), + ( + "capture_status", + models.CharField( + choices=[ + ("authorized", "Authorized"), + ("captured", "Captured"), + ("cancelled", "Cancelled"), + ("failed", "Failed"), + ], + default="authorized", + max_length=20, + ), + ), + ("captured_at", models.DateTimeField(blank=True, null=True)), + ("cancelled_at", models.DateTimeField(blank=True, null=True)), + ("card_brand", models.CharField(blank=True, default="", max_length=50)), + ("card_last4", models.CharField(blank=True, default="", max_length=4)), + ("receipt_url", models.URLField(blank=True, default="", max_length=500)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ( + "conference", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="terminal_payments", + to="program_conference.conference", + ), + ), + ( + "payment", + models.OneToOneField( + on_delete=django.db.models.deletion.CASCADE, + related_name="terminal_detail", + to="program_registration.payment", + ), + ), + ], + options={ + "ordering": ["-created_at"], + }, + ), + ] diff --git a/src/django_program/registration/models.py b/src/django_program/registration/models.py index aade958..70eba22 100644 --- a/src/django_program/registration/models.py +++ b/src/django_program/registration/models.py @@ -485,6 +485,7 @@ class Method(models.TextChoices): """Supported payment methods.""" STRIPE = "stripe", "Stripe" + TERMINAL = "terminal", "Terminal" COMP = "comp", "Complimentary" CREDIT = "credit", "Credit" MANUAL = "manual", "Manual" @@ -693,6 +694,7 @@ def __str__(self) -> str: SpeakerCondition, TimeOrStockLimitCondition, ) +from django_program.registration.terminal import TerminalPayment # noqa: E402 __all__ = [ "AddOn", @@ -717,6 +719,7 @@ def __str__(self) -> str: "SpeakerCondition", "StripeCustomer", "StripeEvent", + "TerminalPayment", "TicketType", "TimeOrStockLimitCondition", "Voucher", diff --git a/src/django_program/registration/stripe_client.py b/src/django_program/registration/stripe_client.py index a27478a..aa39d25 100644 --- a/src/django_program/registration/stripe_client.py +++ b/src/django_program/registration/stripe_client.py @@ -198,3 +198,91 @@ def create_refund( params=params, options={"idempotency_key": idempotency_key}, ) + + # -- Terminal methods ----------------------------------------------------- + + def create_connection_token(self) -> str: + """Create a Stripe Terminal connection token for the JS SDK. + + Returns: + The connection token secret string. + """ + token = self.client.v1.terminal.connection_tokens.create() + return token.secret + + def list_readers(self, *, location: str | None = None) -> list[stripe.terminal.Reader]: + """List available Terminal readers, optionally filtered by location. + + Args: + location: Optional Stripe Location ID to filter by. + + Returns: + List of reader objects. + """ + params: dict[str, object] = {"limit": 100} + if location: + params["location"] = location + result = self.client.v1.terminal.readers.list(params=params) + return list(result.data) + + def create_terminal_payment_intent( + self, + *, + amount: Decimal, + currency: str, + customer_id: str | None = None, + metadata: dict[str, str] | None = None, + description: str = "", + ) -> stripe.PaymentIntent: + """Create a PaymentIntent with capture_method=manual for Terminal pre-auth. + + Args: + amount: Payment amount as Decimal. + currency: ISO 4217 currency code. + customer_id: Optional Stripe customer ID. + metadata: Optional metadata dict. + description: Optional description. + + Returns: + The created PaymentIntent object. + """ + api_amount = convert_amount_for_api(amount, currency) + params: dict[str, object] = { + "amount": api_amount, + "currency": currency.lower(), + "capture_method": "manual", + "payment_method_types": ["card_present"], + } + if customer_id: + params["customer"] = customer_id + if metadata: + params["metadata"] = metadata + if description: + params["description"] = description + return self.client.v1.payment_intents.create(params=params) + + def process_terminal_payment(self, *, reader_id: str, payment_intent_id: str) -> stripe.terminal.Reader: + """Send a PaymentIntent to a Terminal reader for card collection. + + Args: + reader_id: The Stripe Terminal Reader ID. + payment_intent_id: The PaymentIntent ID to collect. + + Returns: + The updated Reader object with action status. + """ + return self.client.v1.terminal.readers.process_payment_intent( + reader_id, + params={"payment_intent": payment_intent_id}, + ) + + def cancel_reader_action(self, reader_id: str) -> stripe.terminal.Reader: + """Cancel the current action on a Terminal reader. + + Args: + reader_id: The Stripe Terminal Reader ID. + + Returns: + The updated Reader object. + """ + return self.client.v1.terminal.readers.cancel_action(reader_id) diff --git a/src/django_program/registration/terminal.py b/src/django_program/registration/terminal.py new file mode 100644 index 0000000..51e83a2 --- /dev/null +++ b/src/django_program/registration/terminal.py @@ -0,0 +1,68 @@ +"""Stripe Terminal payment model for in-person POS transactions. + +Records terminal-specific metadata (reader ID, card details, capture lifecycle) +alongside the base Payment record created during checkout at the registration desk. +""" + +from django.db import models + + +class TerminalPayment(models.Model): + """Records an in-person Stripe Terminal payment at the registration desk. + + Extends the payment tracking with terminal-specific fields for reader + identification, pre-authorization state, and capture lifecycle. + """ + + class CaptureStatus(models.TextChoices): + """Lifecycle states for a terminal payment capture.""" + + AUTHORIZED = "authorized", "Authorized" + CAPTURED = "captured", "Captured" + CANCELLED = "cancelled", "Cancelled" + FAILED = "failed", "Failed" + + payment = models.OneToOneField( + "program_registration.Payment", + on_delete=models.CASCADE, + related_name="terminal_detail", + ) + conference = models.ForeignKey( + "program_conference.Conference", + on_delete=models.CASCADE, + related_name="terminal_payments", + ) + terminal_id = models.CharField( + max_length=200, + blank=True, + default="", + help_text="Stripe Terminal Location ID.", + ) + reader_id = models.CharField( + max_length=200, + blank=True, + default="", + help_text="Stripe Terminal Reader ID that processed the payment.", + ) + payment_intent_id = models.CharField( + max_length=200, + help_text="Stripe PaymentIntent ID for this terminal transaction.", + ) + capture_status = models.CharField( + max_length=20, + choices=CaptureStatus.choices, + default=CaptureStatus.AUTHORIZED, + ) + captured_at = models.DateTimeField(null=True, blank=True) + cancelled_at = models.DateTimeField(null=True, blank=True) + card_brand = models.CharField(max_length=50, blank=True, default="") + card_last4 = models.CharField(max_length=4, blank=True, default="") + receipt_url = models.URLField(max_length=500, blank=True, default="") + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + ordering = ["-created_at"] + + def __str__(self) -> str: + return f"Terminal {self.payment_intent_id} ({self.capture_status})" diff --git a/src/django_program/registration/urls.py b/src/django_program/registration/urls.py index 9cac3dc..c41318b 100644 --- a/src/django_program/registration/urls.py +++ b/src/django_program/registration/urls.py @@ -27,6 +27,16 @@ RedeemView, ScanView, ) +from django_program.registration.views_terminal import ( + CancelPaymentView, + CapturePaymentView, + CartOperationsView, + ConnectionTokenView, + CreatePaymentIntentView, + FetchAttendeeView, + FetchInventoryView, + ListReadersView, +) from django_program.registration.webhooks import stripe_webhook app_name = "registration" @@ -43,4 +53,13 @@ path("checkin/lookup//", LookupView.as_view(), name="checkin-lookup"), path("checkin/redeem/", RedeemView.as_view(), name="checkin-redeem"), path("checkin/preload/", OfflinePreloadView.as_view(), name="checkin-preload"), + # Stripe Terminal API (staff-only, JSON endpoints for POS UI) + path("terminal/connection-token/", ConnectionTokenView.as_view(), name="terminal-connection-token"), + path("terminal/create-payment-intent/", CreatePaymentIntentView.as_view(), name="terminal-create-intent"), + path("terminal/capture/", CapturePaymentView.as_view(), name="terminal-capture"), + path("terminal/cancel/", CancelPaymentView.as_view(), name="terminal-cancel"), + path("terminal/attendee//", FetchAttendeeView.as_view(), name="terminal-attendee"), + path("terminal/inventory/", FetchInventoryView.as_view(), name="terminal-inventory"), + path("terminal/cart/", CartOperationsView.as_view(), name="terminal-cart"), + path("terminal/readers/", ListReadersView.as_view(), name="terminal-readers"), ] diff --git a/src/django_program/registration/views_checkin.py b/src/django_program/registration/views_checkin.py index 6970265..e2ce30f 100644 --- a/src/django_program/registration/views_checkin.py +++ b/src/django_program/registration/views_checkin.py @@ -103,7 +103,7 @@ def _parse_json_body(request: HttpRequest) -> dict[str, object] | None: """ try: payload = json.loads(request.body) - except (json.JSONDecodeError, ValueError): + except json.JSONDecodeError, ValueError: return None if not isinstance(payload, dict): return None @@ -391,7 +391,9 @@ def _serialize_preload_attendee(attendee: Attendee) -> dict[str, object]: # Use prefetched redemptions to avoid N+1 queries redeemed_counts: dict[int, int] = {} for redemption in attendee.redemptions.all(): - redeemed_counts[redemption.order_line_item_id] = redeemed_counts.get(redemption.order_line_item_id, 0) + 1 + redeemed_counts[redemption.order_line_item_id] = ( + redeemed_counts.get(redemption.order_line_item_id, 0) + 1 + ) for item in order.line_items.all(): product_data = _serialize_line_item(item) product_data["redeemed_count"] = redeemed_counts.get(item.pk, 0) diff --git a/src/django_program/registration/views_terminal.py b/src/django_program/registration/views_terminal.py new file mode 100644 index 0000000..37f1736 --- /dev/null +++ b/src/django_program/registration/views_terminal.py @@ -0,0 +1,799 @@ +"""Staff-facing JSON API views for Stripe Terminal point-of-sale operations. + +These views power the on-site POS UI used by registration desk staff. +All endpoints require staff or superuser authentication and are scoped +to a conference via the ``conference_slug`` URL kwarg. +""" + +import json +import logging +import secrets +import string +from decimal import Decimal, InvalidOperation +from typing import TYPE_CHECKING + +import stripe +from django.db import models, transaction +from django.http import JsonResponse +from django.utils import timezone +from django.views import View + +from django_program.registration.attendee import Attendee +from django_program.registration.models import ( + AddOn, + Cart, + CartItem, + Credit, + Order, + OrderLineItem, + Payment, + TicketType, +) +from django_program.registration.services.checkin import CheckInService +from django_program.registration.stripe_client import StripeClient +from django_program.registration.terminal import TerminalPayment +from django_program.registration.views_checkin import StaffRequiredMixin +from django_program.settings import get_config + +if TYPE_CHECKING: + from django.http import HttpRequest + +logger = logging.getLogger(__name__) + + +def _parse_json_body(request: HttpRequest) -> dict[str, object] | None: + """Parse JSON from request body, returning None on failure.""" + try: + return json.loads(request.body) # type: ignore[no-any-return] + except json.JSONDecodeError, ValueError: + return None + + +def _generate_order_reference() -> str: + """Generate a unique order reference like ``ORD-A1B2C3``.""" + chars = string.ascii_uppercase + string.digits + suffix = "".join(secrets.choice(chars) for _ in range(6)) + return f"ORD-{suffix}" + + +def _stripe_error_response(exc: stripe.StripeError) -> JsonResponse: + """Build a consistent JSON error response from a Stripe exception.""" + logger.warning("Stripe error: %s", exc) + return JsonResponse( + {"error": "Payment processing error. Please try again."}, + status=502, + ) + + +class ConnectionTokenView(StaffRequiredMixin, View): + """Create a Stripe Terminal connection token for the JS SDK. + + The frontend calls this endpoint to initialize the Stripe Terminal + SDK with a short-lived connection token scoped to the conference's + Stripe account. + """ + + def post(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Return a new connection token secret. + + Args: + request: The incoming HTTP request. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with ``secret`` on success, or an error payload. + """ + try: + client = StripeClient(self.conference) + secret = client.create_connection_token() + except ValueError as exc: + return JsonResponse({"error": str(exc)}, status=400) + except stripe.StripeError as exc: + return _stripe_error_response(exc) + + return JsonResponse({"secret": secret}) + + +class CreatePaymentIntentView(StaffRequiredMixin, View): + """Create a PaymentIntent and send it to a Stripe Terminal reader. + + Supports two modes: + - Order-linked: provide ``order_id`` to pay for an existing order. + - Walk-up sale: provide ``amount`` directly for ad-hoc charges. + + Both modes require a ``reader_id`` to dispatch the payment to a + physical reader. + """ + + def post(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Create a terminal PaymentIntent and dispatch to a reader. + + Args: + request: The incoming HTTP request with JSON body. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with intent details and reader action on success. + """ + body = _parse_json_body(request) + if body is None: + return JsonResponse({"error": "Invalid JSON body"}, status=400) + + reader_id = str(body.get("reader_id", "")).strip() + if not reader_id: + return JsonResponse({"error": "reader_id is required"}, status=400) + + result = self._resolve_order_and_amount(body) + if isinstance(result, JsonResponse): + return result + order, amount = result + + return self._create_and_dispatch(request, reader_id, order, amount) + + def _resolve_order_and_amount(self, body: dict[str, object]) -> tuple[Order | None, Decimal] | JsonResponse: + """Resolve the order and amount from the request body.""" + order_id = body.get("order_id") + raw_amount = body.get("amount") + + if not order_id and not raw_amount: + return JsonResponse({"error": "Either order_id or amount is required"}, status=400) + + if order_id: + try: + order = Order.objects.get(pk=int(order_id), conference=self.conference) # type: ignore[arg-type] + except Order.DoesNotExist, TypeError, ValueError: + return JsonResponse({"error": "Order not found"}, status=404) + return order, order.total + + try: + amount = Decimal(str(raw_amount)) + if amount <= 0: + return JsonResponse({"error": "Amount must be positive"}, status=400) + except InvalidOperation, TypeError: + return JsonResponse({"error": "Invalid amount"}, status=400) + return None, amount + + def _create_and_dispatch( + self, + request: HttpRequest, + reader_id: str, + order: Order | None, + amount: Decimal, + ) -> JsonResponse: + """Create PaymentIntent, dispatch to reader, and record in DB.""" + try: + client = StripeClient(self.conference) + except ValueError as exc: + return JsonResponse({"error": str(exc)}, status=400) + + config = get_config() + currency = config.currency + + metadata: dict[str, str] = { + "conference_id": str(self.conference.pk), + "conference_slug": str(self.conference.slug), + "terminal": "true", + "staff_user_id": str(request.user.pk), + } + description = f"Terminal payment for {self.conference.name}" + if order is not None: + metadata["order_id"] = str(order.pk) + metadata["reference"] = str(order.reference) + description = f"Order {order.reference} (terminal) for {self.conference.name}" + + try: + intent = client.create_terminal_payment_intent( + amount=amount, + currency=currency.lower(), + metadata=metadata, + description=description, + ) + reader_result = client.process_terminal_payment( + reader_id=reader_id, + payment_intent_id=intent.id, + ) + except stripe.StripeError as exc: + return _stripe_error_response(exc) + + with transaction.atomic(): + if order is None: + order = Order.objects.create( + conference=self.conference, + user=request.user, + status=Order.Status.PENDING, + subtotal=amount, + total=amount, + billing_name=str(getattr(request.user, "get_full_name", lambda: "")()), + billing_email=str(getattr(request.user, "email", "")), + reference=_generate_order_reference(), + ) + + payment = Payment.objects.create( + order=order, + method=Payment.Method.TERMINAL, + status=Payment.Status.PENDING, + amount=amount, + stripe_payment_intent_id=intent.id, + created_by=request.user, + ) + + TerminalPayment.objects.create( + payment=payment, + conference=self.conference, + reader_id=reader_id, + payment_intent_id=intent.id, + capture_status=TerminalPayment.CaptureStatus.AUTHORIZED, + ) + + reader_action = {} + if hasattr(reader_result, "action") and reader_result.action: + action = reader_result.action + reader_action = { + "status": getattr(action, "status", None), + "type": getattr(action, "type", None), + } + + return JsonResponse( + { + "payment_intent_id": intent.id, + "status": "processing", + "order_id": order.pk, + "order_reference": str(order.reference), + "client_secret": intent.client_secret, + "reader_action": reader_action, + } + ) + + +class CapturePaymentView(StaffRequiredMixin, View): + """Capture a previously authorized terminal PaymentIntent. + + After the cardholder taps/inserts their card and the payment is + authorized, this endpoint captures the funds and marks the order + as paid. + """ + + def post(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Capture the authorized PaymentIntent. + + Args: + request: The incoming HTTP request with JSON body. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with capture status on success. + """ + body = _parse_json_body(request) + if body is None: + return JsonResponse({"error": "Invalid JSON body"}, status=400) + + payment_intent_id = str(body.get("payment_intent_id", "")).strip() + if not payment_intent_id: + return JsonResponse({"error": "payment_intent_id is required"}, status=400) + + try: + terminal_payment = TerminalPayment.objects.select_related( + "payment__order", + ).get(payment_intent_id=payment_intent_id, conference=self.conference) + except TerminalPayment.DoesNotExist: + return JsonResponse({"error": "Terminal payment not found"}, status=404) + + try: + client = StripeClient(self.conference) + client.capture_payment_intent(payment_intent_id) + except ValueError as exc: + return JsonResponse({"error": str(exc)}, status=400) + except stripe.StripeError as exc: + return _stripe_error_response(exc) + + now = timezone.now() + with transaction.atomic(): + terminal_payment.capture_status = TerminalPayment.CaptureStatus.CAPTURED + terminal_payment.captured_at = now + terminal_payment.save(update_fields=["capture_status", "captured_at", "updated_at"]) + + payment = terminal_payment.payment + payment.status = Payment.Status.SUCCEEDED + payment.save(update_fields=["status"]) + + order = payment.order + if order.status == Order.Status.PENDING: + order.status = Order.Status.PAID + order.save(update_fields=["status", "updated_at"]) + + return JsonResponse( + { + "status": "captured", + "payment_intent_id": payment_intent_id, + "order_id": order.pk, + "order_reference": str(order.reference), + } + ) + + +class CancelPaymentView(StaffRequiredMixin, View): + """Cancel a terminal payment in progress. + + Cancels the reader action and the PaymentIntent, then marks the + local records as failed/cancelled. + """ + + def post(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Cancel the terminal payment and reader action. + + Args: + request: The incoming HTTP request with JSON body. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with cancellation status on success. + """ + body = _parse_json_body(request) + if body is None: + return JsonResponse({"error": "Invalid JSON body"}, status=400) + + payment_intent_id = str(body.get("payment_intent_id", "")).strip() + reader_id = str(body.get("reader_id", "")).strip() + + if not payment_intent_id: + return JsonResponse({"error": "payment_intent_id is required"}, status=400) + + try: + terminal_payment = TerminalPayment.objects.select_related( + "payment__order", + ).get(payment_intent_id=payment_intent_id, conference=self.conference) + except TerminalPayment.DoesNotExist: + return JsonResponse({"error": "Terminal payment not found"}, status=404) + + try: + client = StripeClient(self.conference) + except ValueError as exc: + return JsonResponse({"error": str(exc)}, status=400) + + try: + if reader_id: + client.cancel_reader_action(reader_id) + client.client.v1.payment_intents.cancel(payment_intent_id) + except stripe.StripeError as exc: + return _stripe_error_response(exc) + + now = timezone.now() + with transaction.atomic(): + terminal_payment.capture_status = TerminalPayment.CaptureStatus.CANCELLED + terminal_payment.cancelled_at = now + terminal_payment.save(update_fields=["capture_status", "cancelled_at", "updated_at"]) + + payment = terminal_payment.payment + payment.status = Payment.Status.FAILED + payment.save(update_fields=["status"]) + + return JsonResponse( + { + "status": "cancelled", + "payment_intent_id": payment_intent_id, + } + ) + + +class FetchAttendeeView(StaffRequiredMixin, View): + """Look up an attendee by access code for the POS terminal. + + Returns attendee info, order status, and available store credits + so the registration desk can identify walk-up attendees and apply + credits to new purchases. + """ + + def get(self, request: HttpRequest, access_code: str, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Return attendee details by access code. + + Args: + request: The incoming HTTP request. + access_code: The attendee's unique access code from the URL. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with attendee info, order status, and available credits. + """ + try: + attendee = CheckInService.lookup_attendee( + conference=self.conference, + access_code=access_code.strip(), + ) + except Attendee.DoesNotExist: + return JsonResponse( + {"error": "Attendee not found", "access_code": access_code}, + status=404, + ) + + order = attendee.order + order_data: dict[str, object] | None = None + if order is not None: + line_items = [ + { + "id": item.pk, + "description": str(item.description), + "quantity": item.quantity, + "unit_price": str(item.unit_price), + "line_total": str(item.line_total), + "ticket_type_slug": str(item.ticket_type.slug) if item.ticket_type else None, + "addon_slug": str(item.addon.slug) if item.addon else None, + } + for item in order.line_items.all() + ] + order_data = { + "id": order.pk, + "reference": str(order.reference), + "status": str(order.status), + "total": str(order.total), + "line_items": line_items, + } + + available_credits = Credit.objects.filter( + user=attendee.user, + conference=self.conference, + status=Credit.Status.AVAILABLE, + ) + credits_total = sum(c.remaining_amount for c in available_credits) + credits_data = [ + { + "id": c.pk, + "amount": str(c.amount), + "remaining_amount": str(c.remaining_amount), + } + for c in available_credits + ] + + return JsonResponse( + { + "attendee": { + "id": attendee.pk, + "access_code": str(attendee.access_code), + "name": str(getattr(attendee.user, "get_full_name", lambda: "")()), + "email": str(getattr(attendee.user, "email", "")), + "checked_in": attendee.checked_in_at is not None, + "checked_in_at": (attendee.checked_in_at.isoformat() if attendee.checked_in_at else None), + }, + "order": order_data, + "credits": credits_data, + "credits_total": str(credits_total), + "registered": order is not None + and order.status + in { + Order.Status.PAID, + Order.Status.PARTIALLY_REFUNDED, + }, + } + ) + + +class FetchInventoryView(StaffRequiredMixin, View): + """Return available ticket types and add-ons for the conference. + + Filters to active items currently within their availability window + and includes remaining quantity information for the POS display. + """ + + def get(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Return the conference inventory. + + Args: + request: The incoming HTTP request. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with ticket_types and addons arrays. + """ + now = timezone.now() + + ticket_types = ( + TicketType.objects.filter( + conference=self.conference, + is_active=True, + ) + .filter( + models.Q(available_from__isnull=True) | models.Q(available_from__lte=now), + ) + .filter( + models.Q(available_until__isnull=True) | models.Q(available_until__gte=now), + ) + ) + + addons = ( + AddOn.objects.filter( + conference=self.conference, + is_active=True, + ) + .filter( + models.Q(available_from__isnull=True) | models.Q(available_from__lte=now), + ) + .filter( + models.Q(available_until__isnull=True) | models.Q(available_until__gte=now), + ) + ) + + return JsonResponse( + { + "ticket_types": [ + { + "id": tt.pk, + "name": str(tt.name), + "slug": str(tt.slug), + "description": str(tt.description), + "price": str(tt.price), + "remaining_quantity": tt.remaining_quantity, + "requires_voucher": tt.requires_voucher, + } + for tt in ticket_types + ], + "addons": [ + { + "id": addon.pk, + "name": str(addon.name), + "slug": str(addon.slug), + "description": str(addon.description), + "price": str(addon.price), + } + for addon in addons + ], + } + ) + + +class CartOperationsView(StaffRequiredMixin, View): + """Handle cart operations for the terminal POS. + + Supports two actions via JSON body: + - ``update``: create or update cart items for an attendee. + - ``checkout``: convert the cart into an order. + """ + + def post(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Dispatch cart action (update or checkout). + + Args: + request: The incoming HTTP request with JSON body. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with cart or order details on success. + """ + body = _parse_json_body(request) + if body is None: + return JsonResponse({"error": "Invalid JSON body"}, status=400) + + action = str(body.get("action", "")).strip() + if action not in {"update", "checkout"}: + return JsonResponse({"error": "action must be 'update' or 'checkout'"}, status=400) + + if action == "update": + return self._handle_update(request, body) + return self._handle_checkout(request, body) + + def _handle_update(self, request: HttpRequest, body: dict[str, object]) -> JsonResponse: + """Create or update cart items for an attendee.""" + access_code = str(body.get("attendee_access_code", "")).strip() + items = body.get("items", []) + + if not isinstance(items, list): + return JsonResponse({"error": "items must be a list"}, status=400) + + attendee: Attendee | None = None + if access_code: + try: + attendee = CheckInService.lookup_attendee( + conference=self.conference, + access_code=access_code, + ) + except Attendee.DoesNotExist: + return JsonResponse( + {"error": "Attendee not found", "access_code": access_code}, + status=404, + ) + + cart_user = attendee.user if attendee else request.user + + with transaction.atomic(): + cart, _created = Cart.objects.get_or_create( + user=cart_user, + conference=self.conference, + status=Cart.Status.OPEN, + ) + + cart.items.all().delete() + + cart_total = Decimal("0.00") + cart_items: list[dict[str, object]] = [] + for item_data in items: + result = self._add_cart_item(cart, item_data) + if isinstance(result, JsonResponse): + return result + if result is not None: + cart_total += Decimal(str(result["line_total"])) + cart_items.append(result) + + return JsonResponse( + { + "cart_id": cart.pk, + "items": cart_items, + "total": str(cart_total), + } + ) + + def _add_cart_item( + self, cart: Cart, item_data: object + ) -> dict[str, object] | JsonResponse | None: + """Process a single cart item from the request payload.""" + if not isinstance(item_data, dict): + return None + ticket_type_id = item_data.get("ticket_type_id") + addon_id = item_data.get("addon_id") + try: + quantity = int(item_data.get("quantity", 1)) # type: ignore[arg-type] + except (TypeError, ValueError): + return JsonResponse({"error": "Invalid quantity value"}, status=400) + + if ticket_type_id: + return self._add_ticket_item(cart, ticket_type_id, quantity) + if addon_id: + return self._add_addon_item(cart, addon_id, quantity) + return None + + def _add_ticket_item( + self, cart: Cart, ticket_type_id: object, quantity: int + ) -> dict[str, object] | None: + """Create a ticket CartItem and return its data.""" + try: + tt = TicketType.objects.get(pk=int(ticket_type_id), conference=self.conference) # type: ignore[arg-type] + except (TicketType.DoesNotExist, TypeError, ValueError): + return None + ci = CartItem.objects.create(cart=cart, ticket_type=tt, quantity=quantity) + line_total = tt.price * quantity + return { + "id": ci.pk, "ticket_type_id": tt.pk, "name": str(tt.name), + "quantity": quantity, "unit_price": str(tt.price), "line_total": str(line_total), + } + + def _add_addon_item( + self, cart: Cart, addon_id: object, quantity: int + ) -> dict[str, object] | None: + """Create an addon CartItem and return its data.""" + try: + addon = AddOn.objects.get(pk=int(addon_id), conference=self.conference) # type: ignore[arg-type] + except (AddOn.DoesNotExist, TypeError, ValueError): + return None + ci = CartItem.objects.create(cart=cart, addon=addon, quantity=quantity) + line_total = addon.price * quantity + return { + "id": ci.pk, "addon_id": addon.pk, "name": str(addon.name), + "quantity": quantity, "unit_price": str(addon.price), "line_total": str(line_total), + } + + def _handle_checkout(self, request: HttpRequest, body: dict[str, object]) -> JsonResponse: + """Convert the cart into a pending order.""" + access_code = str(body.get("attendee_access_code", "")).strip() + billing_name = str(body.get("billing_name", "")).strip() + billing_email = str(body.get("billing_email", "")).strip() + + attendee: Attendee | None = None + if access_code: + try: + attendee = CheckInService.lookup_attendee( + conference=self.conference, + access_code=access_code, + ) + except Attendee.DoesNotExist: + return JsonResponse( + {"error": "Attendee not found", "access_code": access_code}, + status=404, + ) + + cart_user = attendee.user if attendee else request.user + + try: + cart = Cart.objects.prefetch_related( + "items__ticket_type", + "items__addon", + ).get( + user=cart_user, + conference=self.conference, + status=Cart.Status.OPEN, + ) + except Cart.DoesNotExist: + return JsonResponse({"error": "No open cart found"}, status=404) + + cart_items = list(cart.items.all()) + if not cart_items: + return JsonResponse({"error": "Cart is empty"}, status=400) + + with transaction.atomic(): + subtotal = sum(item.line_total for item in cart_items) + order = Order.objects.create( + conference=self.conference, + user=cart_user, + status=Order.Status.PENDING, + subtotal=subtotal, + total=subtotal, + billing_name=billing_name or str(getattr(cart_user, "get_full_name", lambda: "")()), + billing_email=billing_email or str(getattr(cart_user, "email", "")), + reference=_generate_order_reference(), + ) + + for item in cart_items: + description = str(item.ticket_type.name if item.ticket_type else item.addon.name if item.addon else "") + OrderLineItem.objects.create( + order=order, + description=description, + quantity=item.quantity, + unit_price=item.unit_price, + line_total=item.line_total, + ticket_type=item.ticket_type, + addon=item.addon, + ) + + cart.status = Cart.Status.CHECKED_OUT + cart.save(update_fields=["status", "updated_at"]) + + line_items = [ + { + "id": li.pk, + "description": str(li.description), + "quantity": li.quantity, + "unit_price": str(li.unit_price), + "line_total": str(li.line_total), + } + for li in order.line_items.all() + ] + + return JsonResponse( + { + "order_id": order.pk, + "reference": str(order.reference), + "status": str(order.status), + "total": str(order.total), + "line_items": line_items, + } + ) + + +class ListReadersView(StaffRequiredMixin, View): + """List available Stripe Terminal readers for the conference. + + Optionally filtered by Stripe location ID via the ``location`` + query parameter. + """ + + def get(self, request: HttpRequest, **kwargs: str) -> JsonResponse: # noqa: ARG002 + """Return the list of terminal readers. + + Args: + request: The incoming HTTP request. + **kwargs: URL keyword arguments (unused). + + Returns: + JSON with a ``readers`` array. + """ + location = request.GET.get("location", "").strip() or None + + try: + client = StripeClient(self.conference) + readers = client.list_readers(location=location) + except ValueError as exc: + return JsonResponse({"error": str(exc)}, status=400) + except stripe.StripeError as exc: + return _stripe_error_response(exc) + + return JsonResponse( + { + "readers": [ + { + "id": getattr(r, "id", None), + "label": getattr(r, "label", ""), + "status": getattr(r, "status", ""), + "device_type": getattr(r, "device_type", ""), + "location": getattr(r, "location", None), + "serial_number": getattr(r, "serial_number", ""), + } + for r in readers + ], + } + ) diff --git a/tests/test_registration/test_terminal.py b/tests/test_registration/test_terminal.py new file mode 100644 index 0000000..ea3946a --- /dev/null +++ b/tests/test_registration/test_terminal.py @@ -0,0 +1,568 @@ +"""Tests for Stripe Terminal POS payments and on-site registration views.""" + +import json +from datetime import date +from decimal import Decimal +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +import pytest +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Permission +from django.test import Client +from django.urls import reverse + +from django_program.conference.models import Conference +from django_program.registration.models import AddOn, Order, Payment, TicketType +from django_program.registration.terminal import TerminalPayment + +User = get_user_model() + +pytestmark = pytest.mark.django_db + + +# -- Helpers ------------------------------------------------------------------ + + +def _make_conference(**kwargs: object) -> Conference: + defaults: dict[str, object] = { + "name": "TestCon", + "slug": f"testcon-{uuid4().hex[:6]}", + "start_date": date(2026, 7, 1), + "end_date": date(2026, 7, 5), + "stripe_secret_key": "sk_test_fake123", + } + defaults.update(kwargs) + return Conference.objects.create(**defaults) + + +def _make_user(**kwargs: object) -> object: + defaults: dict[str, object] = { + "username": f"user-{uuid4().hex[:8]}", + "email": f"{uuid4().hex[:8]}@test.com", + } + defaults.update(kwargs) + return User.objects.create_user(**defaults) + + +def _make_staff_user(**kwargs: object) -> object: + """Create a staff user with the change_conference permission required by terminal API.""" + defaults: dict[str, object] = { + "username": f"staff-{uuid4().hex[:8]}", + "email": f"{uuid4().hex[:8]}@test.com", + "is_staff": True, + "password": "testpass123", + } + defaults.update(kwargs) + user = User.objects.create_user(**defaults) + perm = Permission.objects.get(content_type__app_label="program_conference", codename="change_conference") + user.user_permissions.add(perm) + return user + + +def _make_order(*, conference: Conference, user: object, status: str = Order.Status.PAID) -> Order: + return Order.objects.create( + conference=conference, + user=user, + status=status, + subtotal=Decimal("100.00"), + total=Decimal("100.00"), + reference=f"ORD-{uuid4().hex[:8].upper()}", + ) + + +def _make_payment(*, order: Order, method: str = Payment.Method.TERMINAL, **kwargs: object) -> Payment: + defaults: dict[str, object] = { + "order": order, + "method": method, + "amount": order.total, + "stripe_payment_intent_id": f"pi_{uuid4().hex[:16]}", + } + defaults.update(kwargs) + return Payment.objects.create(**defaults) + + +def _make_ticket_type(*, conference: Conference, **kwargs: object) -> TicketType: + defaults: dict[str, object] = { + "name": "General Admission", + "slug": f"general-{uuid4().hex[:6]}", + "price": Decimal("100.00"), + "is_active": True, + } + defaults.update(kwargs) + return TicketType.objects.create(conference=conference, **defaults) + + +def _make_addon(*, conference: Conference, **kwargs: object) -> AddOn: + defaults: dict[str, object] = { + "name": "Workshop", + "slug": f"workshop-{uuid4().hex[:6]}", + "price": Decimal("50.00"), + "is_active": True, + } + defaults.update(kwargs) + return AddOn.objects.create(conference=conference, **defaults) + + +def _make_attendee(*, conference: Conference, user: object, order: Order | None = None) -> object: + from django_program.registration.attendee import Attendee + + return Attendee.objects.create(user=user, conference=conference, order=order) + + +# -- Model Tests -------------------------------------------------------------- + + +@pytest.mark.unit +class TestTerminalPaymentModel: + """Tests for the TerminalPayment model.""" + + def test_terminal_payment_creation(self) -> None: + conf = _make_conference() + user = _make_user() + order = _make_order(conference=conf, user=user) + payment = _make_payment(order=order) + + terminal_payment = TerminalPayment.objects.create( + payment=payment, + conference=conf, + payment_intent_id=f"pi_{uuid4().hex[:16]}", + terminal_id="tml_test123", + reader_id="tmr_test456", + card_brand="visa", + card_last4="4242", + ) + + assert terminal_payment.pk is not None + assert terminal_payment.payment == payment + assert terminal_payment.conference == conf + assert terminal_payment.terminal_id == "tml_test123" + assert terminal_payment.reader_id == "tmr_test456" + assert terminal_payment.card_brand == "visa" + assert terminal_payment.card_last4 == "4242" + assert terminal_payment.captured_at is None + assert terminal_payment.cancelled_at is None + + def test_terminal_payment_str(self) -> None: + conf = _make_conference() + user = _make_user() + order = _make_order(conference=conf, user=user) + payment = _make_payment(order=order) + + terminal_payment = TerminalPayment.objects.create( + payment=payment, + conference=conf, + payment_intent_id="pi_abc123", + ) + + result = str(terminal_payment) + assert "pi_abc123" in result + assert "authorized" in result.lower() + + def test_terminal_payment_default_capture_status_is_authorized(self) -> None: + conf = _make_conference() + user = _make_user() + order = _make_order(conference=conf, user=user) + payment = _make_payment(order=order) + + terminal_payment = TerminalPayment.objects.create( + payment=payment, + conference=conf, + payment_intent_id="pi_defaultstatus", + ) + + assert terminal_payment.capture_status == TerminalPayment.CaptureStatus.AUTHORIZED + + def test_capture_status_choices_exist(self) -> None: + assert TerminalPayment.CaptureStatus.AUTHORIZED == "authorized" + assert TerminalPayment.CaptureStatus.CAPTURED == "captured" + assert TerminalPayment.CaptureStatus.CANCELLED == "cancelled" + assert TerminalPayment.CaptureStatus.FAILED == "failed" + + +@pytest.mark.unit +class TestPaymentMethodTerminal: + """Tests for Payment.Method.TERMINAL.""" + + def test_terminal_method_exists(self) -> None: + assert Payment.Method.TERMINAL == "terminal" + + def test_terminal_payment_with_method(self) -> None: + conf = _make_conference() + user = _make_user() + order = _make_order(conference=conf, user=user) + payment = _make_payment(order=order, method=Payment.Method.TERMINAL) + + assert payment.method == "terminal" + assert payment.get_method_display() == "Terminal" + + +# -- View Tests: Terminal API Endpoints ---------------------------------------- + + +@pytest.mark.integration +class TestConnectionTokenView: + """Tests for the terminal connection token endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-connection-token", args=[conference.slug]) + + def test_returns_401_for_anonymous(self) -> None: + conf = _make_conference() + client = Client() + response = client.post( + self._url(conf), + content_type="application/json", + ) + assert response.status_code == 401 + + def test_returns_403_for_non_permitted_user(self) -> None: + conf = _make_conference() + user = _make_user(password="testpass123") + client = Client() + client.force_login(user) + response = client.post( + self._url(conf), + content_type="application/json", + ) + assert response.status_code == 403 + + @patch("django_program.registration.views_terminal.StripeClient") + def test_returns_connection_token(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + staff = _make_staff_user() + + mock_client = MagicMock() + mock_client.create_connection_token.return_value = "pst_test_secret_abc" + mock_client_cls.return_value = mock_client + + client = Client() + client.force_login(staff) + response = client.post( + self._url(conf), + content_type="application/json", + ) + + assert response.status_code == 200 + data = response.json() + assert data["secret"] == "pst_test_secret_abc" + + +@pytest.mark.integration +class TestFetchAttendeeView: + """Tests for the terminal attendee lookup endpoint.""" + + def _url(self, conference: Conference, access_code: str) -> str: + return reverse("registration:terminal-attendee", args=[conference.slug, access_code]) + + def test_returns_attendee_data(self) -> None: + conf = _make_conference() + user = _make_user(first_name="Alice", last_name="Smith") + order = _make_order(conference=conf, user=user) + attendee = _make_attendee(conference=conf, user=user, order=order) + staff = _make_staff_user() + + client = Client() + client.force_login(staff) + response = client.get(self._url(conf, attendee.access_code)) + + assert response.status_code == 200 + data = response.json() + assert data["attendee"]["access_code"] == str(attendee.access_code) + + def test_returns_404_for_unknown_code(self) -> None: + conf = _make_conference() + staff = _make_staff_user() + + client = Client() + client.force_login(staff) + response = client.get(self._url(conf, "NOTFOUND")) + + assert response.status_code == 404 + + +@pytest.mark.integration +class TestFetchInventoryView: + """Tests for the terminal inventory endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-inventory", args=[conference.slug]) + + def test_returns_available_items(self) -> None: + conf = _make_conference() + _make_ticket_type(conference=conf, name="GA", slug="ga") + _make_addon(conference=conf, name="Swag Bag", slug="swag-bag") + staff = _make_staff_user() + + client = Client() + client.force_login(staff) + response = client.get(self._url(conf)) + + assert response.status_code == 200 + data = response.json() + assert "tickets" in data or "ticket_types" in data or "items" in data + + +@pytest.mark.integration +class TestListReadersView: + """Tests for the terminal readers list endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-readers", args=[conference.slug]) + + @patch("django_program.registration.views_terminal.StripeClient") + def test_returns_readers(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + staff = _make_staff_user() + + mock_reader = MagicMock() + mock_reader.id = "tmr_abc123" + mock_reader.label = "Front Desk Reader" + mock_reader.status = "online" + mock_reader.device_type = "bbpos_wisepos_e" + mock_reader.location = "tml_location_1" + mock_reader.serial_number = "SN12345" + + mock_client = MagicMock() + mock_client.list_readers.return_value = [mock_reader] + mock_client_cls.return_value = mock_client + + client = Client() + client.force_login(staff) + response = client.get(self._url(conf)) + + assert response.status_code == 200 + data = response.json() + assert "readers" in data + + +@pytest.mark.integration +class TestCreatePaymentIntentView: + """Tests for the terminal create payment intent endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-create-intent", args=[conference.slug]) + + @patch("django_program.registration.views_terminal.StripeClient") + def test_creates_payment_intent(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + staff = _make_staff_user() + + mock_intent = MagicMock() + mock_intent.id = "pi_terminal_test_123" + mock_intent.status = "requires_payment_method" + mock_intent.client_secret = "pi_terminal_test_123_secret_abc" + + mock_reader_result = MagicMock() + mock_reader_result.action = None + + mock_client = MagicMock() + mock_client.create_terminal_payment_intent.return_value = mock_intent + mock_client.process_terminal_payment.return_value = mock_reader_result + mock_client_cls.return_value = mock_client + + client = Client() + client.force_login(staff) + response = client.post( + self._url(conf), + data=json.dumps({"amount": "100.00", "currency": "usd", "reader_id": "tmr_test123"}), + content_type="application/json", + ) + + assert response.status_code == 200 + data = response.json() + assert data["payment_intent_id"] == "pi_terminal_test_123" + assert data["status"] == "processing" + + +@pytest.mark.integration +class TestCapturePaymentView: + """Tests for the terminal capture payment endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-capture", args=[conference.slug]) + + @patch("django_program.registration.views_terminal.StripeClient") + def test_captures_payment(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + staff = _make_staff_user() + user = _make_user() + order = _make_order(conference=conf, user=user) + payment = _make_payment(order=order) + TerminalPayment.objects.create( + payment=payment, + conference=conf, + payment_intent_id="pi_to_capture", + capture_status=TerminalPayment.CaptureStatus.AUTHORIZED, + ) + + mock_intent = MagicMock() + mock_intent.id = "pi_to_capture" + mock_intent.status = "succeeded" + + mock_client = MagicMock() + mock_client.capture_payment_intent.return_value = mock_intent + mock_client_cls.return_value = mock_client + + client = Client() + client.force_login(staff) + response = client.post( + self._url(conf), + data=json.dumps({"payment_intent_id": "pi_to_capture"}), + content_type="application/json", + ) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "captured" + + +@pytest.mark.integration +class TestCancelPaymentView: + """Tests for the terminal cancel payment endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-cancel", args=[conference.slug]) + + @patch("django_program.registration.views_terminal.StripeClient") + def test_cancels_payment(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + staff = _make_staff_user() + user = _make_user() + order = _make_order(conference=conf, user=user) + payment = _make_payment(order=order) + TerminalPayment.objects.create( + payment=payment, + conference=conf, + payment_intent_id="pi_to_cancel", + capture_status=TerminalPayment.CaptureStatus.AUTHORIZED, + ) + + mock_intent = MagicMock() + mock_intent.id = "pi_to_cancel" + mock_intent.status = "canceled" + + mock_client = MagicMock() + mock_client.cancel_payment_intent.return_value = mock_intent + mock_client_cls.return_value = mock_client + + client = Client() + client.force_login(staff) + response = client.post( + self._url(conf), + data=json.dumps({"payment_intent_id": "pi_to_cancel"}), + content_type="application/json", + ) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "cancelled" + + +@pytest.mark.integration +class TestCartOperationsView: + """Tests for the terminal cart operations endpoint.""" + + def _url(self, conference: Conference) -> str: + return reverse("registration:terminal-cart", args=[conference.slug]) + + @patch("django_program.registration.views_terminal.StripeClient") + def test_create_cart(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + ticket = _make_ticket_type(conference=conf, name="Day Pass", slug="day-pass") + staff = _make_staff_user() + + mock_client_cls.return_value = MagicMock() + + client = Client() + client.force_login(staff) + response = client.post( + self._url(conf), + data=json.dumps( + { + "action": "update", + "items": [{"ticket_type_id": ticket.pk, "quantity": 1}], + } + ), + content_type="application/json", + ) + + assert response.status_code in (200, 201) + + @patch("django_program.registration.views_terminal.StripeClient") + def test_cart_checkout(self, mock_client_cls: MagicMock) -> None: + conf = _make_conference() + ticket = _make_ticket_type(conference=conf, name="Day Pass", slug="day-pass-co") + staff = _make_staff_user() + + mock_client_cls.return_value = MagicMock() + + client = Client() + client.force_login(staff) + + # First create a cart with items via the update action + response = client.post( + self._url(conf), + data=json.dumps( + { + "action": "update", + "items": [{"ticket_type_id": ticket.pk, "quantity": 1}], + } + ), + content_type="application/json", + ) + assert response.status_code == 200 + + # Now checkout the cart + response = client.post( + self._url(conf), + data=json.dumps( + { + "action": "checkout", + } + ), + content_type="application/json", + ) + + assert response.status_code in (200, 201) + + +# -- Manage View Tests: Terminal POS ------------------------------------------ + + +@pytest.mark.integration +class TestTerminalPOSView: + """Tests for the manage terminal POS page.""" + + def _url(self, conference: Conference) -> str: + return reverse("manage:terminal-pos", args=[conference.slug]) + + def test_anonymous_redirected(self) -> None: + conf = _make_conference() + client = Client() + response = client.get(self._url(conf)) + assert response.status_code == 302 + + def test_non_permitted_user_denied(self) -> None: + conf = _make_conference() + user = _make_user(password="testpass123") + client = Client() + client.force_login(user) + response = client.get(self._url(conf)) + assert response.status_code == 403 + + def test_staff_with_permission_gets_200(self) -> None: + conf = _make_conference() + staff = _make_staff_user() + client = Client() + client.force_login(staff) + response = client.get(self._url(conf)) + assert response.status_code == 200 + + def test_context_has_active_nav(self) -> None: + conf = _make_conference() + staff = _make_staff_user() + client = Client() + client.force_login(staff) + response = client.get(self._url(conf)) + assert response.context["active_nav"] == "terminal"