booking_backend/payment/tests.py

from datetime import timedelta

from django.contrib.auth import get_user_model
from django.db import IntegrityError
from django.test import TestCase
from django.utils import timezone
from rest_framework import status
from rest_framework.test import APITestCase

from accounts.models import VendorProfile
from booking.models import Booking
from equipment.models import EquipmentCategory, EquipmentItem

from .models import PaymentRecord, WebhookEvent

User = get_user_model()


class PaymentModelTests(TestCase):
    def setUp(self):
        vendor_user = User.objects.create_user(
            email="vendor-payment@example.com",
            password="Pass123456!",
            is_vendor=True,
            is_customer=False,
        )
        vendor = VendorProfile.objects.create(user=vendor_user, business_name="Payment Vendor")
        customer = User.objects.create_user(
            email="customer-payment@example.com",
            password="Pass123456!",
            is_vendor=False,
            is_customer=True,
        )
        category = EquipmentCategory.objects.create(name="Canoe", slug="canoe")
        item = EquipmentItem.objects.create(
            vendor=vendor,
            category=category,
            title="Canoe",
            public_id="canoe-001",
            price_per_day="45.00",
            is_active=True,
        )
        starts = timezone.now() + timedelta(days=3)
        ends = starts + timedelta(days=1)
        self.booking = Booking.objects.create(
            customer=customer,
            vendor=vendor,
            equipment_item=item,
            starts_at=starts,
            ends_at=ends,
            status=Booking.Status.APPROVED,
            total_price="45.00",
        )

    def test_payment_idempotency_constraints(self):
        PaymentRecord.objects.create(
            booking=self.booking,
            stripe_payment_intent_id="pi_mock_dup",
            amount="45.00",
            currency="usd",
        )
        with self.assertRaises(IntegrityError):
            PaymentRecord.objects.create(
                booking=self.booking,
                stripe_payment_intent_id="pi_mock_dup",
                amount="45.00",
                currency="usd",
            )

        WebhookEvent.objects.create(
            stripe_event_id="evt_mock_dup",
            event_type="payment_intent.succeeded",
            payload={},
        )
        with self.assertRaises(IntegrityError):
            WebhookEvent.objects.create(
                stripe_event_id="evt_mock_dup",
                event_type="payment_intent.succeeded",
                payload={},
            )


class PaymentApiTests(APITestCase):
    def setUp(self):
        self.vendor_user = User.objects.create_user(
            email="vendor-api-payment@example.com",
            password="Pass123456!",
            is_vendor=True,
            is_customer=False,
        )
        self.vendor = VendorProfile.objects.create(user=self.vendor_user, business_name="Webhook Vendor")
        self.customer = User.objects.create_user(
            email="customer-api-payment@example.com",
            password="Pass123456!",
            is_vendor=False,
            is_customer=True,
        )
        category = EquipmentCategory.objects.create(name="Raft", slug="raft")
        item = EquipmentItem.objects.create(
            vendor=self.vendor,
            category=category,
            title="River Raft",
            public_id="raft-001",
            price_per_day="110.00",
            is_active=True,
        )
        starts = timezone.now() + timedelta(days=2)
        ends = starts + timedelta(days=1)
        self.booking = Booking.objects.create(
            customer=self.customer,
            vendor=self.vendor,
            equipment_item=item,
            starts_at=starts,
            ends_at=ends,
            status=Booking.Status.APPROVED,
            total_price="110.00",
        )

    def test_mock_payment_intent_status_and_webhook_flow(self):
        self.client.force_authenticate(self.customer)
        create_res = self.client.post(
            "/api/v1/payment/intents/",
            {"booking_id": self.booking.id, "currency": "usd"},
            format="json",
        )
        self.assertEqual(create_res.status_code, status.HTTP_201_CREATED)
        payment_id = create_res.data["payment"]["id"]
        intent_id = create_res.data["payment"]["stripe_payment_intent_id"]

        status_res = self.client.get(f"/api/v1/payment/{payment_id}/status/")
        self.assertEqual(status_res.status_code, status.HTTP_200_OK)
        self.assertEqual(status_res.data["status"], PaymentRecord.Status.REQUIRES_PAYMENT)

        webhook_res = self.client.post(
            "/api/v1/payment/webhooks/stripe/",
            {
                "stripe_event_id": "evt_mock_1001",
                "event_type": "payment_intent.succeeded",
                "stripe_payment_intent_id": intent_id,
                "payload": {"source": "test"},
            },
            format="json",
        )
        self.assertEqual(webhook_res.status_code, status.HTTP_200_OK)
        self.booking.refresh_from_db()
        self.assertEqual(self.booking.status, Booking.Status.CONFIRMED)

        idempotent_res = self.client.post(
            "/api/v1/payment/webhooks/stripe/",
            {
                "stripe_event_id": "evt_mock_1001",
                "event_type": "payment_intent.succeeded",
                "stripe_payment_intent_id": intent_id,
                "payload": {},
            },
            format="json",
        )
        self.assertEqual(idempotent_res.status_code, status.HTTP_200_OK)
        self.assertTrue(idempotent_res.data["idempotent"])