97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
from app.core.config import get_settings
|
|
from app.services.payments.gateway import MockPaymentGateway, get_payment_gateway
|
|
from tests.helpers import auth_headers, client, create_pending_order
|
|
|
|
|
|
def _mock_signature(body: bytes) -> str:
|
|
secret = get_settings().payment_mock_secret
|
|
return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def test_verify_payment_issues_tickets() -> None:
|
|
token, checkout = create_pending_order()
|
|
assert checkout["status"] == "pending"
|
|
payment = checkout["payment"]
|
|
assert payment["provider"] == "mock"
|
|
|
|
gateway = get_payment_gateway()
|
|
assert isinstance(gateway, MockPaymentGateway)
|
|
payment_id = f"mock_pay_{checkout['id']}"
|
|
signature = gateway.build_mock_checkout_signature(payment["provider_order_id"], payment_id)
|
|
|
|
verify = 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 verify.status_code == 200
|
|
assert verify.json()["status"] == "paid"
|
|
|
|
tickets = client.get("/api/tickets/my", headers=auth_headers(token)).json()
|
|
assert any(t["order_id"] == checkout["id"] for t in tickets)
|
|
|
|
|
|
def test_rejects_invalid_signature() -> None:
|
|
token, checkout = create_pending_order()
|
|
payment = checkout["payment"]
|
|
verify = client.post(
|
|
f"/api/orders/{checkout['id']}/payment/verify",
|
|
headers=auth_headers(token),
|
|
json={
|
|
"provider_order_id": payment["provider_order_id"],
|
|
"provider_payment_id": "mock_pay_bad",
|
|
"signature": "not-valid",
|
|
},
|
|
)
|
|
assert verify.status_code == 400
|
|
|
|
|
|
def test_webhook_idempotent() -> None:
|
|
token, checkout = create_pending_order()
|
|
payment = checkout["payment"]
|
|
payment_id = f"mock_pay_wh_{checkout['id']}"
|
|
import uuid
|
|
|
|
event_id = f"evt_duplicate_{uuid.uuid4().hex}"
|
|
payload = {
|
|
"id": event_id,
|
|
"event": "payment.captured",
|
|
"payload": {
|
|
"payment": {
|
|
"entity": {
|
|
"id": payment_id,
|
|
"order_id": payment["provider_order_id"],
|
|
"amount": checkout["total_paise"],
|
|
"currency": "INR",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
body = json.dumps(payload).encode()
|
|
signature = _mock_signature(body)
|
|
headers = {"X-Mock-Signature": signature}
|
|
|
|
first = client.post("/api/webhooks/mock", content=body, headers=headers)
|
|
assert first.status_code == 200
|
|
assert first.json()["status"] == "paid"
|
|
|
|
second = client.post("/api/webhooks/mock", content=body, headers=headers)
|
|
assert second.status_code == 200
|
|
assert second.json()["status"] == "duplicate_ignored"
|
|
|
|
tickets = client.get("/api/tickets/my", headers=auth_headers(token)).json()
|
|
assert len([t for t in tickets if t["order_id"] == checkout["id"]]) == 1
|
|
|
|
|
|
def test_cancel_pending_order_releases_inventory() -> None:
|
|
token, checkout = create_pending_order()
|
|
cancel = client.post(f"/api/orders/{checkout['id']}/cancel", headers=auth_headers(token))
|
|
assert cancel.status_code == 200
|
|
assert cancel.json()["status"] == "cancelled"
|