149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
import asyncio
|
||
from unittest.mock import AsyncMock
|
||
|
||
from core.huya.activity_structs import ActPrizeDetailItem, GetActPrizeDetailResp
|
||
from core.huya.http_client import SHOP_BIZ_UA, HuyaHttpClient
|
||
from core.huya.shop_structs import (
|
||
ListPayChannelRsp,
|
||
OrderDetailData,
|
||
OrderDetailOrder,
|
||
OrderDetailRsp,
|
||
ShopCodeRsp,
|
||
)
|
||
from core.huya.taf_protocol import TafInputStream, TafOutputStream, TafType
|
||
from core.huya.wss_client import TAIL_BYTES, HuyaWssClient, WssCommand, WssMessage
|
||
from web.backend.services.huya_runner_core import HuyaBatchRunnerCore
|
||
|
||
|
||
def _round_trip(value):
|
||
output = TafOutputStream()
|
||
output.write_struct_begin(0)
|
||
value.write_to(output)
|
||
output.write_struct_end()
|
||
stream = TafInputStream(output.get_bytes())
|
||
_tag, _dtype = stream.read_head()
|
||
value_type = type(value)
|
||
decoded = value_type()
|
||
decoded.read_from(stream)
|
||
_end_tag, end_type = stream.read_head()
|
||
assert end_type == TafType.STRUCT_END
|
||
return decoded
|
||
|
||
|
||
def test_shop_user_matches_latest_wss_identity():
|
||
user = HuyaHttpClient._build_shop_user(123, "yyuid=123; udb_passport=test")
|
||
assert user.sHuYaUA == SHOP_BIZ_UA
|
||
assert user.sGuid == ""
|
||
assert user.lUid == 123
|
||
|
||
|
||
def test_act_prize_detail_round_trip_preserves_exchange_guards():
|
||
prize = ActPrizeDetailItem()
|
||
prize.prizeId = 12861
|
||
prize.name = "幸运币礼包（小）"
|
||
prize.score = 20
|
||
prize.newScore = 20
|
||
prize.leftNum = 15000
|
||
prize.isShowNum = 1
|
||
prize.isCanExchange = 1
|
||
prize.exchangeStartTime = 1782874800
|
||
prize.exchangeEndTime = 1790783400
|
||
response = GetActPrizeDetailResp()
|
||
response.status = 200
|
||
response.msg = "请求成功"
|
||
response.prize = prize
|
||
decoded = _round_trip(response)
|
||
assert decoded.status == 200
|
||
assert decoded.prize is not None
|
||
assert decoded.prize.prizeId == 12861
|
||
assert decoded.prize.newScore == 20
|
||
assert decoded.prize.isCanExchange == 1
|
||
|
||
|
||
def test_order_detail_round_trip_exposes_payment_status():
|
||
order = OrderDetailOrder()
|
||
order.orderId = 10938643
|
||
order.orderStatus = 50
|
||
order.createTime = 1788226648183
|
||
order.payTime = 1788226656000
|
||
order.totalPrice = 3000
|
||
response = OrderDetailRsp()
|
||
response.code = 200
|
||
response.data = OrderDetailData()
|
||
response.data.order = order
|
||
decoded = _round_trip(response)
|
||
assert decoded.code == 200
|
||
assert decoded.order is not None
|
||
assert decoded.order.orderId == 10938643
|
||
assert decoded.order.orderStatus == 50
|
||
assert decoded.order.payTime > 0
|
||
|
||
|
||
def test_shop_preflight_response_contracts():
|
||
channels = ListPayChannelRsp()
|
||
channels.channels = ["Zfb", "Weixin"]
|
||
channels.code = 200
|
||
decoded_channels = _round_trip(channels)
|
||
assert decoded_channels.channels == ["Zfb", "Weixin"]
|
||
assert decoded_channels.code == 200
|
||
|
||
status = ShopCodeRsp()
|
||
status.code = 200
|
||
status.message = ""
|
||
decoded_status = _round_trip(status)
|
||
assert decoded_status.code == 200
|
||
|
||
|
||
def test_wss_requires_web_device_cookie_fields():
|
||
assert not HuyaBatchRunnerCore._has_web_wss_cookie(
|
||
"udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1"
|
||
)
|
||
assert HuyaBatchRunnerCore._has_web_wss_cookie(
|
||
"udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1; sdid=s; guid=g; "
|
||
"udb_guiddata=gd; udb_deviceid=w_1234567890123456789; game_did=did; "
|
||
"_qimei_uuid42=q; udb_anobiztoken=ab; __yamid_new=y"
|
||
)
|
||
|
||
|
||
def test_activity_auth_frame_matches_latest_capture_layout():
|
||
"""对齐 9.1 活动 AUTH（828 连接）逐字段结构。
|
||
|
||
9.1: tag3 ZERO 首字段 -> tag0 uid -> tag1 UA -> tag2 cookie(STRING4)
|
||
-> tag3 STRING1 guid(32hex) -> tag4 INT8=1 -> tag5 "HUYA&ZH&2052"
|
||
-> tag6 "" -> TAIL_BYTES
|
||
"""
|
||
async def build_frame():
|
||
client = HuyaWssClient()
|
||
client._activity_mode = True
|
||
client.ws = type("FakeWs", (), {"send": AsyncMock()})()
|
||
await client.send_auth(
|
||
1199664135026,
|
||
"guid=" + "0a" * 16 + "; udb_cred=" + "x" * 300,
|
||
sequence=0x1D000109,
|
||
)
|
||
return client.ws.send.call_args.args[0]
|
||
|
||
message = WssMessage.decode(asyncio.run(build_frame()))
|
||
assert message.command == WssCommand.AUTH
|
||
assert message.sequence == 0x1D000109
|
||
assert message.body.endswith(TAIL_BYTES)
|
||
|
||
stream = TafInputStream(message.body[: -len(TAIL_BYTES)])
|
||
# 首字段：tag3 ZERO（9.1 活动帧的占位字段）
|
||
assert stream.read_head() == (3, TafType.ZERO)
|
||
assert stream.read_int64(0) == 1199664135026
|
||
assert stream.read_string(1) == "webh5&0.0.1&websocket&&diypc_52775"
|
||
assert stream.read_head() == (2, TafType.STRING4)
|
||
cookie_len = int.from_bytes(stream.buf.read(4), "big")
|
||
assert cookie_len > 255
|
||
stream.buf.read(cookie_len)
|
||
# tag3 = STRING1 guid（9.1 中为 Cookie 里的 32hex guid）
|
||
guid = stream.read_string(3)
|
||
assert guid == "0a" * 16
|
||
assert stream.read_int8(4) == 1
|
||
assert stream.read_string(5) == "HUYA&ZH&2052"
|
||
assert stream.read_string(6) == ""
|
||
|
||
# body 必须与 9.1 828 AUTH 同样以 TAIL_BYTES 收尾
|
||
assert message.body[-len(TAIL_BYTES):] == TAIL_BYTES
|