import json import re from types import SimpleNamespace from unittest.mock import Mock import pytest from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError class TestXpdExchange: def setup_method(self): self.client = DouyuActivityClient( "acf_uid=100; acf_stk=stk; acf_ltkid=ltkid", ) def test_exchange_uses_fresh_embed_authorization_and_point_price(self): self.client._request = Mock( return_value=SimpleNamespace( text='var buyInfo={"ret":"0","msg":"ok","serial":"order-1","newBalance":88};', ) ) result = self.client.xpd_exchange_goods( embed_query={ "gameId": "cjm", "actId": "46195", "appId": "bp_cf", "livePlatId": "douyu", "code": "fresh-code", "timestamp": "123", "v": "2.0", "sig": "fresh-signature", }, act_id="46195", rid="9263298", commodity_id="goods-1", price=1200, picture="https://cdn.example/item.jpg", pay_type=1, action_id="46195", ) assert result["serial"] == "order-1" assert result["new_balance"] == 88 call = self.client._request.call_args assert call.args[:2] == ("get", self.client.XPD_LIVE_BUY_API) params = call.kwargs["params"] assert params["code"] == "fresh-code" assert params["sig"] == "fresh-signature" assert params["propid"] == "goods-1" assert params["paytype"] == "1" assert json.loads(params["appext"]) == { "user_price": 1200, "sGoodsPic": "https://cdn.example/item.jpg", "paytype": 1, } def test_role_and_balance_preserve_ios_platform_zero(self): self.client._request = Mock( return_value=SimpleNamespace( text=( 'var info={"gameOpenId":"openid-ios","roleId":"role-ios",' '"roleName":"iOS角色","type":"wx","platId":0,"area":1};' ), ) ) role = self.client.xpd_get_role(embed_query={}, act_id="46195", rid="9263298") assert role["plat_id"] == "0" assert role["area"] == "1" self.client._request = Mock( return_value=SimpleNamespace( text='var banlanceInfo={"ret":0,"balance":2199};', ) ) result = self.client.xpd_balance( embed_query={}, act_id="46195", openid="openid-ios", roleid="role-ios", plat="0", areaid="1", ) params = self.client._request.call_args.kwargs["params"] assert params["plat"] == "0" assert re.fullmatch(r"\d{13}", params["_time"]) assert result["balance"] == 2199 def test_exchange_surfaces_daoju_failure_message(self): self.client._request = Mock( return_value=SimpleNamespace( text='var buyInfo={"ret":"-1","msg":"点券不足"};', ) ) with pytest.raises(DouyuActivityError, match="点券不足"): self.client.xpd_exchange_goods( embed_query={"code": "fresh-code"}, act_id="46195", rid="9263298", commodity_id="goods-1", price=1200, ) def test_list_goods_requests_and_preserves_inventory(self): self.client._request = Mock( return_value=SimpleNamespace( text=( 'var recommend={"data":{"client_data":{"itemsdetail":[{' '"iGoodsId":"goods-1","sGoodsName":"测试商品","iPrice":"6800",' '"iJb2Price":"0","iGoodsLeft":"0"}]}}};' ), ) ) result = self.client.xpd_list_goods( embed_query={}, act_id="46195", openid="openid", roleid="roleid", ) assert result["goods"][0]["price"] == 6800 assert result["goods"][0]["goods_left"] == 0 params = self.client._request.call_args.kwargs["params"] assert params["excludeFields"] == "iCategoryId_76" assert "category" not in params def test_list_goods_fetches_every_page_and_deduplicates_goods(self): first_page = [ {"iGoodsId": f"goods-{index}", "sGoodsName": f"商品 {index}"} for index in range(10) ] second_page = [ {"iGoodsId": "goods-9", "sGoodsName": "重复商品"}, {"iGoodsId": "goods-10", "sGoodsName": "商品 10"}, ] self.client._request = Mock( side_effect=[ SimpleNamespace( text=f'var recommend={{"errcode":0,"data":{{"client_data":{{"itemsdetail":{json.dumps(first_page)}}}}}}};' ), SimpleNamespace( text=f'var recommend={{"errcode":0,"data":{{"client_data":{{"itemsdetail":{json.dumps(second_page)}}}}}}};' ), ] ) result = self.client.xpd_list_goods( embed_query={}, act_id="46195", openid="openid", roleid="roleid", ) assert len(result["goods"]) == 11 assert [item["commodity_id"] for item in result["goods"]][-1] == "goods-10" assert self.client._request.call_count == 2 first_params = self.client._request.call_args_list[0].kwargs["params"] second_params = self.client._request.call_args_list[1].kwargs["params"] assert (first_params["page_begin"], first_params["page_num"]) == ("0", "1") assert (second_params["page_begin"], second_params["page_num"]) == ("10", "2") def test_purchase_records_parses_order_list_and_paginates(self): first_record = { "sSerialNum": "order-1", "dtBuyTime": "2026-08-08 01:00:42", "dtPayTime": "2026-08-08 01:00:55", "iStatus": "3", "iPrice": "800", "iPayAmount": "800", "sRoleName": "测试角色", "sGoodsInfo": json.dumps( { "list": [ { "sGoodsName": "改名卡", "sGoodsPic": "https://cdn.example/item.jpg", } ], } ), } self.client._request = Mock( side_effect=[ SimpleNamespace( text=f'var info={{"ret":"0","msg":"ok","data":{{"cnt":11,"list":{json.dumps([first_record])}}}}};' ), SimpleNamespace( text='var info={"ret":"0","msg":"ok","data":{"cnt":11,"list":[]}};' ), ] ) result = self.client.xpd_purchase_records( embed_query={ "gameId": "cjm", "actId": "18882", "appId": "bp_cf", "livePlatId": "douyu", "code": "fresh-code", "timestamp": "123", "v": "2.0", "sig": "fresh-signature", }, act_id="46195", page_size=1, ) assert result["total"] == 11 assert result["records"][0]["goods_name"] == "改名卡" assert result["records"][0]["serial"] == "order-1" assert self.client._request.call_count == 2 params = self.client._request.call_args_list[0].kwargs["params"] assert params["_service"] == "order.list" assert params["pn"] == "1" assert params["code"] == "fresh-code" def test_exchange_keeps_raw_price_from_goods_list(self): self.client._request = Mock( side_effect=[ SimpleNamespace( text='var recommend={"errcode":0,"data":{"client_data":{"itemsdetail":[{' '"iGoodsId":"goods-1","sGoodsName":"测试商品","iPrice":"6800",' '"iJb2Price":"0","iGoodsLeft":"-1"}]}}};', ), SimpleNamespace( text='var buyInfo={"ret":"0","msg":"ok","serial":"order-1","newBalance":88};' ), ] ) goods = self.client.xpd_list_goods( embed_query={}, act_id="46195", openid="openid", roleid="roleid", )["goods"] result = self.client.xpd_exchange_goods( embed_query={"code": "fresh-code"}, act_id="46195", rid="9263298", commodity_id=goods[0]["commodity_id"], price=goods[0]["price"], picture="", pay_type=1, action_id="46195", ) assert goods[0]["price"] == 6800 assert result["serial"] == "order-1" params = self.client._request.call_args.kwargs["params"] assert json.loads(params["appext"])["user_price"] == 6800