"""查询积分改走积分中心新接口（2026-10-15 真机实测）。 背景：旧活动接口 ``webActUI.getUserScore`` 对任何 sid 都返回 603「活动已结束」（该接口已下线，与工作台里的 SID/活动配置无关）； 积分中心用 ``shopMiddleUI.getRedeemNeedGoods`` 查积分： ``tag2``=当前积分、``tag3``=兑换门槛。下面固化的响应字节来自真机 HTTP 通道。 抓包/真机字面值直接写在测试里（不引用实现常量），避免断言跟着实现漂移。 """ import asyncio from types import SimpleNamespace from core.huya.frame_decoder import _decode_wup_body, decode_struct_bytes from core.huya.http_client import HuyaHttpClient from core.huya.shop_structs import ShopRawRsp from core.huya.wss_client import HuyaWssClient, WssMessage from web.backend.models import HuyaAccount from web.backend.services.huya_runner import HuyaBatchRunner UID = 1199665170519 GUID = "0a8995af9218ce6a340261946294a9e7" COOKIE = f"udb_uid={UID}; yyuid={UID}; guid={GUID}" # 真机（2026-10-15，账号 hy_297832664）getRedeemNeedGoods 响应：200 / 积分 280 / 门槛 20 REDEEM_RSP_HEX = "0a0100c8160ce69fa5e8afa2e68890e58a9f21011830144c50016001790c0b" # 真机 getGrowLevelUserInfo 响应：成长值 980 / 成长等级 3 / 昵称 / 头像 GROW_RSP_HEX = ( "0a0a0c1606e68890e58a9f0b1a0103d41003230000019f124082043c4c50" "0360ff7101f48103e80b260fe8bdafe8b88fe697a5e6bdbae6b9b236386874" "7470733a2f2f6469792d6173736574732e6d737374617469632e636f6d2f6371" "68666f6c6465722f4e657744656661756c742e706e670b" ) class _FakeWs: """记录发出的帧，并立刻兑现 RPC future（离线跑完整链路）。""" def __init__(self, client: HuyaWssClient): self.client = client self.sent: list[bytes] = [] async def send(self, data: bytes): self.sent.append(data) while self.client.rpc_queue: future = self.client.rpc_queue.popleft() if not future.done(): future.set_result(b"") def last_request(self) -> dict: return _decode_wup_body(WssMessage.decode(self.sent[-1]).body) def _points_client() -> HuyaWssClient: client = HuyaWssClient(logger=lambda _message: None) client._points_mode = True client.ws = _FakeWs(client) return client def _raw_rsp(payload_hex: str) -> ShopRawRsp: return ShopRawRsp.from_payload( decode_struct_bytes(bytes.fromhex("".join(payload_hex.split()))) ) class _StubClient: """按需返回真机响应字节的积分中心客户端。""" def __init__(self, redeem=True, grow=True): self.redeem = _raw_rsp(REDEEM_RSP_HEX) if redeem else None self.grow = _raw_rsp(GROW_RSP_HEX) if grow else None self.calls: list[tuple[str, dict]] = [] def get_redeem_need_goods(self, **kwargs): self.calls.append(("get_redeem_need_goods", kwargs)) return self.redeem def get_grow_level_user_info(self, **kwargs): self.calls.append(("get_grow_level_user_info", kwargs)) return self.grow def _runner( monkeypatch, client, *, ck_valid: bool = True ) -> tuple[HuyaBatchRunner, list]: runner = HuyaBatchRunner(db=None, batch_id="test-batch", task_type="query_points") marks: list[tuple[str, str, dict | None]] = [] monkeypatch.setattr(runner, "_http_client", lambda: client, raising=False) probe_status = SimpleNamespace(status=200, msg="") if ck_valid else None monkeypatch.setattr( runner, "_probe_bind_status", lambda **_kwargs: (probe_status, {}), raising=False, ) monkeypatch.setattr( runner, "_mark_task", lambda _db, _task, status, message, result=None: marks.append( (status, message, result) ), raising=False, ) return runner, marks def test_wss_redeem_need_goods_request_matches_real_call(): async def run(): client = _points_client() await client.get_redeem_need_goods(uid=UID, guid=GUID, cookie=COOKIE) message = WssMessage.decode(client.ws.sent[-1]) # 积分中心业务帧统一 0x1d000107（只有下单/支付用 0x1d000108） assert message.sequence == 0x1D000107 decoded = client.ws.last_request() assert (decoded["servant"], decoded["func"]) == ( "shopMiddleUI", "getRedeemNeedGoods", ) t_req = decoded["tReq"] assert t_req["tag0"]["tag0"] == UID assert t_req["tag1"]["tag0"] == "huya" assert t_req["tag1"]["tag4"] == 7 assert t_req["tag1"]["tag5"] == "yellowcarlist" # 默认兑换商品号必须是真机可用的 hy-8315982 的数字形式 assert t_req["tag2"] == 8315982 asyncio.run(run()) def test_wss_grow_level_request_carries_only_user_id(): async def run(): client = _points_client() await client.get_grow_level_user_info(uid=UID, guid=GUID, cookie=COOKIE) decoded = client.ws.last_request() assert (decoded["servant"], decoded["func"]) == ( "GameBizUI", "getGrowLevelUserInfo", ) t_req = decoded["tReq"] assert t_req["tag0"]["tag0"] == UID # 抓包 tReq 只有 tag0，没有 shopAppInfo assert set(t_req) == {"tag0"} asyncio.run(run()) def test_http_redeem_need_goods_request_matches_real_call(): client = HuyaHttpClient(logger=lambda _message: None) recorded: list[tuple] = [] def fake_call_rpc( service, method, req_struct, rsp_class=None, uid=0, guid="", cookie="", **kwargs ): recorded.append((service, method, req_struct)) client.call_rpc = fake_call_rpc # type: ignore[method-assign] client.get_redeem_need_goods(uid=UID, guid="", cookie=COOKIE) service, method, req = recorded[-1] assert (service, method) == ("shopMiddleUI", "getRedeemNeedGoods") assert req.userId.lUid == UID assert req.shopAppInfo.scene == 7 assert req.shopAppInfo.sourceId == "yellowcarlist" assert req.goodsNum == 8315982 def test_real_redeem_response_bytes_are_points_and_threshold(): rsp = _raw_rsp(REDEEM_RSP_HEX) assert rsp.code == 200 assert rsp.message == "查询成功" # tag2 = 当前积分，tag3 = 兑换门槛积分（真机 280 / 20） assert rsp.value("tag2") == 280 assert rsp.value("tag3") == 20 def test_real_grow_level_response_bytes_are_level_and_nickname(): rsp = _raw_rsp(GROW_RSP_HEX) assert rsp.code == 0 assert rsp.message == "成功" assert rsp.value("tag1", "tag0") == 980 # 成长值 assert rsp.value("tag1", "tag1") == 3 # 成长等级 assert rsp.value("tag2") == "软踏日潺湲" assert str(rsp.value("tag3")).startswith("https://diy-assets.msstatic.com/") def test_query_points_reads_points_from_points_center(monkeypatch): client = _StubClient() runner, marks = _runner(monkeypatch, client) account = HuyaAccount(uid=str(UID), cookie=COOKIE) runner._execute_query_points( None, None, account, {"uid": UID, "cookie": COOKIE}, {} ) assert [name for name, _kwargs in client.calls] == [ "get_redeem_need_goods", "get_grow_level_user_info", ] assert client.calls[0][1]["goods_num"] == 8315982 status, message, result = marks[-1] assert status == "success" assert message == "积分: 280" assert account.points == 280 assert result["points"] == 280 assert result["redeem_threshold"] == 20 assert result["grow_level"] == 3 assert result["grow_value"] == 980 assert result["nickname"] == "软踏日潺湲" def test_query_points_uses_http_and_prechecks_ck_with_new_probe(monkeypatch): client = _StubClient() runner, marks = _runner(monkeypatch, client) probe_calls: list[dict] = [] monkeypatch.setattr( runner, "_probe_bind_status", lambda **kwargs: ( probe_calls.append(kwargs) or SimpleNamespace(status=200, msg=""), {}, ), raising=False, ) monkeypatch.setattr( runner, "_shop_client", lambda *_args, **_kwargs: (_ for _ in ()).throw( AssertionError("查询积分不应再建 WSS 商城会话") ), raising=False, ) account = HuyaAccount(uid=str(UID), cookie=COOKIE) runner._execute_query_points( None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271", "outer_act_id": "17096"}, ) assert probe_calls == [ { "uid": UID, "cookie": COOKIE, "b_act_id_int": 9271, "outer_act_id": "17096", } ] assert marks[-1][0] == "success" assert [name for name, _kwargs in client.calls] == [ "get_redeem_need_goods", "get_grow_level_user_info", ] def test_query_points_keeps_working_when_grow_level_fails(monkeypatch): client = _StubClient(grow=False) runner, marks = _runner(monkeypatch, client) account = HuyaAccount(uid=str(UID), cookie=COOKIE) runner._execute_query_points( None, None, account, {"uid": UID, "cookie": COOKIE}, {} ) status, message, result = marks[-1] assert status == "success" assert message == "积分: 280" assert "grow_level" not in result def test_query_points_reports_server_error_of_points_center(monkeypatch): class _FailingClient(_StubClient): def get_redeem_need_goods(self, **kwargs): self.calls.append(("get_redeem_need_goods", kwargs)) return ShopRawRsp.from_payload( decode_struct_bytes( bytes.fromhex( "0a0101f4161ee7b3bbe7bb9fe5bc82e5b8b8efbc8ce8afb7e7a88de5" "908ee9878de8af952c3c4c5c6c0b" ) ) ) client = _FailingClient() runner, marks = _runner(monkeypatch, client) account = HuyaAccount(uid=str(UID), cookie=COOKIE) runner._execute_query_points( None, None, account, {"uid": UID, "cookie": COOKIE}, {} ) status, message, result = marks[-1] assert status == "failed" assert message == "系统异常，请稍后重试" assert result["code"] == 500 assert result["goods_num"] == 8315982 def test_query_points_reports_points_center_offline_without_legacy_fallback( monkeypatch, ): client = _StubClient(redeem=False) runner, marks = _runner(monkeypatch, client) account = HuyaAccount(uid=str(UID), cookie=COOKIE) runner._execute_query_points( None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271", "outer_act_id": "17096"}, ) status, message, result = marks[-1] assert status == "error" assert "积分中心接口无响应" in message assert result["ck_valid"] is True # 旧接口已下线，不能再让它多等一个 15 秒超时。 assert [name for name, _kwargs in client.calls] == ["get_redeem_need_goods"] def test_query_points_marks_ck_expired_before_calling_points_center(monkeypatch): client = _StubClient() runner, marks = _runner(monkeypatch, client, ck_valid=False) account = HuyaAccount(uid=str(UID), cookie=COOKIE, status="imported") runner._execute_query_points( None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271", "outer_act_id": "17096"}, ) status, message, result = marks[-1] assert status == "error" assert "CK 已失效" in message assert result["ck_expired"] is True assert result["ck_expired_reason"] == "角色信息接口无响应" assert account.status == "ck_expired" assert client.calls == []