live-hub-py/tests/test_huya_exchange_proxy.py

"""Huya elite exchange proxy scoping and transport tests."""

from __future__ import annotations

import urllib.request
from types import SimpleNamespace
from unittest.mock import Mock

from core.huya.http_client import HuyaHttpClient
from web.backend.models import HuyaAccount, ProxyConfig
from web.backend.services import huya_points_exchange as exchange
from web.backend.services import huya_runner_goods
from web.backend.services.huya_runner import HuyaBatchRunner


def _cfg(**overrides) -> ProxyConfig:
    values = {
        "enabled": False,
        "api_url": "",
        "http": "",
        "https": "",
        "whitelist_enabled": False,
        "whitelist_platform": "xiequ",
        "whitelist_credentials": None,
        "whitelist_uid": "",
        "whitelist_ukey": "",
    }
    values.update(overrides)
    cfg = ProxyConfig()
    for key, value in values.items():
        setattr(cfg, key, value)
    return cfg


def _runner(cfg: ProxyConfig, task_type: str = "exchange_goods"):
    db = Mock()
    db.query.return_value.first.return_value = cfg
    return HuyaBatchRunner(
        db=db,
        batch_id="proxy-test",
        task_type=task_type,
        payload={"product_id": "hy-8315982"},
    )


def _direct_client():
    client = Mock()
    client.get_redeem_need_goods.return_value = SimpleNamespace(
        code=200,
        message="",
        value=lambda *path: {
            ("tag2",): 20,
            ("tag3",): 20,
            ("tag4",): 0,
            ("tag5",): 1,
            ("tag6",): 1,
            ("tag7",): [],
        }.get(path),
        to_dict=dict,
    )
    client.get_goods_info.return_value = SimpleNamespace(
        code=200,
        message="",
        to_dict=lambda: {
            "sku_id": 7747527,
            "name": "gift",
            "price": 20,
            "goods_type": 5,
            "virtual_type": 507,
        },
    )
    client.create_order_v6.return_value = SimpleNamespace(
        code=200,
        message="",
        to_dict=lambda: {"order_id": 101},
    )
    return client


def _mock_live_goods(monkeypatch):
    monkeypatch.setattr(
        huya_runner_goods,
        "fetch_points_center_catalog",
        lambda *_args, **_kwargs: SimpleNamespace(
            goods=[
                {
                    "product_id": "hy-8315982",
                    "name": "gift",
                    "price": 20,
                    "remain_text": "100",
                }
            ],
            message="已拉取 1 个礼包",
        ),
    )


def test_non_exchange_task_does_not_read_proxy_config():
    db = Mock()
    runner = HuyaBatchRunner(
        db=db,
        batch_id="query-test",
        task_type="query_points",
    )

    db.query.assert_not_called()
    assert runner._proxy_cfg is None
    assert runner._http_client().proxies == {}


def test_static_proxy_is_used_for_entire_exchange_task(monkeypatch):
    runner = _runner(
        _cfg(
            enabled=True,
            http="http://10.0.0.1:8080",
            https="http://10.0.0.1:8080",
        )
    )
    client = _direct_client()
    _mock_live_goods(monkeypatch)
    calls = []
    marks = []

    monkeypatch.setattr(
        runner,
        "_http_client",
        lambda **kwargs: calls.append(kwargs) or client,
    )
    monkeypatch.setattr(
        runner,
        "_mark_task",
        lambda _db, _task, status, message, result=None: marks.append(
            (status, message, result)
        ),
    )

    runner._execute_exchange_goods(
        None,
        SimpleNamespace(id=1),
        HuyaAccount(uid="100", cookie="yyuid=100"),
        {"uid": 100, "cookie": "yyuid=100"},
        {
            "room_pid": "123",
            "pay_channel": "Zfb",
            "exchange_use_proxy": "1",
        },
    )

    assert calls == [
        {
            "proxies": {
                "http": "http://10.0.0.1:8080",
                "https": "http://10.0.0.1:8080",
            }
        }
    ]
    assert marks[-1][0] == "success"
    assert marks[-1][2]["proxy_used"] is True
    assert marks[-1][2]["proxy_source"] == "static"


def test_api_proxy_is_fetched_once_per_exchange_task(monkeypatch):
    runner = _runner(_cfg(enabled=True, api_url="https://proxy.example/fetch"))
    fetcher = Mock()
    fetcher.fetch_new_proxy.return_value = "http://10.0.0.2:3128"
    runner._proxy_fetcher = fetcher
    client = _direct_client()
    _mock_live_goods(monkeypatch)
    calls = []
    marks = []

    monkeypatch.setattr(
        runner,
        "_http_client",
        lambda **kwargs: calls.append(kwargs) or client,
    )
    monkeypatch.setattr(
        runner,
        "_mark_task",
        lambda _db, _task, status, message, result=None: marks.append(
            (status, message, result)
        ),
    )

    runner._execute_exchange_goods(
        None,
        SimpleNamespace(id=2),
        HuyaAccount(uid="100", cookie="yyuid=100"),
        {"uid": 100, "cookie": "yyuid=100"},
        {
            "room_pid": "123",
            "pay_channel": "Zfb",
            "exchange_use_proxy": "1",
        },
    )

    fetcher.fetch_new_proxy.assert_called_once_with(max_attempts=3)
    assert calls[0]["proxies"] == {
        "http": "http://10.0.0.2:3128",
        "https": "http://10.0.0.2:3128",
    }
    assert marks[-1][2]["proxy_source"] == "api"


def test_enabled_proxy_without_source_fails_exchange_before_request(monkeypatch):
    runner = _runner(_cfg(enabled=True))
    marks = []

    monkeypatch.setattr(
        exchange,
        "prepare_exchange",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            AssertionError("exchange must not run without a proxy")
        ),
    )
    monkeypatch.setattr(
        runner,
        "_mark_task",
        lambda _db, _task, status, message, result=None: marks.append(
            (status, message, result)
        ),
    )

    runner._execute_exchange_goods(
        None,
        SimpleNamespace(id=3),
        HuyaAccount(uid="100", cookie="yyuid=100"),
        {"uid": 100, "cookie": "yyuid=100"},
        {"room_pid": "123", "exchange_use_proxy": "1"},
    )

    assert marks == [
        (
            "failed",
            "兑换代理不可用：已启用代理，但未配置静态代理或代理 API",
            {"proxy_used": False, "proxy_source": "error"},
        )
    ]


def test_exchange_proxy_switch_defaults_to_direct_connection(monkeypatch):
    runner = _runner(
        _cfg(
            enabled=True,
            http="http://10.0.0.1:8080",
            https="http://10.0.0.1:8080",
        )
    )
    client = _direct_client()
    _mock_live_goods(monkeypatch)
    calls = []
    marks = []

    monkeypatch.setattr(
        runner,
        "_http_client",
        lambda **kwargs: calls.append(kwargs) or client,
    )
    monkeypatch.setattr(
        runner,
        "_exchange_proxies",
        lambda: (_ for _ in ()).throw(
            AssertionError("proxy switch is off; proxy config must not be read")
        ),
    )
    monkeypatch.setattr(
        runner,
        "_mark_task",
        lambda _db, _task, status, message, result=None: marks.append(
            (status, message, result)
        ),
    )

    runner._execute_exchange_goods(
        None,
        SimpleNamespace(id=4),
        HuyaAccount(uid="100", cookie="yyuid=100"),
        {"uid": 100, "cookie": "yyuid=100"},
        {"room_pid": "123", "pay_channel": "Zfb"},
    )

    assert calls == [{"proxies": None}]
    assert marks[-1][0] == "success"
    assert marks[-1][2]["proxy_used"] is False
    assert marks[-1][2]["proxy_source"] == "disabled"


def test_huya_http_client_routes_methods_through_configured_opener(monkeypatch):
    opener = Mock()
    opener.open.return_value = SimpleNamespace()
    monkeypatch.setattr(urllib.request, "build_opener", lambda _handler: opener)

    client = HuyaHttpClient(
        logger=lambda _message: None,
        proxies={"http": "http://10.0.0.3:8080"},
    )
    request = urllib.request.Request("https://cdnws.api.huya.com/")

    assert client._urlopen(request, timeout=5) is opener.open.return_value
    opener.open.assert_called_once_with(request, timeout=5)


def test_exchange_rpc_and_payment_both_use_proxied_transport(monkeypatch):
    client = HuyaHttpClient(
        logger=lambda _message: None,
        proxies={"http": "http://10.0.0.4:8080"},
    )
    calls = []

    def stop_request(request, timeout):
        calls.append((request.full_url, timeout))
        raise RuntimeError("stop before network")

    monkeypatch.setattr(client, "_urlopen", stop_request)

    assert (
        client.get_redeem_need_goods(
            uid=100,
            guid="",
            cookie="yyuid=100",
        )
        is None
    )
    assert (
        client.pay_order_submit(
            uid=100,
            guid="",
            cookie="yyuid=100",
            order_id=101,
        )
        is None
    )

    assert len(calls) == 2
    assert all(url.startswith("https://cdnws.api.huya.com/") for url, _ in calls)