Desi-Clans/backend/tests/test_orders.py

import uuid
from concurrent.futures import ThreadPoolExecutor

import pytest
from sqlalchemy import select

from app.db.session import SessionLocal
from app.main import app
from app.models.catalog import TicketType
from app.models.enums import OrderStatus, TicketTypeStatus
from app.models.user import User
from app.schemas.orders import OrderCreate, OrderLineCreate
from app.services.orders import InsufficientInventoryError, create_order
from app.services.payments.gateway import MockPaymentGateway, get_payment_gateway
from tests.helpers import (
    auth_headers,
    client,
    create_pending_order,
    first_published_event,
    pick_tier,
    register_user,
)


def _complete_mock_payment(token: str, checkout: dict) -> None:
    gateway = get_payment_gateway()
    assert isinstance(gateway, MockPaymentGateway)
    payment = checkout["payment"]
    payment_id = f"mock_pay_{checkout['id']}"
    signature = gateway.build_mock_checkout_signature(payment["provider_order_id"], payment_id)
    response = client.post(
        f"/api/orders/{checkout['id']}/payment/verify",
        headers=auth_headers(token),
        json={
            "provider_order_id": payment["provider_order_id"],
            "provider_payment_id": payment_id,
            "signature": signature,
        },
    )
    assert response.status_code == 200


def test_list_event_tickets() -> None:
    event = first_published_event()
    response = client.get(f"/api/events/{event['id']}/tickets")
    assert response.status_code == 200
    tiers = response.json()
    assert tiers
    assert tiers[0]["event_id"] == event["id"]
    assert "remaining_quantity" in tiers[0]
    assert "price_paise" in tiers[0]


def test_create_order_pending_until_verified() -> None:
    token, checkout = create_pending_order()
    assert checkout["status"] == OrderStatus.pending.value
    assert checkout["payment"]["amount_paise"] == checkout["total_paise"]
    assert checkout["payment"]["provider_order_id"]

    tickets_before = client.get("/api/tickets/my", headers=auth_headers(token)).json()
    assert not any(t["order_id"] == checkout["id"] for t in tickets_before)

    _complete_mock_payment(token, checkout)

    my_tickets = client.get("/api/tickets/my", headers=auth_headers(token))
    assert my_tickets.status_code == 200
    tickets = my_tickets.json()
    assert any(t["order_id"] == checkout["id"] for t in tickets)
    assert all(t["code"].startswith("DC-") for t in tickets)


def test_order_requires_authentication() -> None:
    event = first_published_event()
    response = client.post(
        "/api/orders",
        json={"event_id": event["id"], "items": [{"ticket_type_id": 1, "quantity": 1}]},
    )
    assert response.status_code == 401


def test_cannot_oversell_inventory() -> None:
    _, token = register_user()
    event = first_published_event()
    tiers = client.get(f"/api/events/{event['id']}/tickets").json()
    tier = pick_tier(tiers, 1)
    if not tier:
        pytest.skip("No ticket inventory available for test")
    if tier["remaining_quantity"] >= 20:
        pytest.skip("Quantity cap prevents oversell test for this tier")

    response = client.post(
        "/api/orders",
        headers=auth_headers(token),
        json={
            "event_id": event["id"],
            "items": [{"ticket_type_id": tier["id"], "quantity": tier["remaining_quantity"] + 1}],
        },
    )
    assert response.status_code == 409


def test_server_calculates_total_not_client() -> None:
    token, checkout = create_pending_order()
    tiers = client.get(f"/api/events/{checkout['event_id']}/tickets").json()
    tier = next(t for t in tiers if t["id"] == checkout["items"][0]["ticket_type_id"])
    qty = checkout["items"][0]["quantity"]
    assert checkout["total_paise"] == tier["price_paise"] * qty


def test_concurrent_purchases_do_not_oversell() -> None:
    event = first_published_event()
    db = SessionLocal()
    try:
        tier = db.scalar(
            select(TicketType).where(TicketType.event_id == event["id"]).order_by(TicketType.id.asc())
        )
        assert tier is not None
        tier.quantity_sold = max(tier.quantity_total - 1, 0)
        tier.quantity_total = tier.quantity_sold + 1
        tier.status = TicketTypeStatus.active
        db.commit()
        ticket_type_id = tier.id
        event_id = event["id"]
    finally:
        db.close()

    user_ids: list[int] = []
    for _ in range(2):
        email = f"race-{uuid.uuid4().hex}@example.com"
        reg = client.post(
            "/api/auth/register",
            json={"email": email, "password": "securepass1"},
        )
        assert reg.status_code == 201
        db = SessionLocal()
        try:
            user = db.scalar(select(User).where(User.email == email))
            assert user is not None
            user_ids.append(user.id)
        finally:
            db.close()

    def _buy(user_id: int) -> str:
        session = SessionLocal()
        try:
            user = session.get(User, user_id)
            create_order(
                session,
                user,
                OrderCreate(
                    event_id=event_id,
                    items=[OrderLineCreate(ticket_type_id=ticket_type_id, quantity=1)],
                ),
            )
            return "success"
        except InsufficientInventoryError:
            session.rollback()
            return "conflict"
        finally:
            session.close()

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(_buy, user_ids))

    assert results.count("success") == 1
    assert results.count("conflict") == 1

    db = SessionLocal()
    try:
        refreshed = db.get(TicketType, ticket_type_id)
        assert refreshed is not None
        assert refreshed.quantity_sold == refreshed.quantity_total
    finally:
        db.close()