✅ 本节目标

我们将实现一个:

🧪 基于 Flask 的本地 Mock 支付回调服务

📡 可接收并验证签名的回调请求

🧾 返回标准响应格式(如微信/支付宝)

🔄 支持不同场景(支付成功、失败、超时等)‍

📦 所需依赖

确保你已安装以下库:

pip install flask pytest respx requests‍

📁 推荐目录结构


  1. payment_automation/

  2. ├── mock_server/

  3. │ ├── app.py # 回调服务主程序

  4. │ └── config.py # 回调配置(如签名密钥)

  5. ├── testcases/

  6. │ └── test_payment_callback.py # 测试用例

  7. ├── utils/

  8. │ └── sign_utils.py # 签名工具类

  9. ├── config/

  10. │ └── payment_config.py # 支付配置

  11. └── requirements.txt‍

🔧 核心代码实现

1️⃣ 回调配置:mock_server/config.py

  1. # mock_server/config.py

  2. CALLBACK_CONFIG = {

  3. "host": "127.0.0.1",

  4. "port": 5000,

  5. "public_key": "third_party_public_key", # 第三方公钥(用于验签)

  6. "success_response": {"return_code": "SUCCESS", "return_msg": "OK"},

  7. "fail_response": {"return_code": "FAIL", "return_msg": "SIGNATURE_FAILED"}

  8. }

2️⃣ 签名验证工具类:utils/sign_utils.py

  1. # utils/sign_utils.py

  2. import hashlib

  3. import hmac

  4. from urllib.parse import quote_plus

  5. def generate_sign(params: dict, secret_key: str) -> str:

  6. """

  7. 生成签名(按 key 排序后拼接)

  8. """

  9. sorted_params = "&".join(f"{k}={quote_plus(str(v))}" for k, v in sorted(params.items()))

  10. sign = hmac.new(secret_key.encode(), digestmod=hashlib.sha256)

  11. sign.update(sorted_params.encode())

  12. return sign.hexdigest()

  13. def verify_sign(received_sign: str, params: dict, secret_key: str) -> bool:

  14. """

  15. 验证签名是否一致

  16. """

  17. expected_sign = generate_sign(params, secret_key)

  18. return hmac.compare_digest(expected_sign, received_sign)

3️⃣ Mock 回调服务:mock_server/app.py

  1. # mock_server/app.py

  2. from flask import Flask, request, jsonify

  3. from utils.sign_utils import verify_sign

  4. from mock_server.config import CALLBACK_CONFIG

  5. app = Flask(__name__)

  6. PUBLIC_KEY = CALLBACK_CONFIG["public_key"]

  7. SUCCESS_RESPONSE = CALLBACK_CONFIG["success_response"]

  8. FAIL_RESPONSE = CALLBACK_CONFIG["fail_response"]

  9. @app.route("/wechat/notify", methods=["POST"])

  10. def wechat_notify():

  11. data = request.json

  12. sign = data.pop("sign", None)

  13. if not sign or not verify_sign(sign, data, PUBLIC_KEY):

  14. return jsonify(FAIL_RESPONSE), 200

  15. # 模拟业务处理逻辑(更新订单状态)

  16. print("收到微信支付回调数据:", data)

  17. return jsonify(SUCCESS_RESPONSE), 200

  18. @app.route("/alipay/notify", methods=["POST"])

  19. def alipay_notify():

  20. data = request.form.to_dict()

  21. sign = data.pop("sign", None)

  22. if not sign or not verify_sign(sign, data, PUBLIC_KEY):

  23. return "failure"

  24. print("收到支付宝支付回调数据:", data)

  25. return "success"

  26. if __name__ == "__main__":

  27. app.run(host="0.0.0.0", port=5000)

启动方式:

  1. cd mock_server

  2. python app.py

🧪 使用示例:启动 Mock 服务 + 发送模拟回调

你可以使用 ngrok 或 localtunnel 将本地服务暴露为公网 URL:

ngrok http 5000

将 https://xxx.ngrok.io/wechat/notify 设置为你的支付平台回调地址即可。

📌 示例测试用例:testcases/test_payment_callback.py

  1. # testcases/test_payment_callback.py

  2. import requests

  3. import json

  4. import pytest

  5. from utils.sign_utils import generate_sign

  6. from mock_server.config import CALLBACK_CONFIG

  7. BASE_URL = "http://localhost:5000"

  8. SECRET_KEY = CALLBACK_CONFIG["public_key"]

  9. def send_wechat_notify(order_id: str, status: str = "SUCCESS"):

  10. data = {

  11. "transaction_id": "T1234567890",

  12. "out_trade_no": order_id,

  13. "return_code": status,

  14. "total_fee": 9990, # 单位分

  15. "time_end": "20250405120000"

  16. }

  17. data["sign"] = generate_sign(data, SECRET_KEY)

  18. response = requests.post(f"{BASE_URL}/wechat/notify", json=data)

  19. return response

  20. def test_wechat_callback_success():

  21. order_id = "ORDER123456"

  22. response = send_wechat_notify(order_id)

  23. assert response.status_code == 200

  24. assert response.json()["return_code"] == "SUCCESS"

  25. def test_wechat_callback_invalid_sign():

  26. data = {

  27. "transaction_id": "T1234567890",

  28. "out_trade_no": "ORDER789012",

  29. "return_code": "SUCCESS",

  30. "total_fee": 9990,

  31. "time_end": "20250405120000",

  32. "sign": "invalid-sign-here"

  33. }

  34. response = requests.post(f"{BASE_URL}/wechat/notify", json=data)

  35. assert response.status_code == 200

  36. assert response.json()["return_code"] == "FAIL"

🧩 进阶功能建议

功能  描述

✅ 多种支付结果模拟  如 success / failed / timeout

✅ 支持动态返回值   通过 URL 参数控制返回内容

✅ 支持数据库记录日志 存储每次回调请求详情

✅ 异步回调重试机制  模拟支付平台多次通知

✅ 结合 pytest fixtures    自动启动/关闭 Mock 服务‍

📦 启动命令汇总

# 启动 Mock 回调服务

cd mock_server

python app.py

# 在另一个终端运行测试

cd ..

pytest testcases/test_payment_callback.py -v‍

✅ 总结

这个 Mock 支付回调模拟器 具备以下优点:

特性  说明

✅ 安全签名验证    支持 HMAC-SHA256 签名校验

✅ 多支付平台支持   微信、支付宝回调均可模拟

✅ 易于扩展  可添加更多支付渠道和异常场景

✅ 与测试框架集成   支持 pytest 自动化断言

✅ 支持公网访问    配合 ngrok 可供外部调用‍

感谢每一个认真阅读我文章的人,礼尚往来总是要有的,虽然不是什么很值钱的东西,如果你用得到的话可以直接拿走:

这些资料,对于【软件测试】的朋友来说应该是最全面最完整的备战仓库,这个仓库也陪伴上万个测试工程师们走过最艰难的路程,希望也能帮助到你!有需要的小伙伴可以点击下方小卡片领取   

更多推荐