366 lines
17 KiB
Python
366 lines
17 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
|
|
from app.services.payment import get_payment_gateway
|
|
from app.services.payment.base import PaymentGateway, PaymentOrder, PaymentCallback
|
|
from app.services.payment.mock_gateway import MockGateway
|
|
from app.services.payment.wechat_pay import WeChatPayGateway
|
|
from app.services.payment.alipay import AlipayGateway
|
|
|
|
|
|
class TestPaymentGatewaySelection:
|
|
def test_mock_mode_returns_mock_gateway(self):
|
|
with patch("app.config.settings") as mock_settings:
|
|
mock_settings.PAYMENT_MODE = "mock"
|
|
mock_settings.WECHAT_PAY_MCH_ID = ""
|
|
gateway = get_payment_gateway("wechat")
|
|
assert isinstance(gateway, MockGateway)
|
|
|
|
def test_unconfigured_wechat_returns_mock_gateway(self):
|
|
with patch("app.config.settings") as mock_settings:
|
|
mock_settings.PAYMENT_MODE = "live"
|
|
mock_settings.WECHAT_PAY_MCH_ID = ""
|
|
gateway = get_payment_gateway("wechat")
|
|
assert isinstance(gateway, MockGateway)
|
|
|
|
def test_configured_wechat_returns_wechat_gateway(self):
|
|
with patch("app.config.settings") as mock_settings:
|
|
mock_settings.PAYMENT_MODE = "live"
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_key"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx123456"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
mock_settings.ALIPAY_APP_ID = ""
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = ""
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = ""
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gateway = get_payment_gateway("wechat")
|
|
assert isinstance(gateway, WeChatPayGateway)
|
|
|
|
def test_configured_alipay_returns_alipay_gateway(self):
|
|
with patch("app.config.settings") as mock_settings:
|
|
mock_settings.PAYMENT_MODE = "live"
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_key"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx123456"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
mock_settings.ALIPAY_APP_ID = "alipay_app_123"
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/key"
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/pub"
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gateway = get_payment_gateway("alipay")
|
|
assert isinstance(gateway, AlipayGateway)
|
|
|
|
def test_unknown_provider_returns_mock_gateway(self):
|
|
with patch("app.config.settings") as mock_settings:
|
|
mock_settings.PAYMENT_MODE = "live"
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_key"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx123456"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
mock_settings.ALIPAY_APP_ID = ""
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = ""
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = ""
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gateway = get_payment_gateway("unknown")
|
|
assert isinstance(gateway, MockGateway)
|
|
|
|
|
|
class TestMockGateway:
|
|
@pytest.mark.asyncio
|
|
async def test_create_order(self):
|
|
gw = MockGateway()
|
|
result = await gw.create_order("order1", 99.9, "test", "user1", "pro")
|
|
assert isinstance(result, PaymentOrder)
|
|
assert result.order_id == "order1"
|
|
assert result.amount == 99.9
|
|
assert result.status == "pending"
|
|
assert "mock://pay/order1" == result.pay_url
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_callback(self):
|
|
gw = MockGateway()
|
|
result = await gw.verify_callback({"order_id": "order1", "amount": "99.9"})
|
|
assert isinstance(result, PaymentCallback)
|
|
assert result.order_id == "order1"
|
|
assert result.status == "success"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_order(self):
|
|
gw = MockGateway()
|
|
result = await gw.query_order("order1")
|
|
assert isinstance(result, PaymentOrder)
|
|
assert result.order_id == "order1"
|
|
assert result.status == "completed"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refund(self):
|
|
gw = MockGateway()
|
|
result = await gw.refund("order1", 99.9, "test")
|
|
assert result is True
|
|
|
|
|
|
class TestWeChatPayGatewayUnconfigured:
|
|
def test_is_configured_false(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = ""
|
|
mock_settings.WECHAT_PAY_API_KEY = ""
|
|
mock_settings.WECHAT_PAY_APP_ID = ""
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
assert gw._is_configured() is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_fallback_to_mock(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = ""
|
|
mock_settings.WECHAT_PAY_API_KEY = ""
|
|
mock_settings.WECHAT_PAY_APP_ID = ""
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
result = await gw.create_order("order1", 100.0, "test", "user1", "pro")
|
|
assert isinstance(result, PaymentOrder)
|
|
assert result.order_id == "order1"
|
|
assert "mock://pay/" in result.pay_url
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_callback_fallback_to_mock(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = ""
|
|
mock_settings.WECHAT_PAY_API_KEY = ""
|
|
mock_settings.WECHAT_PAY_APP_ID = ""
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
result = await gw.verify_callback({"order_id": "order1", "amount": "100"})
|
|
assert isinstance(result, PaymentCallback)
|
|
assert result.status == "success"
|
|
|
|
|
|
class TestWeChatPayGatewayConfigured:
|
|
def test_is_configured_true(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = "/certs/apiclient_cert.pem"
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = "https://example.com/callback"
|
|
gw = WeChatPayGateway()
|
|
assert gw._is_configured() is True
|
|
|
|
def test_build_signature(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
sig = gw._build_signature("POST", "/v3/pay/transactions/native", "1234567890", "abc123", '{"test":1}')
|
|
assert isinstance(sig, str)
|
|
assert len(sig) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_calls_api(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = "https://example.com/callback"
|
|
gw = WeChatPayGateway()
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"code_url": "weixin://wxpay/bizpayurl?pr=test123"}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("app.services.payment.wechat_pay.requests.post", return_value=mock_resp):
|
|
result = await gw.create_order("order1", 100.0, "test plan", "user1", "pro")
|
|
assert isinstance(result, PaymentOrder)
|
|
assert result.order_id == "order1"
|
|
assert result.pay_url == "weixin://wxpay/bizpayurl?pr=test123"
|
|
assert result.amount == 100.0
|
|
assert result.status == "pending"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_api_failure(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = "https://example.com/callback"
|
|
gw = WeChatPayGateway()
|
|
|
|
import requests as req
|
|
with patch("app.services.payment.wechat_pay.requests.post", side_effect=req.RequestException("timeout")):
|
|
with pytest.raises(RuntimeError, match="WeChat Pay create_order failed"):
|
|
await gw.create_order("order1", 100.0, "test", "user1", "pro")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_order_calls_api(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {
|
|
"trade_state": "SUCCESS",
|
|
"amount": {"total": 10000},
|
|
}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("app.services.payment.wechat_pay.requests.get", return_value=mock_resp):
|
|
result = await gw.query_order("order1")
|
|
assert result.status == "completed"
|
|
assert result.amount == 100.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refund_calls_api(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"status": "SUCCESS"}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("app.services.payment.wechat_pay.requests.post", return_value=mock_resp):
|
|
result = await gw.refund("order1", 100.0, "reason")
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_callback_bad_signature(self):
|
|
with patch("app.services.payment.wechat_pay.settings") as mock_settings:
|
|
mock_settings.WECHAT_PAY_MCH_ID = "1234567890"
|
|
mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough"
|
|
mock_settings.WECHAT_PAY_APP_ID = "wx1234567890"
|
|
mock_settings.WECHAT_PAY_CERT_PATH = ""
|
|
mock_settings.WECHAT_PAY_NOTIFY_URL = ""
|
|
gw = WeChatPayGateway()
|
|
|
|
result = await gw.verify_callback({
|
|
"timestamp": "1234567890",
|
|
"nonce": "abc",
|
|
"body": "test",
|
|
"signature": "wrong_signature",
|
|
})
|
|
assert result.status == "failed"
|
|
|
|
|
|
class TestAlipayGatewayUnconfigured:
|
|
def test_is_configured_false(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = ""
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = ""
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = ""
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gw = AlipayGateway()
|
|
assert gw._is_configured() is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_fallback_to_mock(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = ""
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = ""
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = ""
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gw = AlipayGateway()
|
|
result = await gw.create_order("order1", 100.0, "test", "user1", "pro")
|
|
assert isinstance(result, PaymentOrder)
|
|
assert "mock://pay/" in result.pay_url
|
|
|
|
|
|
class TestAlipayGatewayConfigured:
|
|
def test_is_configured_true(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = "alipay_2021"
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem"
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem"
|
|
mock_settings.ALIPAY_NOTIFY_URL = "https://example.com/callback"
|
|
gw = AlipayGateway()
|
|
assert gw._is_configured() is True
|
|
|
|
def test_build_sign_content(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = "alipay_2021"
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem"
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem"
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gw = AlipayGateway()
|
|
content = gw._build_sign_content({"b": "2", "a": "1", "c": "3"})
|
|
assert content == "a=1&b=2&c=3"
|
|
|
|
def test_build_common_params(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = "alipay_2021"
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem"
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem"
|
|
mock_settings.ALIPAY_NOTIFY_URL = "https://example.com/notify"
|
|
gw = AlipayGateway()
|
|
params = gw._build_common_params("alipay.trade.wap.pay")
|
|
assert params["app_id"] == "alipay_2021"
|
|
assert params["method"] == "alipay.trade.wap.pay"
|
|
assert params["sign_type"] == "RSA2"
|
|
assert params["charset"] == "utf-8"
|
|
assert params["version"] == "1.0"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_callback_bad_signature(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = "alipay_2021"
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem"
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem"
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gw = AlipayGateway()
|
|
|
|
with patch.object(gw, "_rsa2_verify", return_value=False):
|
|
result = await gw.verify_callback({
|
|
"sign": "bad_sign",
|
|
"sign_type": "RSA2",
|
|
"out_trade_no": "order1",
|
|
"trade_no": "trade1",
|
|
"total_amount": "100.00",
|
|
"trade_status": "TRADE_SUCCESS",
|
|
})
|
|
assert result.status == "failed"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_callback_good_signature(self):
|
|
with patch("app.services.payment.alipay.settings") as mock_settings:
|
|
mock_settings.ALIPAY_APP_ID = "alipay_2021"
|
|
mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem"
|
|
mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem"
|
|
mock_settings.ALIPAY_NOTIFY_URL = ""
|
|
gw = AlipayGateway()
|
|
|
|
with patch.object(gw, "_rsa2_verify", return_value=True):
|
|
result = await gw.verify_callback({
|
|
"sign": "good_sign",
|
|
"sign_type": "RSA2",
|
|
"out_trade_no": "order1",
|
|
"trade_no": "trade1",
|
|
"total_amount": "100.00",
|
|
"trade_status": "TRADE_SUCCESS",
|
|
})
|
|
assert result.status == "success"
|
|
assert result.order_id == "order1"
|
|
assert result.payment_id == "trade1"
|
|
assert result.amount == 100.0
|