Pine
7a9741a333
- profitsharing.py：submit_applyment / query_applyment / upload_image（复用服务商 API 证书加密敏感字段）
- service.py：create_merchant_applyment（幂等：已有 applying/active 商户绑定返回已有）；query_merchant_applyment（FINISHED 自动回填 sub_mchid + 添加 MERCHANT_ID 接收方激活；REJECTED 标记+原因）
- routers.py：POST upload-media / POST applyment / GET {binding_id}/applyment
- payment_bindings 增加 detail 列（0040_payment_binding_detail，接在 0040_incubator 之后）
- 冒烟测试覆盖进件闭环（提交幂等/通过自动激活/驳回）
171 lines
8.8 KiB
Python
171 lines
8.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""微信服务商分账改造 · 冒烟测试（临时 sqlite，不碰主库）。
|
||
|
||
覆盖：
|
||
1. release_escrow：未接微信收单/分账未配置 → 降级台账结算（channel=manual, released）
|
||
2. release_escrow：配置齐备 + 接单者 active 绑定 → 发起微信分账（mock create_split）
|
||
3. handle_split_notify：回调确认 → escrow released + log shared（幂等）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
# 临时库：优先 sqlite 内存/临时文件，避免触碰 .env 指定的 MySQL 主库
|
||
_tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
||
_tmp.close()
|
||
os.environ["PINEAGENTS_DEMO_DATABASE_URL"] = f"sqlite+aiosqlite:///{_tmp.name}"
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from unittest.mock import patch # noqa: E402
|
||
|
||
from app.infrastructure.db import Base # noqa: E402
|
||
from app.infrastructure.repositories import Database # noqa: E402
|
||
import app.pay.models # noqa: E402,F401 确保新模型注册
|
||
import app.infrastructure.models # noqa: E402,F401
|
||
from app.services.settlement_service import SettlementService # noqa: E402
|
||
|
||
|
||
async def _setup(db: Database):
|
||
async with db._engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
|
||
async def _mk_completed_task(db, claimer: str = "") -> dict:
|
||
return await db.tasks.create({
|
||
"task_code": "smk001", "title": "冒烟任务", "status": "completed",
|
||
"budget_min": 1000, "budget_max": 1000, "claimed_by": claimer or None,
|
||
})
|
||
|
||
|
||
async def main() -> None:
|
||
db = Database()
|
||
await _setup(db)
|
||
|
||
# ── 1) 未配置分账 / 无收单 → 降级台账结算 ────────────────────────────────
|
||
task = await _mk_completed_task(db)
|
||
svc = SettlementService(db)
|
||
r = await svc.release_escrow(task["id"])
|
||
print("① 降级台账:", r["status"], r["channel"], r["share_status"])
|
||
assert r["status"] == "released" and r["channel"] == "manual"
|
||
# 直接查库验证流水
|
||
import sqlite3
|
||
conn = sqlite3.connect(_tmp.name)
|
||
row = conn.execute("SELECT channel,status,detail FROM settlement_logs ORDER BY created_at DESC LIMIT 1").fetchone()
|
||
conn.close()
|
||
print("① 流水:", row)
|
||
assert row and row[0] == "manual" and row[1] == "shared"
|
||
|
||
# ── 2) 配置齐备 + 接单者 active 绑定 → 发起分账 ─────────────────────────
|
||
from app.pay.repository import PaymentBindingRepository
|
||
# 造一个接单者用户 + 商户绑定
|
||
user = await db.users.create("smk_opc", "pw123456", nickname="接单者", role="opc_member")
|
||
bind = await PaymentBindingRepository(db.session).create({
|
||
"user_id": user["id"], "bind_type": "merchant",
|
||
"sub_mchid": "1900000109", "status": "active",
|
||
})
|
||
task2 = await _mk_completed_task(db)
|
||
await db.tasks.claim(task2["id"], user["id"]) # 接单者=用户
|
||
await db.tasks.set_status(task2["id"], "completed") # 接单后置完成（验收通过）
|
||
esc = await db.escrows.create(task2["id"], task2["title"], 1000, 50)
|
||
await db.escrows.update_fields(esc["id"], {"transaction_id": "4200001234"})
|
||
with patch.object(
|
||
__import__("app.pay.config", fromlist=["profitsharing_enabled"]),
|
||
"profitsharing_enabled", return_value=True,
|
||
), patch("app.pay.profitsharing.create_split") as mock_split:
|
||
r2 = await svc.release_escrow(task2["id"])
|
||
mock_split.assert_called_once()
|
||
call = mock_split.call_args.kwargs
|
||
print("② 分账发起: channel=%s share_status=%s" % (r2["channel"], r2["share_status"]))
|
||
print("② 接收方:", [(x["type"], x["amount"]) for x in call["receivers"]])
|
||
assert r2["channel"] == "wx_split" and r2["share_status"] == "sharing"
|
||
assert call["receivers"][0]["amount"] == 95000 and call["receivers"][1]["amount"] == 5000
|
||
assert call["receivers"][0]["type"] == "MERCHANT_ID" # 商户绑定 → MERCHANT_ID
|
||
share_no = call["out_order_no"]
|
||
|
||
# ── 3) 回调确认（幂等）───────────────────────────────────────────────────
|
||
ok1 = await svc.handle_split_notify({"out_order_no": share_no})
|
||
conn = sqlite3.connect(_tmp.name)
|
||
esc_state = conn.execute("SELECT status,share_status FROM escrows WHERE id=?", (esc["id"],)).fetchone()
|
||
log_state = conn.execute("SELECT status FROM settlement_logs WHERE share_order_no=?", (share_no,)).fetchone()
|
||
conn.close()
|
||
print("③ 回调后 escrow:", esc_state, "log:", log_state)
|
||
assert ok1 and esc_state == ("released", "shared") and log_state == ("shared",)
|
||
ok2 = await svc.handle_split_notify({"out_order_no": share_no}) # 重放
|
||
assert ok2 # 幂等不报错
|
||
print("③ 幂等重放 OK")
|
||
|
||
# ── 4) 特约商户进件闭环（mock 微信侧；用独立用户避开 ② 的 active 绑定）───
|
||
from app.pay import service as pay_service
|
||
user2 = await db.users.create("smk_opc2", "pw123456", nickname="进件方", role="opc_member")
|
||
applyment_body = {
|
||
"contact_info": {"contact_name": "张三", "mobile_phone": "13900000000"},
|
||
"subject_info": {"subject_type": "SUBJECT_TYPE_ENTERPRISE",
|
||
"business_license_info": {"merchant_name": "测试公司",
|
||
"legal_person": "张三"}},
|
||
"business_info": {"merchant_shortname": "测试商户"},
|
||
"settlement_info": {"settlement_id": "719"},
|
||
"bank_account_info": {"account_name": "测试公司", "account_bank": "工商银行",
|
||
"bank_address_code": "110000", "account_number": "1234567890"},
|
||
}
|
||
with patch.object(
|
||
__import__("app.pay.config", fromlist=["profitsharing_enabled"]),
|
||
"profitsharing_enabled", return_value=True,
|
||
), patch("app.pay.profitsharing.submit_applyment",
|
||
return_value={"applyment_id": "2000001234567890",
|
||
"applyment_state": "AUDITING"}) as mock_sub:
|
||
r4 = await pay_service.create_merchant_applyment(db, user2, **applyment_body)
|
||
print("④ 进件提交: state=%s binding.status=%s applyment_id=%s"
|
||
% (r4["applyment_state"], r4["binding"]["status"], r4["applyment_id"]))
|
||
assert r4["binding"]["status"] == "applying"
|
||
mock_sub.assert_called_once()
|
||
# 幂等：重复提交返回已有，不二次调微信
|
||
r4b = await pay_service.create_merchant_applyment(db, user2, **applyment_body)
|
||
assert mock_sub.call_count == 1 and r4b["applyment_state"] == "EXISTING"
|
||
print("④ 幂等：重复进件被拦截（微信仅调用 1 次）")
|
||
binding_id = r4["binding"]["id"]
|
||
|
||
# 4b) 查询进件：审核通过 → 回填 sub_mchid + 添加接收方 → active
|
||
with patch.object(
|
||
__import__("app.pay.config", fromlist=["profitsharing_enabled"]),
|
||
"profitsharing_enabled", return_value=True,
|
||
), patch("app.pay.profitsharing.query_applyment",
|
||
return_value={"applyment_state": "FINISHED", "sub_mchid": "1900000109"}), \
|
||
patch("app.pay.profitsharing.add_receiver", return_value={}) as mock_add:
|
||
r5 = await pay_service.query_merchant_applyment(db, user2, binding_id)
|
||
print("④ 进件通过: state=%s sub_mchid=%s binding.status=%s"
|
||
% (r5["applyment_state"], r5["sub_mchid"], r5["binding"]["status"]))
|
||
assert r5["sub_mchid"] == "1900000109" and r5["binding"]["status"] == "active"
|
||
mock_add.assert_called_once_with(account_type="MERCHANT_ID", account="1900000109")
|
||
print("④ 自动添加 MERCHANT_ID 接收方 OK")
|
||
|
||
# 4c) 驳回场景：新开一个绑定（绕开幂等）→ REJECTED 标记
|
||
with patch.object(
|
||
__import__("app.pay.config", fromlist=["profitsharing_enabled"]),
|
||
"profitsharing_enabled", return_value=True,
|
||
), patch("app.pay.profitsharing.query_applyment",
|
||
return_value={"applyment_state": "REJECTED",
|
||
"audit_detail": "营业执照信息不清晰"}) as mock_q:
|
||
# 直接落一条 applying 记录再查
|
||
b_rej = await PaymentBindingRepository(db.session).create({
|
||
"user_id": user2["id"], "bind_type": "merchant",
|
||
"applyment_id": "2000000000000001", "status": "applying"})
|
||
r6 = await pay_service.query_merchant_applyment(db, user2, b_rej["id"])
|
||
print("④ 进件驳回: state=%s status=%s detail=%s"
|
||
% (r6["applyment_state"], r6["binding"]["status"], r6["binding"]["detail"]))
|
||
assert r6["binding"]["status"] == "rejected" and "营业执照" in r6["binding"]["detail"]
|
||
mock_q.assert_called()
|
||
|
||
print("\n✅ 冒烟测试全部通过")
|
||
await db.close()
|
||
os.unlink(_tmp.name)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|