TradeMate Dev
3e39cf0170
Switch from direct WeChat Pay / Alipay integrations to the unified
宇之然 pay-api gateway (HMAC-SHA256 auth). Removes wechat_pay.py,
keeps PaymentGateway abstraction, adds UnifiedPayService. Simplifies
payment.py create_order to {plan, pay_type} params. Single webhook
endpoint replaces separate WeChat/Alipay notify handlers.
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
async def _setup_csrf(client: AsyncClient, auth_headers: dict) -> dict:
|
|
resp = await client.get("/api/v1/payment/plans", headers=auth_headers)
|
|
csrf = resp.headers.get("X-CSRF-Token", "")
|
|
return {**auth_headers, "X-CSRF-Token": csrf}
|
|
|
|
|
|
class TestPaymentAPI:
|
|
async def test_get_plans(self, client: AsyncClient):
|
|
response = await client.get("/api/v1/payment/plans")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "plans" in data
|
|
assert "gateways" in data
|
|
plan_ids = [p["id"] for p in data["plans"]]
|
|
assert "free" in plan_ids
|
|
assert "pro" in plan_ids
|
|
assert "enterprise" in plan_ids
|
|
|
|
async def test_get_subscription_no_auth(self, client: AsyncClient):
|
|
response = await client.get("/api/v1/payment/subscription")
|
|
assert response.status_code == 401
|
|
|
|
async def test_get_subscription_authenticated(self, client: AsyncClient,
|
|
auth_headers: dict):
|
|
response = await client.get(
|
|
"/api/v1/payment/subscription",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "plan" in data
|
|
assert "status" in data
|
|
|
|
async def test_create_free_plan_order(self, client: AsyncClient,
|
|
auth_headers: dict):
|
|
csrf_headers = await _setup_csrf(client, auth_headers)
|
|
response = await client.post(
|
|
"/api/v1/payment/create-order",
|
|
json={"plan": "free", "pay_type": "alipay"},
|
|
headers=csrf_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert data["plan"] == "free"
|
|
assert data["amount"] == 0
|
|
|
|
async def test_query_order_not_found(self, client: AsyncClient,
|
|
auth_headers: dict):
|
|
response = await client.get(
|
|
"/api/v1/payment/query/NONEXISTENT",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
async def test_list_transactions(self, client: AsyncClient,
|
|
auth_headers: dict):
|
|
response = await client.get(
|
|
"/api/v1/payment/transactions",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
|
|
async def test_admin_list_payments_no_auth(self, client: AsyncClient):
|
|
response = await client.get("/api/v1/admin/payments")
|
|
assert response.status_code == 401
|
|
|
|
async def test_admin_payment_stats_no_auth(self, client: AsyncClient):
|
|
response = await client.get("/api/v1/admin/payments/stats")
|
|
assert response.status_code == 401
|
|
|
|
async def test_refund_no_auth(self, client: AsyncClient):
|
|
response = await client.post("/api/v1/payment/refund",
|
|
json={"order_no": "test", "reason": ""})
|
|
assert response.status_code == 401
|