145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import pytest
|
|
from unittest.mock import Mock
|
|
|
|
import requests
|
|
|
|
from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError
|
|
|
|
|
|
class TestEliteLockExchange:
|
|
def setup_method(self):
|
|
self.client = DouyuActivityClient("acf_uid=100")
|
|
self.client.csrf_token = Mock(side_effect=["csrf-create", "csrf-pay"])
|
|
|
|
@staticmethod
|
|
def multipart_values(files):
|
|
return {key: value[1] for key, value in files.items()}
|
|
|
|
def test_exchange_locks_then_pays_with_fresh_csrf_tokens(self):
|
|
self.client._request_json = Mock(
|
|
side_effect=[
|
|
{
|
|
"error": 0,
|
|
"msg": "请求正常",
|
|
"data": {
|
|
"orderId": "5874",
|
|
"commodityName": "套装-兔兔白日梦",
|
|
"commodityImage": "https://example.test/item.png",
|
|
"score": 1280,
|
|
"num": 1,
|
|
"expireSeconds": "300",
|
|
},
|
|
},
|
|
{
|
|
"error": 0,
|
|
"msg": "请求正常",
|
|
"data": {
|
|
"orderId": "5874",
|
|
"exchangeId": "11098933",
|
|
"commodityType": 3,
|
|
"commodityImage": "https://example.test/item.png",
|
|
"exchangeNum": 1,
|
|
"sType": 1,
|
|
"gType": 2,
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
result = self.client.exchange_goods(
|
|
manual_id="G4KA4Qnz4LDp7",
|
|
rid="9263298",
|
|
commodity_id="20260120QYOOB_bag10",
|
|
)
|
|
|
|
assert result["order_id"] == "5874"
|
|
assert result["exchange_id"] == "11098933"
|
|
assert self.client.csrf_token.call_count == 2
|
|
create_call, pay_call = self.client._request_json.call_args_list
|
|
assert create_call.args[:3] == (
|
|
"post",
|
|
self.client.CREATE_EXCHANGE_ORDER_API,
|
|
"锁定兑换商品",
|
|
)
|
|
assert self.multipart_values(create_call.kwargs["files"]) == {
|
|
"manualID": "G4KA4Qnz4LDp7",
|
|
"rid": "9263298",
|
|
"commodityID": "20260120QYOOB_bag10",
|
|
"num": "1",
|
|
"csrfToken": "csrf-create",
|
|
}
|
|
assert "Content-Type" not in create_call.kwargs["headers"]
|
|
assert pay_call.args[:3] == (
|
|
"post",
|
|
self.client.PAY_EXCHANGE_ORDER_API,
|
|
"支付锁单",
|
|
)
|
|
assert self.multipart_values(pay_call.kwargs["files"]) == {
|
|
"orderID": "5874",
|
|
"manualID": "G4KA4Qnz4LDp7",
|
|
"csrfToken": "csrf-pay",
|
|
}
|
|
|
|
def test_lock_requires_order_id_in_success_response(self):
|
|
self.client._request_json = Mock(return_value={"error": 0, "data": {}})
|
|
|
|
with pytest.raises(DouyuActivityError, match="没有 orderId"):
|
|
self.client.create_exchange_order(
|
|
manual_id="manual",
|
|
rid="9263298",
|
|
commodity_id="goods",
|
|
)
|
|
|
|
def test_locked_order_list_uses_captured_query_shape(self):
|
|
self.client._request_json = Mock(
|
|
return_value={
|
|
"error": 0,
|
|
"data": {"total": 1, "list": [{"orderId": "5874"}]},
|
|
}
|
|
)
|
|
|
|
result = self.client.exchange_orders(manual_id="manual", page=2, page_size=20)
|
|
|
|
assert result["total"] == 1
|
|
assert result["orders"] == [{"orderId": "5874"}]
|
|
call = self.client._request_json.call_args
|
|
assert call.args[:3] == (
|
|
"get",
|
|
self.client.EXCHANGE_ORDER_LIST_API,
|
|
"查询锁单",
|
|
)
|
|
assert call.kwargs["params"] == {
|
|
"page": 2,
|
|
"pageSize": 20,
|
|
"manualID": "manual",
|
|
}
|
|
|
|
def test_redirect_is_not_followed_and_is_reported_without_query_values(self):
|
|
response = requests.Response()
|
|
response.status_code = 302
|
|
response.url = "https://www.douyu.com/japi/example?request_secret=hidden"
|
|
response.headers["Location"] = (
|
|
"https://www.douyu.com/member/login?ticket=secret"
|
|
)
|
|
self.client.session.request = Mock(return_value=response)
|
|
self.client.logger = Mock()
|
|
|
|
with pytest.raises(
|
|
DouyuActivityError, match="Cookie 已失效，请重新登录"
|
|
) as caught:
|
|
self.client._request(
|
|
"get",
|
|
"https://www.douyu.com/japi/example?request_secret=hidden",
|
|
source="测试活动接口",
|
|
)
|
|
|
|
self.client.session.request.assert_called_once()
|
|
assert not self.client.session.request.call_args.kwargs["allow_redirects"]
|
|
message = str(caught.value)
|
|
assert "method=GET" in message
|
|
assert "url=https://www.douyu.com/japi/example" in message
|
|
assert "location=https://www.douyu.com/member/login" in message
|
|
assert "request_secret" not in message
|
|
assert "ticket=secret" not in message
|
|
assert "已拦截重定向" in self.client.logger.call_args.args[0]
|