apigateway / tests /test_payments_router.py
jebin2's picture
ref
a42ab7e
"""
Rigorous Tests for Payments Router.
Tests cover:
1. Helper functions (generate_transaction_id, update_verified_by, process_successful_payment)
2. GET /packages endpoint
3. POST /create-order endpoint
4. POST /verify endpoint
5. POST /webhook/razorpay endpoint
6. GET /history endpoint
Uses mocked Razorpay service and database.
"""
import pytest
import json
from datetime import datetime
from unittest.mock import patch, MagicMock, AsyncMock
from fastapi.testclient import TestClient
# =============================================================================
# 1. Helper Function Tests
# =============================================================================
class TestHelperFunctions:
"""Test helper functions in payments router."""
def test_generate_transaction_id_format(self):
"""Generated transaction IDs have correct format."""
from routers.payments import generate_transaction_id
txn_id = generate_transaction_id()
assert txn_id.startswith("txn_")
assert len(txn_id) == 20 # "txn_" + 16 hex chars
def test_generate_transaction_id_unique(self):
"""Each generated ID is unique."""
from routers.payments import generate_transaction_id
ids = [generate_transaction_id() for _ in range(100)]
assert len(set(ids)) == 100 # All unique
def test_update_verified_by_client_first(self):
"""First verification by client sets verified_by to 'client'."""
from routers.payments import update_verified_by
transaction = MagicMock()
transaction.verified_by = None
changed = update_verified_by(transaction, "client")
assert transaction.verified_by == "client"
assert changed == True
def test_update_verified_by_webhook_first(self):
"""First verification by webhook sets verified_by to 'webhook'."""
from routers.payments import update_verified_by
transaction = MagicMock()
transaction.verified_by = None
changed = update_verified_by(transaction, "webhook")
assert transaction.verified_by == "webhook"
assert changed == True
def test_update_verified_by_both_sources(self):
"""Second verification from other source sets verified_by to 'both'."""
from routers.payments import update_verified_by
# Client first, then webhook
transaction = MagicMock()
transaction.verified_by = "client"
changed = update_verified_by(transaction, "webhook")
assert transaction.verified_by == "both"
assert changed == True
def test_update_verified_by_same_source_no_change(self):
"""Same source verification doesn't change value."""
from routers.payments import update_verified_by
transaction = MagicMock()
transaction.verified_by = "client"
changed = update_verified_by(transaction, "client")
assert transaction.verified_by == "client"
assert changed == False
# =============================================================================
# 2. GET /packages Tests
# =============================================================================
class TestGetPackages:
"""Test GET /packages endpoint."""
def test_list_packages_returns_all(self):
"""List all available packages."""
from routers.payments import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
response = client.get("/payments/packages")
assert response.status_code == 200
data = response.json()
assert "packages" in data
assert len(data["packages"]) >= 3 # At least starter, standard, pro
def test_packages_have_required_fields(self):
"""Each package has all required fields."""
from routers.payments import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
response = client.get("/payments/packages")
data = response.json()
for pkg in data["packages"]:
assert "id" in pkg
assert "name" in pkg
assert "credits" in pkg
assert "amount_paise" in pkg
assert "currency" in pkg
# =============================================================================
# 3. POST /create-order Tests
# =============================================================================
class TestCreateOrder:
"""Test POST /create-order endpoint."""
def test_create_order_requires_auth(self):
"""Create order requires authentication."""
from routers.payments import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
response = client.post(
"/payments/create-order",
json={"package_id": "starter"}
)
# Should fail with auth error (401 or 403)
assert response.status_code in [401, 403, 422]
def test_create_order_invalid_package(self):
"""Reject invalid package_id."""
from routers.payments import router
from fastapi import FastAPI
from core.dependencies import get_current_user
app = FastAPI()
# Mock authenticated user
mock_user = MagicMock()
mock_user.user_id = "test-user"
mock_user.credits = 100
app.dependency_overrides[get_current_user] = lambda: mock_user
app.include_router(router)
client = TestClient(app)
with patch('routers.payments.is_razorpay_configured', return_value=True):
response = client.post(
"/payments/create-order",
json={"package_id": "invalid_package"}
)
assert response.status_code == 400
assert "Invalid package" in response.json()["detail"]
def test_create_order_razorpay_not_configured(self):
"""Return 503 if Razorpay not configured."""
from routers.payments import router
from fastapi import FastAPI
from core.dependencies import get_current_user
app = FastAPI()
mock_user = MagicMock()
mock_user.user_id = "test-user"
app.dependency_overrides[get_current_user] = lambda: mock_user
app.include_router(router)
client = TestClient(app)
with patch('routers.payments.is_razorpay_configured', return_value=False):
response = client.post(
"/payments/create-order",
json={"package_id": "starter"}
)
assert response.status_code == 503
assert "not configured" in response.json()["detail"]
# =============================================================================
# 4. POST /verify Tests
# =============================================================================
class TestVerifyPayment:
"""Test POST /verify endpoint."""
def test_verify_requires_auth(self):
"""Verify requires authentication."""
from routers.payments import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
response = client.post(
"/payments/verify",
json={
"razorpay_order_id": "order_123",
"razorpay_payment_id": "pay_123",
"razorpay_signature": "sig_123"
}
)
assert response.status_code in [401, 403, 422]
def test_verify_transaction_not_found(self):
"""Return 404 for unknown transaction."""
from routers.payments import router
from fastapi import FastAPI
from core.dependencies import get_current_user
from core.database import get_db
app = FastAPI()
mock_user = MagicMock()
mock_user.user_id = "test-user"
mock_user.credits = 100
# Mock database that returns no transaction
async def mock_get_db():
mock_db = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = None
mock_db.execute.return_value = mock_result
yield mock_db
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
with patch('routers.payments.get_razorpay_service') as mock_service:
mock_service.return_value = MagicMock()
response = client.post(
"/payments/verify",
json={
"razorpay_order_id": "order_unknown",
"razorpay_payment_id": "pay_123",
"razorpay_signature": "sig_123"
}
)
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
# =============================================================================
# 5. POST /webhook/razorpay Tests
# =============================================================================
class TestWebhook:
"""Test POST /webhook/razorpay endpoint."""
def test_webhook_requires_signature(self):
"""Webhook requires X-Razorpay-Signature header."""
from routers.payments import router
from fastapi import FastAPI
from core.database import get_db
app = FastAPI()
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
response = client.post(
"/payments/webhook/razorpay",
json={"event": "payment.captured"}
)
assert response.status_code == 401
assert "signature" in response.json()["detail"].lower()
def test_webhook_rejects_invalid_signature(self):
"""Webhook rejects invalid signature."""
from routers.payments import router
from fastapi import FastAPI
from core.database import get_db
app = FastAPI()
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
with patch('routers.payments.get_razorpay_service') as mock_service:
service_instance = MagicMock()
service_instance.verify_webhook_signature.return_value = False
mock_service.return_value = service_instance
response = client.post(
"/payments/webhook/razorpay",
json={"event": "payment.captured"},
headers={"X-Razorpay-Signature": "invalid-sig"}
)
assert response.status_code == 401
assert "invalid" in response.json()["detail"].lower()
def test_webhook_accepts_valid_signature(self):
"""Webhook accepts valid signature."""
from routers.payments import router
from fastapi import FastAPI
from core.database import get_db
app = FastAPI()
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
with patch('routers.payments.get_razorpay_service') as mock_service:
service_instance = MagicMock()
service_instance.verify_webhook_signature.return_value = True
mock_service.return_value = service_instance
response = client.post(
"/payments/webhook/razorpay",
json={"event": "unknown.event"},
headers={"X-Razorpay-Signature": "valid-sig"}
)
assert response.status_code == 200
assert response.json()["status"] == "ok"
# =============================================================================
# 6. GET /history Tests
# =============================================================================
class TestPaymentHistory:
"""Test GET /history endpoint."""
def test_history_requires_auth(self):
"""History requires authentication."""
from routers.payments import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
response = client.get("/payments/history")
assert response.status_code in [401, 403, 422]
def test_history_returns_empty_list(self):
"""History returns empty list for user with no transactions."""
from routers.payments import router
from fastapi import FastAPI
from core.dependencies import get_current_user
from core.database import get_db
app = FastAPI()
mock_user = MagicMock()
mock_user.user_id = "test-user"
async def mock_get_db():
mock_db = AsyncMock()
# Mock count query
mock_count_result = MagicMock()
mock_count_result.scalar.return_value = 0
# Mock transactions query
mock_txn_result = MagicMock()
mock_txn_result.scalars.return_value.all.return_value = []
mock_db.execute.side_effect = [mock_count_result, mock_txn_result]
yield mock_db
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
response = client.get("/payments/history")
assert response.status_code == 200
data = response.json()
assert data["transactions"] == []
assert data["total_count"] == 0
def test_history_pagination_params(self):
"""History respects pagination parameters."""
from routers.payments import router
from fastapi import FastAPI
from core.dependencies import get_current_user
from core.database import get_db
app = FastAPI()
mock_user = MagicMock()
mock_user.user_id = "test-user"
async def mock_get_db():
mock_db = AsyncMock()
mock_count_result = MagicMock()
mock_count_result.scalar.return_value = 50
mock_txn_result = MagicMock()
mock_txn_result.scalars.return_value.all.return_value = []
mock_db.execute.side_effect = [mock_count_result, mock_txn_result]
yield mock_db
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
response = client.get("/payments/history?page=2&limit=10")
assert response.status_code == 200
data = response.json()
assert data["page"] == 2
assert data["limit"] == 10
assert data["total_count"] == 50
# =============================================================================
# 7. Response Model Tests
# =============================================================================
class TestResponseModels:
"""Test response model schemas."""
def test_package_response_model(self):
"""PackageResponse model validates correctly."""
from routers.payments import PackageResponse
pkg = PackageResponse(
id="starter",
name="Starter",
credits=100,
amount_paise=9900,
amount_rupees=99.0,
currency="INR"
)
assert pkg.id == "starter"
assert pkg.credits == 100
def test_verify_payment_response_model(self):
"""VerifyPaymentResponse model validates correctly."""
from routers.payments import VerifyPaymentResponse
resp = VerifyPaymentResponse(
success=True,
message="Payment successful",
transaction_id="txn_abc123",
credits_added=100,
new_balance=500
)
assert resp.success == True
assert resp.credits_added == 100
def test_payment_history_item_model(self):
"""PaymentHistoryItem model validates correctly."""
from routers.payments import PaymentHistoryItem
item = PaymentHistoryItem(
transaction_id="txn_123",
package_id="starter",
credits_amount=100,
amount_paise=9900,
currency="INR",
status="paid",
gateway="razorpay",
created_at="2024-01-01T00:00:00"
)
assert item.transaction_id == "txn_123"
assert item.status == "paid"
if __name__ == "__main__":
pytest.main([__file__, "-v"])