yml2213
0b9a01ab3f
- 限购商品（精英宝典 buy_limit=1）挂着未支付旧单时，真正先失败的是购买权限 校验 checkUserBuyAuth（真机 703「已超过该商品的购买上限，下单失败~」）， createOrderV5 根本不会执行；现在 checkUserBuyAuth 与 createOrderV5 两处失败 都会回退 find_reusable_order，拿同一商品的未支付旧单继续 payOrderSubmitV5。 - 新增支付宝收银台直达地址解析（resolve_pay_entry_url）：跟随网关 302 后的 excashier.alipay.com/standard/auth.htm?payOrderId=…（93 字符，原始网关地址 1616 字符）写入结果 pay_cashier_url；解析失败回落本站短链 /api/huya/pay/<task_id>?k=<hmac>，后端跑在 localhost 时客户也扫得开。 - 前端 QRCode 显式 color="#000" bgColor="#fff"：antd v6 暗色主题默认 color=token.colorText、bgColor=transparent，白码落在白底容器里根本看不见。 - 支付成功后列表「二维码」列不再展示二维码（qrDoneLabel → 已支付/已开通）。 测试：pytest 767 passed / 6 skipped；typecheck、eslint、ruff 通过。
182 lines
6.1 KiB
Python
182 lines
6.1 KiB
Python
"""开通支付码二维码内容：1600+ 字符的支付地址 → 手机扫得动、打得开的地址。
|
||
|
||
生产实测（2026-10-15）：`payOrderSubmitV5` 返回的支付宝网关地址有 **1626** 字符
|
||
（`https://openapi.alipay.com/gateway.do?...&sign=...`）。直接把它编成二维码是
|
||
version 29+（140+ 个模块），在 150px 的画面上每个模块只有约 1 像素 —— 手机扫不上。
|
||
|
||
现在的优先级：
|
||
|
||
1. **支付宝收银台直达地址**（`resolve_pay_entry_url`）：网关地址 302 两次后落到
|
||
`https://excashier.alipay.com/standard/auth.htm?payOrderId=...`（约 74 字符），
|
||
公网域名，后端跑在 localhost 时客户也扫得开；
|
||
2. **本站短链** `/api/huya/pay/<task_id>?k=<hmac>`（约 38 字符）：后端 302 跳到网关地址，
|
||
但要求本页域名手机能访问到。
|
||
"""
|
||
|
||
import pytest
|
||
from fastapi import HTTPException
|
||
|
||
from web.backend.models import HuyaAccount, HuyaTask, User
|
||
from web.backend.routers.huya.pay import redirect_huya_pay
|
||
from web.backend.services import huya_pay_link
|
||
from web.backend.services.huya_pay_link import (
|
||
PAY_PATH_PREFIX,
|
||
is_alipay_url,
|
||
pay_qr_path,
|
||
pay_token,
|
||
resolve_pay_entry_url,
|
||
verify_pay_token,
|
||
)
|
||
|
||
# 真机取到的支付宝网关地址形态（长度同量级）
|
||
ALIPAY_LONG_URL = (
|
||
"https://openapi.alipay.com/gateway.do"
|
||
"?alipay_sdk=alipay-sdk-java-4.40.272.ALL&app_id=2017123456789012"
|
||
"&method=alipay.trade.page.pay&sign=" + "a" * 1500
|
||
)
|
||
# 真机实测：上面的网关地址 302 两次后的落点（任务 191 原样取回）
|
||
CASHIER_URL = (
|
||
"https://excashier.alipay.com/standard/auth.htm"
|
||
"?payOrderId=6cb86f1d259642ac843522b85dab8a52.25"
|
||
)
|
||
|
||
|
||
def _create_pay_task(db, pay_url: str) -> HuyaTask:
|
||
user = User(username="pay-link-test", password_hash="x", role="support")
|
||
db.add(user)
|
||
db.commit()
|
||
account = HuyaAccount(uid="1199665170519", cookie="yyuid=1199665170519", tag="")
|
||
db.add(account)
|
||
db.commit()
|
||
task = HuyaTask(
|
||
batch_id="test-batch",
|
||
account_id=account.id,
|
||
task_type="create_recharge_order",
|
||
handbook_scope="elite",
|
||
status="running",
|
||
created_by=user.id,
|
||
result={"pay_url": pay_url} if pay_url else {},
|
||
)
|
||
db.add(task)
|
||
db.commit()
|
||
return task
|
||
|
||
|
||
def test_pay_token_round_trip_and_rejects_others():
|
||
token = pay_token(587)
|
||
|
||
assert len(token) == 16
|
||
assert verify_pay_token(587, token) is True
|
||
# 大小写不同就是另一个签名（hex 严格比对）
|
||
assert verify_pay_token(587, token.upper()) is False
|
||
assert verify_pay_token(587, "") is False
|
||
assert verify_pay_token(588, token) is False
|
||
|
||
|
||
def test_pay_qr_path_is_short_enough_to_scan():
|
||
path = pay_qr_path(587)
|
||
|
||
assert path.startswith(f"{PAY_PATH_PREFIX}/587?k=")
|
||
# 短链必须足够短：36 字符 → 二维码约 30 个模块（长链是 140+ 个模块）
|
||
assert len(path) <= 48
|
||
# 原始支付地址长度对比（就是它导致扫不上）
|
||
assert len(ALIPAY_LONG_URL) > 1000
|
||
|
||
|
||
def test_pay_redirect_route_returns_pay_url(db):
|
||
task = _create_pay_task(db, ALIPAY_LONG_URL)
|
||
|
||
response = redirect_huya_pay(task.id, pay_token(task.id), db)
|
||
|
||
assert response.status_code == 302
|
||
assert response.headers["location"] == ALIPAY_LONG_URL
|
||
|
||
|
||
def test_pay_redirect_route_rejects_bad_token(db):
|
||
task = _create_pay_task(db, ALIPAY_LONG_URL)
|
||
|
||
with pytest.raises(HTTPException) as excinfo:
|
||
redirect_huya_pay(task.id, "deadbeefdeadbeef", db)
|
||
|
||
assert excinfo.value.status_code == 404
|
||
|
||
|
||
def test_pay_redirect_route_rejects_task_without_pay_url(db):
|
||
task = _create_pay_task(db, "")
|
||
|
||
with pytest.raises(HTTPException) as excinfo:
|
||
redirect_huya_pay(task.id, pay_token(task.id), db)
|
||
|
||
assert excinfo.value.status_code == 404
|
||
|
||
|
||
class _FakeResponse:
|
||
"""假的 requests 响应：只需要 ``url`` 和 ``close()``。"""
|
||
|
||
def __init__(self, url: str):
|
||
self.url = url
|
||
self.closed = False
|
||
|
||
def close(self) -> None:
|
||
self.closed = True
|
||
|
||
|
||
def _patch_get(monkeypatch, *, final_url: str | None = None, error=None):
|
||
calls: list[dict] = []
|
||
|
||
def fake_get(url, **kwargs):
|
||
calls.append({"url": url, **kwargs})
|
||
if error is not None:
|
||
raise error
|
||
return _FakeResponse(final_url or "")
|
||
|
||
monkeypatch.setattr(huya_pay_link.requests, "get", fake_get)
|
||
return calls
|
||
|
||
|
||
def test_resolve_pay_entry_url_follows_redirect_to_cashier(monkeypatch):
|
||
calls = _patch_get(monkeypatch, final_url=CASHIER_URL)
|
||
|
||
assert resolve_pay_entry_url(ALIPAY_LONG_URL) == CASHIER_URL
|
||
assert calls[0]["url"] == ALIPAY_LONG_URL
|
||
# 必须跟随 302，否则拿到的是网关页自己
|
||
assert calls[0]["allow_redirects"] is True
|
||
assert "Mozilla" in calls[0]["headers"]["User-Agent"]
|
||
|
||
|
||
def test_resolve_pay_entry_url_rejects_non_alipay_target(monkeypatch):
|
||
"""落到非支付宝域名就当没解析出来（调用方回落到本站短链）。"""
|
||
_patch_get(monkeypatch, final_url="https://evil.example.com/cashier")
|
||
|
||
assert resolve_pay_entry_url(ALIPAY_LONG_URL) == ""
|
||
|
||
|
||
def test_resolve_pay_entry_url_swallows_errors(monkeypatch):
|
||
_patch_get(monkeypatch, error=RuntimeError("network down"))
|
||
|
||
assert resolve_pay_entry_url(ALIPAY_LONG_URL) == ""
|
||
|
||
|
||
def test_resolve_pay_entry_url_ignores_blank_or_relative():
|
||
assert resolve_pay_entry_url("") == ""
|
||
assert resolve_pay_entry_url("/api/huya/pay/191?k=x") == ""
|
||
|
||
|
||
def test_resolve_pay_entry_url_is_shorter_than_raw_url(monkeypatch):
|
||
_patch_get(monkeypatch, final_url=CASHIER_URL)
|
||
|
||
resolved = resolve_pay_entry_url(ALIPAY_LONG_URL)
|
||
|
||
# 收银台直达地址约 74 字符 → 二维码模块数远低于原始地址
|
||
assert len(resolved) < 120
|
||
assert len(resolved) < len(ALIPAY_LONG_URL)
|
||
|
||
|
||
def test_is_alipay_url_matches_only_alipay_domains():
|
||
assert is_alipay_url(CASHIER_URL) is True
|
||
assert is_alipay_url("https://openapi.alipay.com/gateway.do?x=1") is True
|
||
assert is_alipay_url("https://alipay.com/") is True
|
||
assert is_alipay_url("https://alipay.com.evil.com/") is False
|
||
assert is_alipay_url("https://excashier.alipay.com.evil.com/") is False
|
||
assert is_alipay_url("") is False
|