90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
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
|
|
|
|
|
|
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
|