yml2213
bf1e2e9361
真机复现（2026-10-15）：上一版把二维码内容放成了**收银台页面地址**
（excashier.alipay.com/standard/auth.htm?payOrderId=…），那是 PC 版页面
（页面里写死 qrcodeWidth=150）。有些安卓机是用系统相机/浏览器打开的，就停在
这张 PC 页上——只显示左上角一个很小的二维码，不触发付款。
收银台页面里真正的扫码内容是隐藏字段 `J_qrCode`：
<input name="qrCode" value="https://qr.alipay.com/upx03110ghym9vm37sx7255c" id="J_qrCode"/>
- huya_pay_link 新增 PayTargets / resolve_pay_targets() / extract_pay_qr_code()：
跟随网关 302 取收银台页面并解析出 qr.alipay.com 扫码码（约 40 字符）。
- 出码时写入 pay_qr_code，保留 pay_cashier_url 兜底；拿不到扫码码会打 warning。
- 前端二维码优先级：pay_qr_code → pay_cashier_url → 本站短链 → 原始地址。
验证：全量 pytest 772 passed / 6 skipped；用真机那张收银台页面跑解析得到
https://qr.alipay.com/upx03110ghym9vm37sx7255c；typecheck / eslint / ruff 通过。
549 lines
20 KiB
Python
549 lines
20 KiB
Python
"""虎牙精英宝典开通状态与「一键查询」（2026-10-15 真机实测）。
|
||
|
||
判定来源与真机结论：
|
||
|
||
* 宝典开通状态 = 积分中心 ``shopMiddleUI.getRedeemNeedGoods`` 的 ``tag5``。
|
||
下面固化的两段响应字节分别来自已开通账号（hy_302016799）与未开通账号
|
||
（hy_260034554），与网页显示的「已开通精英宝典 / 获取积分需先开通精英宝典」
|
||
一致。
|
||
* 开通时间 = ``revenueWebUI.queryUserOrderList`` 里已支付的宝典订单。
|
||
"""
|
||
|
||
import json
|
||
from types import SimpleNamespace
|
||
|
||
from core.huya.frame_decoder import decode_struct_bytes
|
||
from core.huya.http_client import HuyaHttpClient
|
||
from core.huya.shop_structs import ShopRawRsp
|
||
from web.backend.models import HuyaAccount
|
||
from web.backend.services.huya_pay_link import PayTargets, pay_qr_path
|
||
from web.backend.services.huya_runner import HuyaBatchRunner
|
||
|
||
UID = 1199668595203
|
||
COOKIE = f"yyuid={UID}; udb_uid={UID}"
|
||
# 真机 getRedeemNeedGoods 响应：已开通（tag5=1、tag6=1、tag7 为空）
|
||
REDEEM_OPENED_HEX = "0a0100c8160ce69fa5e8afa2e68890e58a9f21011830144c50016001790c0b"
|
||
# 真机 getRedeemNeedGoods 响应：未开通（tag5=0、tag6=0、tag7 列出开通套餐）
|
||
REDEEM_NOT_OPENED_HEX = (
|
||
"0a0100c8160ce69fa5e8afa2e68890e58a9f2c301440145c6c7900040a02007f134f1200"
|
||
"765c0b260ce7b2bee88bb1e5ae9de585b830054101fb51012c610bb87c9001a90cb90cca010"
|
||
"bb8110bb82c310bb8490c5c610bb8790c8c9cacbcccdc0bf00f010b0a02007ede691200766"
|
||
"17f2616e5bc80e9809ae5ae9de585b82be7b2bee88bb1e5b88130054101fc51012d610bc2"
|
||
"70018900020a02007ede6912007f134f2200765c0b310bb841012c5001660ce7b2bee88bb1"
|
||
"e5ae9de585b87c0b0a02007ede6912007f10bc220076662f300a400150016609e7b2bee88b"
|
||
"b1e5b88170010b9ca90cb90cca010bc2110bc22c310bc2490c5c610bc2790c8c9cacbcccdc0b"
|
||
"fc0f0b"
|
||
)
|
||
# 真机订单列表里已支付的宝典订单（pay_time 与 spu 均来自真机响应）
|
||
PAID_ELITE_ORDER = {
|
||
"item_name": "精英宝典",
|
||
"order_status": 60,
|
||
"pay_time": 1791895396000,
|
||
"goods_detail": {"spu_id": "hy-8328015", "sku_id": 7756811},
|
||
}
|
||
UNPAID_ELITE_ORDER = {
|
||
"item_name": "精英宝典",
|
||
"order_status": 10,
|
||
"pay_time": 0,
|
||
"goods_detail": {"spu_id": "hy-8328015", "sku_id": 7756811},
|
||
}
|
||
OTHER_ORDER = {
|
||
"item_name": "电竞徽章",
|
||
"order_status": 60,
|
||
"pay_time": 1788000000000,
|
||
"goods_detail": {"spu_id": "hy-7000000", "sku_id": 1},
|
||
}
|
||
# 真机 queryAllocation 响应里的赛季节点（精英宝典福利季丨Q4：2026-10-09 ~ 2026-12-31）
|
||
ALLOCATION_SEASON_PAYLOAD = {
|
||
"tag0": {"tag0": 200, "tag1": "成功", "tag2": ""},
|
||
"tag1": {
|
||
"tag0": "和平精英宝典积分中心",
|
||
"tag1": "heping_baodian_credit_center_page",
|
||
},
|
||
"tag2": [
|
||
{"tag0": 343, "tag4": 6},
|
||
{
|
||
"tag0": 1,
|
||
"tag2_2": {
|
||
"tag0": {
|
||
"tag0": 15,
|
||
"tag1": "精英宝典福利季",
|
||
"tag2": "精英宝典福利季丨Q4",
|
||
"tag3": 3203,
|
||
"tag5": 17096,
|
||
"tag6": 9815,
|
||
"tag10": 1791475200000,
|
||
"tag11": 1798732799000,
|
||
},
|
||
"tag4": {"tag0": 300},
|
||
},
|
||
},
|
||
],
|
||
}
|
||
|
||
|
||
def _redeem(hex_text: str) -> ShopRawRsp:
|
||
return ShopRawRsp.from_payload(decode_struct_bytes(bytes.fromhex(hex_text)))
|
||
|
||
|
||
class _FakeOrder:
|
||
def __init__(self, payload: dict):
|
||
self._payload = payload
|
||
|
||
def to_dict(self) -> dict:
|
||
return dict(self._payload)
|
||
|
||
|
||
class _FakeOrderList:
|
||
def __init__(self, orders: list[dict], total_count: int | None = None):
|
||
self.orders = [_FakeOrder(item) for item in orders]
|
||
self.totalCount = len(orders) if total_count is None else total_count
|
||
|
||
|
||
class _FakeBindStatus:
|
||
"""只喂给 EliteMixin 用到的那几个字段（角色解析由 BindMixin 自己的用例覆盖）。"""
|
||
|
||
def __init__(self, status: int = 200, msg: str = ""):
|
||
self.status = status
|
||
self.msg = msg
|
||
|
||
def to_dict(self) -> dict:
|
||
return {"status": self.status, "msg": self.msg}
|
||
|
||
|
||
class _StubClient:
|
||
"""积分中心通道的假客户端：可返回宝典状态/订单列表/赛季配置，并记录调用。"""
|
||
|
||
def __init__(self, redeem=None, order_pages=None, season: bool = False):
|
||
self.redeem = redeem
|
||
self.order_pages = order_pages or []
|
||
self.season_payload = ALLOCATION_SEASON_PAYLOAD if season else None
|
||
self.calls: list[tuple[str, dict]] = []
|
||
self.redeem_calls = 0
|
||
|
||
def get_redeem_need_goods(self, **kwargs):
|
||
self.calls.append(("get_redeem_need_goods", kwargs))
|
||
self.redeem_calls += 1
|
||
return self.redeem
|
||
|
||
def play_mall_query_allocation(self, **kwargs):
|
||
self.calls.append(("play_mall_query_allocation", kwargs))
|
||
if self.season_payload is None:
|
||
return None
|
||
return ShopRawRsp.from_payload(self.season_payload)
|
||
|
||
def query_user_order_list(self, **kwargs):
|
||
self.calls.append(("query_user_order_list", kwargs))
|
||
offset = int(kwargs.get("offset") or 0)
|
||
index = offset // 20
|
||
if index >= len(self.order_pages):
|
||
return None
|
||
page = self.order_pages[index]
|
||
return _FakeOrderList(page, total_count=sum(len(p) for p in self.order_pages))
|
||
|
||
|
||
def _runner(monkeypatch, shop_client, activity_client=None):
|
||
runner = HuyaBatchRunner(db=None, batch_id="test-batch", task_type="query_all")
|
||
marks: list[tuple[str, str, dict | None]] = []
|
||
logs: list[tuple[str, str]] = []
|
||
monkeypatch.setattr(
|
||
runner, "_shop_client", lambda uid, cookie: shop_client, raising=False
|
||
)
|
||
if activity_client is not None:
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_activity_client",
|
||
lambda uid, cookie: activity_client,
|
||
raising=False,
|
||
)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_mark_task",
|
||
lambda _db, _task, status, message, result=None: marks.append(
|
||
(status, message, result)
|
||
),
|
||
raising=False,
|
||
)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_push_log",
|
||
lambda level, message: logs.append((level, message)),
|
||
raising=False,
|
||
)
|
||
return runner, marks, logs
|
||
|
||
|
||
def test_real_redeem_bytes_report_elite_opened():
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
|
||
|
||
assert runner._elite_opened_from_redeem(_redeem(REDEEM_OPENED_HEX)) is True
|
||
assert runner._elite_opened_from_redeem(_redeem(REDEEM_NOT_OPENED_HEX)) is False
|
||
# 不是 200 或没有 tag5 时不下结论（界面显示「未查询」而不是猜一个值）
|
||
assert runner._elite_opened_from_redeem(None) is None
|
||
assert (
|
||
runner._elite_opened_from_redeem(ShopRawRsp.from_payload({"tag0": 500})) is None
|
||
)
|
||
|
||
|
||
def test_elite_open_time_uses_latest_paid_handbook_order():
|
||
client = _StubClient(
|
||
order_pages=[[OTHER_ORDER, UNPAID_ELITE_ORDER, PAID_ELITE_ORDER]]
|
||
)
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
|
||
|
||
assert runner._resolve_elite_open_time(client, UID, COOKIE) == 1791895396000
|
||
|
||
|
||
def test_elite_open_time_ignores_unpaid_handbook_order():
|
||
"""未支付的宝典订单不算开通时间（真机：order_status=10、pay_time=0）。"""
|
||
client = _StubClient(order_pages=[[UNPAID_ELITE_ORDER]])
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
|
||
|
||
assert runner._resolve_elite_open_time(client, UID, COOKIE) is None
|
||
|
||
|
||
def test_elite_open_time_scans_following_pages():
|
||
later_order = {**PAID_ELITE_ORDER, "pay_time": 1792000000000}
|
||
client = _StubClient(order_pages=[[OTHER_ORDER] * 20, [later_order]])
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
|
||
|
||
assert runner._resolve_elite_open_time(client, UID, COOKIE) == 1792000000000
|
||
offsets = [
|
||
kwargs.get("offset")
|
||
for name, kwargs in client.calls
|
||
if name == "query_user_order_list"
|
||
]
|
||
assert offsets == [0, 20]
|
||
|
||
|
||
def test_extract_elite_season_from_allocation_payload():
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
|
||
|
||
season = runner._extract_elite_season(ALLOCATION_SEASON_PAYLOAD)
|
||
|
||
assert season == {
|
||
"name": "精英宝典福利季丨Q4",
|
||
"start_ms": 1791475200000,
|
||
"end_ms": 1798732799000,
|
||
"config_id": 15,
|
||
}
|
||
# 认不出赛季结构时不下结论（界面就不显示有效期）
|
||
assert runner._extract_elite_season({"tag0": {"tag1": "x"}}) is None
|
||
assert runner._extract_elite_season(None) is None
|
||
|
||
|
||
def test_query_all_fills_expire_time_from_season(monkeypatch):
|
||
shop_client = _StubClient(
|
||
redeem=_redeem(REDEEM_OPENED_HEX),
|
||
order_pages=[[PAID_ELITE_ORDER]],
|
||
season=True,
|
||
)
|
||
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
|
||
monkeypatch.setattr(
|
||
runner, "_resolve_bind_status", lambda **kwargs: (None, {}), raising=False
|
||
)
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
|
||
runner._execute_query_all(
|
||
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
|
||
)
|
||
|
||
status, message, result = marks[-1]
|
||
assert status == "success"
|
||
assert result["elite_expire_time"] == 1798732799
|
||
assert result["elite_season"]["name"] == "精英宝典福利季丨Q4"
|
||
assert account.elite_expire_time == 1798732799
|
||
assert "有效期至 2026-12-31" in message
|
||
|
||
|
||
def test_refresh_elite_open_after_payment_retries_until_opened():
|
||
"""刚支付完 tag5 可能还是 0（交付延迟）→ 重试到已开通，并写回账号。"""
|
||
|
||
class _FlakyClient(_StubClient):
|
||
def get_redeem_need_goods(self, **kwargs):
|
||
result = super().get_redeem_need_goods(**kwargs)
|
||
if self.redeem_calls > 1:
|
||
return _redeem(REDEEM_OPENED_HEX)
|
||
return result
|
||
|
||
client = _FlakyClient(season=True)
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="create_recharge_order")
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
|
||
state = runner._refresh_elite_open_after_payment(
|
||
client,
|
||
account,
|
||
UID,
|
||
COOKIE,
|
||
goods_num=8315982,
|
||
pay_time_ms=1792053457000,
|
||
attempts=3,
|
||
interval=0,
|
||
)
|
||
|
||
assert client.redeem_calls == 2 # 第一次未开通 → 重试一次即成功
|
||
assert state["elite_opened"] is True
|
||
assert state["elite_open_time"] == 1792053457
|
||
assert state["elite_expire_time"] == 1798732799
|
||
assert state["redeem_points"] == 280
|
||
assert account.elite_opened is True
|
||
assert account.elite_open_time == 1792053457
|
||
assert account.elite_expire_time == 1798732799
|
||
|
||
|
||
def test_refresh_elite_open_after_payment_gives_up_after_attempts():
|
||
client = _StubClient(redeem=_redeem(REDEEM_NOT_OPENED_HEX))
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="create_recharge_order")
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
|
||
state = runner._refresh_elite_open_after_payment(
|
||
client,
|
||
account,
|
||
UID,
|
||
COOKIE,
|
||
goods_num=8315982,
|
||
pay_time_ms=1792053457000,
|
||
attempts=2,
|
||
interval=0,
|
||
)
|
||
|
||
assert client.redeem_calls == 2 # 重试到上限
|
||
assert state["elite_opened"] is False
|
||
assert account.elite_opened is False
|
||
assert account.elite_open_time is None
|
||
|
||
|
||
def test_create_recharge_order_confirms_elite_open_after_payment(db, monkeypatch):
|
||
"""支付成功后自动确认开通状态：任务结果与账号字段都带上宝典状态。"""
|
||
|
||
outcome = SimpleNamespace(
|
||
ok=True,
|
||
message="",
|
||
detail={"spu_id": "hy-8328015", "sku_id": 7756811, "name": "精英宝典"},
|
||
sku_id=7756811,
|
||
product_name="精英宝典",
|
||
unit_price=3000,
|
||
order_raw={"order_id": 11399999},
|
||
pay_raw={"pay_order_id": "pay-1"},
|
||
amount=3000,
|
||
order_id=11399999,
|
||
app_order_id="app-1",
|
||
pay_order_id="pay-1",
|
||
pay_url="https://example.com/pay",
|
||
supplier_name="和平精英官方小店",
|
||
promotion={},
|
||
promotion_info="本单无返利",
|
||
reused_order={},
|
||
)
|
||
|
||
shop_client = _StubClient(redeem=_redeem(REDEEM_OPENED_HEX), season=True)
|
||
runner = HuyaBatchRunner(
|
||
db=None,
|
||
batch_id="t",
|
||
task_type="create_recharge_order",
|
||
payload={"spu_id": "hy-8328015", "sku_id": 7756811, "count": 1},
|
||
)
|
||
marks: list[tuple[str, str, dict | None]] = []
|
||
monkeypatch.setattr(
|
||
runner, "_shop_client", lambda uid, cookie: shop_client, raising=False
|
||
)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_mark_task",
|
||
lambda _db, _task, status, message, result=None: marks.append(
|
||
(status, message, result)
|
||
),
|
||
raising=False,
|
||
)
|
||
monkeypatch.setattr(
|
||
runner, "_update_task_progress", lambda *args, **kwargs: None, raising=False
|
||
)
|
||
monkeypatch.setattr(
|
||
runner, "_push_log", lambda *args, **kwargs: None, raising=False
|
||
)
|
||
|
||
def fake_wait_payment(**kwargs):
|
||
# 真实现会边轮询边写 result，这里照抄「已支付」那几个字段
|
||
target = kwargs.get("result")
|
||
if isinstance(target, dict):
|
||
target.update(
|
||
{
|
||
"payment_status": "paid",
|
||
"payment_status_label": "已支付",
|
||
"payment_paid": True,
|
||
}
|
||
)
|
||
return "paid", {"pay_time": 1792053457000}
|
||
|
||
monkeypatch.setattr(
|
||
runner, "_wait_recharge_payment", fake_wait_payment, raising=False
|
||
)
|
||
monkeypatch.setattr(
|
||
"web.backend.services.huya_runner_recharge.run_open_flow",
|
||
lambda *args, **kwargs: outcome,
|
||
)
|
||
# 二维码内容优先用收银台页面里的扫码内容（真机：支付宝扫码支付码）
|
||
monkeypatch.setattr(
|
||
"web.backend.services.huya_runner_recharge.resolve_pay_targets",
|
||
lambda _pay_url, **_kwargs: PayTargets(
|
||
cashier_url="https://excashier.alipay.com/standard/auth.htm?payOrderId=x.25",
|
||
qr_code="https://qr.alipay.com/upx03110ghym9vm37sx7255c",
|
||
),
|
||
)
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
db.add(account)
|
||
db.commit()
|
||
|
||
runner._execute_create_recharge_order(
|
||
db,
|
||
SimpleNamespace(id=587),
|
||
account,
|
||
{"uid": UID, "cookie": COOKIE},
|
||
{"room_pid": "1199650619883", "pay_channel": "Zfb", "sid": "2203"},
|
||
)
|
||
|
||
status, _message, result = marks[-1]
|
||
assert status == "success"
|
||
assert result["payment_paid"] is True
|
||
assert result["elite_opened"] is True
|
||
assert result["elite_open_time"] == 1792053457
|
||
assert result["elite_expire_time"] == 1798732799
|
||
assert result["post_payment_points"] == 280
|
||
# 二维码内容给前端两条短地址（原始支付地址 1626 字符，扫码扫不动）
|
||
assert result["pay_qr_path"] == pay_qr_path(587)
|
||
# 优先用收银台页面里的扫码内容（支付宝 App 扫码即付款）
|
||
assert result["pay_qr_code"] == "https://qr.alipay.com/upx03110ghym9vm37sx7255c"
|
||
# 收银台直达地址作为兜底保留（PC 打开就是这个页面）
|
||
assert result["pay_cashier_url"].startswith("https://excashier.alipay.com/")
|
||
assert account.elite_opened is True
|
||
assert account.elite_expire_time == 1798732799
|
||
assert account.points == 280
|
||
assert account.status == "recharge_paid"
|
||
|
||
|
||
def test_query_all_combines_points_elite_and_role(monkeypatch):
|
||
shop_client = _StubClient(
|
||
redeem=_redeem(REDEEM_OPENED_HEX), order_pages=[[PAID_ELITE_ORDER]]
|
||
)
|
||
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_resolve_bind_status",
|
||
lambda **kwargs: (_FakeBindStatus(), {"bind_status_source": "outer"}),
|
||
raising=False,
|
||
)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_bind_role_result",
|
||
staticmethod(
|
||
lambda _status: {"role_name": "软踏日潺湲", "game_title": "和平精英"}
|
||
),
|
||
raising=False,
|
||
)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_bind_change_state",
|
||
staticmethod(lambda _status: {"is_bound": True}),
|
||
raising=False,
|
||
)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_role_channel",
|
||
staticmethod(lambda _status: "苹果(iOS) / 手Q"),
|
||
raising=False,
|
||
)
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
|
||
runner._execute_query_all(
|
||
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
|
||
)
|
||
|
||
status, message, result = marks[-1]
|
||
assert status == "success"
|
||
assert result["points"] == 280
|
||
assert result["elite_opened"] is True
|
||
assert result["elite_open_time"] == 1791895396
|
||
assert result["role_name"] == "软踏日潺湲"
|
||
assert result["failures"] == []
|
||
assert account.points == 280
|
||
assert account.elite_opened is True
|
||
assert account.elite_open_time == 1791895396
|
||
assert account.game_name == "软踏日潺湲"
|
||
assert account.game_channel == "苹果(iOS) / 手Q"
|
||
assert account.status == "all_queried"
|
||
assert (
|
||
"积分 280" in message
|
||
and "已开通宝典" in message
|
||
and "角色 软踏日潺湲" in message
|
||
)
|
||
|
||
|
||
def test_query_all_reports_not_opened_without_order_lookup(monkeypatch):
|
||
shop_client = _StubClient(redeem=_redeem(REDEEM_NOT_OPENED_HEX))
|
||
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
|
||
monkeypatch.setattr(
|
||
runner, "_resolve_bind_status", lambda **kwargs: (None, {}), raising=False
|
||
)
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
|
||
runner._execute_query_all(
|
||
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
|
||
)
|
||
|
||
status, message, result = marks[-1]
|
||
# 角色接口挂了不影响宝典/积分结论
|
||
assert status == "success"
|
||
assert result["elite_opened"] is False
|
||
assert account.elite_opened is False
|
||
assert account.elite_open_time is None
|
||
assert "未开通宝典" in message
|
||
assert "角色信息接口无响应" in message
|
||
# 未开通不查订单（省一次请求）
|
||
assert [name for name, _kwargs in shop_client.calls] == ["get_redeem_need_goods"]
|
||
|
||
|
||
def test_query_all_fails_when_points_center_unreachable(monkeypatch):
|
||
shop_client = _StubClient(redeem=None)
|
||
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
|
||
monkeypatch.setattr(
|
||
runner, "_resolve_bind_status", lambda **kwargs: (None, {}), raising=False
|
||
)
|
||
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
|
||
|
||
runner._execute_query_all(
|
||
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
|
||
)
|
||
|
||
status, message, _result = marks[-1]
|
||
assert status == "failed"
|
||
assert "积分中心接口无响应" in message
|
||
|
||
|
||
def test_shop_client_falls_back_to_http_when_wss_init_fails(monkeypatch):
|
||
"""设备态齐全但积分中心端点不回包时（wsLaunch 超时）回退 HTTP，而不是判整批失败。"""
|
||
cookie = (
|
||
"udb_cred=cred; udb_biztoken=biz; udb_uid=1199659425268; yyuid=1199659425268; "
|
||
"sdid=sdid; udb_deviceid=w_1234567890123456789; udb_guiddata=guiddata; "
|
||
"udb_anobiztoken=anobiz"
|
||
)
|
||
logs: list[tuple[str, str]] = []
|
||
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
|
||
|
||
def boom(uid: int, cookie_text: str, kind: str):
|
||
raise RuntimeError(f"虎牙 {kind} WSS 初始化失败")
|
||
|
||
monkeypatch.setattr(runner, "_open_wss_session", boom, raising=False)
|
||
monkeypatch.setattr(
|
||
runner,
|
||
"_push_log",
|
||
lambda level, message: logs.append((level, message)),
|
||
raising=False,
|
||
)
|
||
|
||
client = runner._shop_client(1199659425268, cookie)
|
||
|
||
assert isinstance(client, HuyaHttpClient)
|
||
assert logs and logs[-1][0] == "warning"
|
||
assert "WSS 会话不可用" in logs[-1][1]
|
||
assert "sdid" not in json.dumps(logs[-1][1]) # 日志不带 Cookie 值
|