Spaces:
Sleeping
Sleeping
| """ | |
| 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"]) | |