41 lines
1.7 KiB
Python
41 lines
1.7 KiB
Python
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app import alipay_provider
|
|
|
|
|
|
def test_alipay_callback_reconciles_verified_success(monkeypatch):
|
|
monkeypatch.setattr(alipay_provider, "verify_notification", lambda params: params.get("sign") == "valid")
|
|
out_biz_no = f"RF{uuid4().hex[:20]}"
|
|
with TestClient(main.app) as client:
|
|
with main.SessionLocal() as db:
|
|
user = db.scalar(main.select(main.AdminUser).where(main.AdminUser.username == "operator"))
|
|
order = main.AlipayRefundOrder(
|
|
target_kind="answer_submission", target_id=int(uuid4().int % 1_000_000_000),
|
|
out_biz_no=out_biz_no, amount_fen=300, account="buyer@example.com",
|
|
payee_name="测试收款人", created_by_id=user.id, status="pending",
|
|
)
|
|
db.add(order)
|
|
db.commit()
|
|
order_id = order.id
|
|
response = client.post("/alipay_callback", data={
|
|
"sign": "valid", "out_biz_no": out_biz_no, "status": "SUCCESS",
|
|
"order_id": "ALI-CALLBACK-001",
|
|
})
|
|
assert response.status_code == 200
|
|
assert response.text == "success"
|
|
with main.SessionLocal() as db:
|
|
order = db.get(main.AlipayRefundOrder, order_id)
|
|
assert order.status == "success"
|
|
assert order.alipay_trade_no == "ALI-CALLBACK-001"
|
|
|
|
|
|
def test_alipay_callback_rejects_invalid_signature(monkeypatch):
|
|
monkeypatch.setattr(alipay_provider, "verify_notification", lambda _params: False)
|
|
with TestClient(main.app) as client:
|
|
response = client.post("/alipay_callback", data={"sign": "bad", "out_biz_no": "RF-BAD"})
|
|
assert response.status_code == 400
|
|
assert response.text == "failure"
|