feat: add alipay simulation verification
Some checks failed
CI / pytest (Python 3.10) (push) Has been cancelled
CI / pytest (Python 3.11) (push) Has been cancelled
CI / pytest (Python 3.12) (push) Has been cancelled

This commit is contained in:
Hermes Agent
2026-06-14 07:48:17 +08:00
parent e7706e3b11
commit 85605716b7
9 changed files with 176 additions and 4 deletions

View File

@@ -1,6 +1,9 @@
GAOKAO_PAYMENT_PROVIDER=mock
GAOKAO_PAYMENT_BASE_URL=https://example.com
GAOKAO_PAYMENT_WEBHOOK_SECRET=replace-with-independent-payment-webhook-secret
# 上线前可切到 alipay_sim 做模拟支付宝 checkout / webhook 验证
# GAOKAO_PAYMENT_PROVIDER=alipay_sim
GAOKAO_PAYMENT_NOTIFY_URL=
GAOKAO_PAYMENT_RETURN_URL=
GAOKAO_PAYMENT_APP_ID=

View File

@@ -21,6 +21,7 @@ class ProviderReadinessReport:
_PROVIDER_REQUIREMENTS: dict[str, list[str]] = {
"mock": [],
"alipay_sim": [],
"alipay": [
"GAOKAO_PAYMENT_APP_ID",
"GAOKAO_PAYMENT_PRIVATE_KEY_PATH",
@@ -70,6 +71,11 @@ def build_provider_readiness_report(
if normalized == "mock":
report.notes.append("mock provider 仅用于本地/测试闭环,不代表真实收款上线")
elif normalized == "alipay_sim":
report.notes.extend([
"alipay_sim 用于上线前模拟支付宝支付/回调验证,不代表真实支付宝已联通。",
"上线前仍必须以真实 app_id/证书/notify_url 完成一次真实支付宝测试。",
])
elif normalized == "alipay":
report.notes.extend([
"建议先用支付宝作为首个真实 Web 支付 provider当前架构更贴合 checkout_url 跳转模式。",

View File

@@ -1,3 +1,4 @@
from .alipay_sim import AlipaySimProvider
from .mock_gateway import MockPaymentProvider
__all__ = ["MockPaymentProvider"]
__all__ = ["MockPaymentProvider", "AlipaySimProvider"]

View File

@@ -0,0 +1,67 @@
from __future__ import annotations
import hashlib
import hmac
import json
from decimal import Decimal, ROUND_HALF_UP
from typing import Any
class AlipaySimProvider:
name = "alipay_sim"
def __init__(self, *, base_url: str, secret: str) -> None:
self.base_url = base_url.rstrip("/")
self.secret = secret
def build_checkout_url(self, payment_id: str, portal_token: str) -> str:
return f"/pay/alipay-sim/{payment_id}?token={portal_token}"
def build_webhook_request(
self,
*,
payment_id: str,
amount_cents: int,
provider_trade_no: str,
) -> tuple[dict[str, Any], dict[str, str]]:
payload = {
"out_trade_no": payment_id,
"trade_no": provider_trade_no,
"total_amount": self._format_amount(amount_cents),
"trade_status": "TRADE_SUCCESS",
}
sig = self.sign_payload(payload)
return payload, {"X-Alipay-Sim-Signature": sig}
def normalize_webhook_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
total_amount = str(payload.get("total_amount") or "0")
amount_cents = int(
(
Decimal(total_amount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
* 100
)
)
return {
"payment_id": str(payload.get("out_trade_no") or ""),
"amount_cents": amount_cents,
"provider_trade_no": str(payload.get("trade_no") or ""),
"status": str(payload.get("trade_status") or "TRADE_SUCCESS"),
}
def sign_payload(self, payload: dict[str, Any]) -> str:
body = json.dumps(
payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
)
return hmac.new(
self.secret.encode("utf-8"),
body.encode("utf-8"),
hashlib.sha256,
).hexdigest()
def verify_signature(self, payload: dict[str, Any], signature: str) -> bool:
expected = self.sign_payload(payload)
return hmac.compare_digest(expected, signature)
@staticmethod
def _format_amount(amount_cents: int) -> str:
return f"{Decimal(amount_cents) / Decimal(100):.2f}"

View File

@@ -32,6 +32,14 @@ class MockPaymentProvider:
sig = self.sign_payload(payload)
return payload, {"X-Mock-Signature": sig}
def normalize_webhook_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
return {
"payment_id": str(payload.get("payment_id") or ""),
"amount_cents": int(payload.get("amount_cents") or 0),
"provider_trade_no": str(payload.get("provider_trade_no") or ""),
"status": str(payload.get("status") or "paid"),
}
def sign_payload(self, payload: dict[str, Any]) -> str:
body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
return hmac.new(

View File

@@ -12,6 +12,7 @@ from data.payments.models import (
RefundRequestResult,
WebhookHandleResult,
)
from data.payments.providers.alipay_sim import AlipaySimProvider
from data.payments.providers.mock_gateway import MockPaymentProvider
@@ -23,6 +24,8 @@ def _build_provider(*, provider_name: str, base_url: str, webhook_secret: str):
normalized = (provider_name or "mock").strip().lower()
if normalized == "mock":
return MockPaymentProvider(base_url=base_url, secret=webhook_secret)
if normalized == "alipay_sim":
return AlipaySimProvider(base_url=base_url, secret=webhook_secret)
if normalized == "alipay":
raise PaymentError(
"alipay provider 尚未实装;请先完成 docs/PAYMENT_PROVIDER_ONBOARDING.md 中的真实接入项"
@@ -122,7 +125,8 @@ class PaymentService:
def handle_webhook(self, payload: dict, signature: str) -> WebhookHandleResult:
if not self.provider.verify_signature(payload, signature):
raise PaymentError("invalid payment signature")
payment_id = str(payload.get("payment_id") or "")
normalized_payload = self.provider.normalize_webhook_payload(payload)
payment_id = str(normalized_payload.get("payment_id") or "")
if not payment_id:
raise PaymentError("missing payment_id")
payments = PaymentDAO.for_db(self.db_path)
@@ -130,7 +134,10 @@ class PaymentService:
payment = payments.get(payment_id)
if payment is None:
raise PaymentError("payment not found")
if int(payload.get("amount_cents") or -1) != payment.amount_cents:
if (
int(normalized_payload.get("amount_cents") or -1)
!= payment.amount_cents
):
raise PaymentError("payment amount mismatch")
if payment.status == "paid":
with OrdersDAO.connect(self.db_path) as orders_dao:
@@ -145,7 +152,9 @@ class PaymentService:
payment.id,
status="paid",
provider_trade_no=str(
payload.get("provider_trade_no") or payment.provider_trade_no or ""
normalized_payload.get("provider_trade_no")
or payment.provider_trade_no
or ""
),
callback_payload=payload,
)

View File

@@ -0,0 +1,63 @@
from __future__ import annotations
from data.orders.dao import OrdersDAO
from data.orders.models import Order
from data.payments.service import PaymentService
def _seed_order(db_path: str, order_id: str = "GKO-20260614-ALIPAY-SIM") -> Order:
order = Order(
id=order_id,
source="web",
service_version="standard",
amount_cents=9900,
status="pending",
customer_name="张家长",
customer_phone="13800138000",
candidate_name="张三",
candidate_province="湖南",
)
with OrdersDAO.connect(db_path) as dao:
return dao.create(order, actor="test", reason="seed")
def test_alipay_sim_provider_checkout_shape_and_idempotency(settings):
order = _seed_order(settings.orders_db_path)
service = PaymentService.for_db(
settings.orders_db_path,
base_url=settings.payment_base_url,
webhook_secret=settings.payment_webhook_secret,
provider_name="alipay_sim",
)
first = service.create_checkout(order.id, portal_token="portal-token")
second = service.create_checkout(order.id, portal_token="portal-token")
assert first.payment_id == second.payment_id
assert first.provider == "alipay_sim"
assert first.checkout_url.endswith(
f"/pay/alipay-sim/{first.payment_id}?token=portal-token"
)
def test_alipay_sim_webhook_marks_order_paid(settings):
order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-ALIPAY-WEBHOOK")
service = PaymentService.for_db(
settings.orders_db_path,
base_url=settings.payment_base_url,
webhook_secret=settings.payment_webhook_secret,
provider_name="alipay_sim",
)
checkout = service.create_checkout(order.id, portal_token="portal-token")
payload, headers = service.provider.build_webhook_request(
payment_id=checkout.payment_id,
amount_cents=order.amount_cents,
provider_trade_no="ALIPAY-SIM-001",
)
handled = service.handle_webhook(payload, headers["X-Alipay-Sim-Signature"])
assert handled.idempotent is False
assert handled.order_status == "paid"
again = service.handle_webhook(payload, headers["X-Alipay-Sim-Signature"])
assert again.idempotent is True

View File

@@ -10,6 +10,13 @@ def test_mock_provider_report_is_ready_without_credentials():
assert report.missing_env_vars == []
def test_alipay_sim_report_is_ready_without_real_credentials():
report = build_provider_readiness_report("alipay_sim", env={})
assert report.provider == "alipay_sim"
assert report.ready is True
assert report.missing_env_vars == []
def test_alipay_provider_report_detects_missing_credentials():
report = build_provider_readiness_report("alipay", env={})
assert report.provider == "alipay"

View File

@@ -15,6 +15,7 @@
## 当前代码口径
- 本地闭环 provider: `mock`
- 上线前模拟支付宝 provider: `alipay_sim`
- 配置键:
- `GAOKAO_PAYMENT_PROVIDER`
- `GAOKAO_PAYMENT_BASE_URL`
@@ -36,6 +37,7 @@ python3 scripts/payment_provider_doctor.py
### 期望结果
- `provider=mock`ready=true只代表本地/测试闭环
- `provider=alipay_sim`ready=true代表“上线前模拟支付宝 checkout/回调验证”已可本地执行,但不代表真实支付宝已联通
- `provider=alipay`:若缺少 app_id / key_path / notify_url / return_url会返回 ready=false
## 支付宝最小上线前置
@@ -71,6 +73,12 @@ export GAOKAO_PAYMENT_ALIPAY_PUBLIC_KEY_PATH=/secure/path/alipay_public.pem
## 真正完成 T12 的下一跳
当上面四类外部条件具备前,建议先做:
1. `GAOKAO_PAYMENT_PROVIDER=alipay_sim python3 scripts/payment_provider_doctor.py`
2. `pytest -q data/payments/tests/test_provider_alipay_sim.py`
3.`alipay_sim` 跑本地 checkout / webhook 模拟验收
当上面四类外部条件具备后,继续执行:
1. 实装支付宝 provider