live-hub-py/tests/test_douyu_elite_exchange_flow.py

"""精英手册兑换：csrf 复用、新链路接口与浏览器状态机路由的测试。"""

from unittest.mock import Mock, call

import pytest

from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError


class TestCsrfReuse:
    def test_reuses_cookie_token_without_any_http_call(self):
        client = DouyuActivityClient("acf_uid=100; cvl_csrf_token=existing-token")
        client._request_json = Mock()

        assert client.csrf_token() == "existing-token"
        client._request_json.assert_not_called()

    def test_generates_csrf_when_cookie_missing(self):
        client = DouyuActivityClient("acf_uid=100")

        def fake_generate(*_args, **_kwargs):
            # 模拟 generateCsrf 响应把新 token 写回 Cookie
            client._set_cookie_value("cvl_csrf_token", "new-token")
            return {"error": 0}

        client._request_json = Mock(side_effect=fake_generate)

        assert client.csrf_token() == "new-token"
        client._request_json.assert_called_once()

    def test_force_refresh_rotates_token(self):
        client = DouyuActivityClient("acf_uid=100; cvl_csrf_token=old-token")
        client._request_json = Mock(return_value={"error": 0})
        client._set_cookie_value("cvl_csrf_token", "rotated-token")

        assert client.csrf_token(force_refresh=True) == "rotated-token"

    def test_refreshes_csrf_once_on_token_error(self):
        client = DouyuActivityClient("acf_uid=100")
        client.csrf_token = Mock(side_effect=["bad-token", "good-token"])
        client._request_json = Mock(
            side_effect=[
                {"error": 1, "msg": "csrfToken 校验失败"},
                {"error": 0, "data": {"orderId": "9", "expireSeconds": "300"}},
            ]
        )

        result = client.create_exchange_order(manual_id="m", rid="r", commodity_id="c")

        assert result["order_id"] == "9"
        assert client.csrf_token.call_args_list == [call(), call(force_refresh=True)]

    def test_no_refresh_on_regular_error(self):
        client = DouyuActivityClient("acf_uid=100")
        client.csrf_token = Mock(return_value="t")
        client._request_json = Mock(
            return_value={"error": 1, "msg": "需要开通精英手册，才可以兑换本道具哦"}
        )

        with pytest.raises(DouyuActivityError, match="需要开通精英手册"):
            client.create_exchange_order(manual_id="m", rid="r", commodity_id="c")
        client.csrf_token.assert_called_once_with()

    def test_pay_referer_includes_room_id_like_browser(self):
        client = DouyuActivityClient("acf_uid=100")
        client.csrf_token = Mock(return_value="t")
        client._request_json = Mock(
            return_value={"error": 0, "data": {"exchangeId": "11098933"}}
        )

        client.pay_exchange_order(manual_id="m", order_id="5874", rid="9263298")

        headers = client._request_json.call_args.kwargs["headers"]
        assert (
            headers["Referer"]
            == "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId=9263298"
        )


class TestExchangePlan:
    """对齐浏览器兑换按钮状态机（$e / Ve / Ge）的路由判定。"""

    @staticmethod
    def detail(**overrides):
        base = {
            "status": 0,
            "storeStatus": 1,
            "openStatus": 1,
            "roomLimited": 0,
            "isLimitRoom": 0,
            "eliteLimited": 0,
            "subscribeType": 0,
            "exchangeInfo": {"userStatus": 0},
            "preExchange": {"isExists": 0, "expiringTime": "0"},
            "batchExchange": 0,
        }
        base.update(overrides)
        return base

    def test_normal_item_uses_classic_flow(self):
        plan = DouyuActivityClient.resolve_exchange_plan(self.detail())
        assert plan["action"] == "classic"
        assert plan["max_num"] == 1

    def test_pre_exchange_order_still_classic(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(preExchange={"isExists": 1})
        )
        assert plan["action"] == "classic"

    def test_reached_exchange_limit_blocked(self):
        plan = DouyuActivityClient.resolve_exchange_plan(self.detail(status=1))
        assert plan["action"] == "blocked"
        assert "上限" in plan["text"]

    def test_no_stock_blocked(self):
        plan = DouyuActivityClient.resolve_exchange_plan(self.detail(storeStatus=0))
        assert plan["action"] == "blocked"
        assert "无库存" in plan["text"]

    def test_room_limited_and_not_in_room_blocked(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(roomLimited=1, isLimitRoom=0)
        )
        assert plan["action"] == "blocked"
        assert "限定房间" in plan["text"]

    def test_not_open_subscribe_item_subscribes(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(openStatus=0, subscribeType=1)
        )
        assert plan["action"] == "subscribe"

    def test_not_open_pre_exchange_item_pre_exchanges(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(openStatus=0, subscribeType=2)
        )
        assert plan["action"] == "pre_exchange"

    def test_not_open_plain_item_blocked(self):
        plan = DouyuActivityClient.resolve_exchange_plan(self.detail(openStatus=0))
        assert plan["action"] == "blocked"
        assert "未开启" in plan["text"]

    def test_elite_limited_requires_manual(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(eliteLimited=1), manual_type=0
        )
        assert plan["action"] == "blocked"
        assert "精英手册" in plan["text"]
        # 已开通手册(manualType=1)不受精英专享限制
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(eliteLimited=1), manual_type=1
        )
        assert plan["action"] == "classic"

    def test_pre_exchanged_waits_for_open(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(subscribeType=2, exchangeInfo={"userStatus": 1})
        )
        assert plan["action"] == "wait"

    def test_batch_exchange_clamps_num(self):
        plan = DouyuActivityClient.resolve_exchange_plan(
            self.detail(batchExchange=1), batch_num_limit="5"
        )
        assert plan["max_num"] == 5
        assert plan["batch_exchange"]