chuanqi/cq2_sdk/tests/unit/test_server_list.py

import base64

import pytest

from cq2_sdk.login.server_list import (
    ServerListError,
    decode_detailed_server_list,
    fetch_server_info,
    resolve_server,
)


ENCRYPTED_SERVER_LIST = (
    b"NG1lSm5QeWzEk/B+5xXXfjnMHIpGB/PPbL/y7tEPyFCzWvBJq2fmPD5PiYHcKj/FZ"
    b"+tehXBa0XEMs9YMS3E1N7HCzgBWTVsHbF5MEs9KDXnW4j1XaHjjnjRVMP2z94vFm"
    b"LOtIDVedS9/y8FZfCpM1CigHp3uJMWqZDWQMLiTVL9RZOrKik67VPrH3MBdtPhLRb"
    b"t00X1Uyd+JuagpL6xRX4nTFP3VdO9TWOvsw/ZqOK0C54GFugWGMAjsydz5A3A/mf"
    b"/x+A=="
)


def test_decode_and_resolve_compact_server_fields():
    data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)

    assert data["srvRoot"] == "edge.example"
    server = resolve_server(data, 7, channel_id="110585", game_id=206923)
    assert server.server_id == 7
    assert server.main_server_id == 9
    assert server.name == "Seven"
    assert server.host == "s9-edge.example"
    assert server.port == 8041
    assert server.proxy_port == 9000
    assert server.channel_id == "110585"
    assert server.game_id == 206923


def test_resolve_server_prefers_explicit_ip():
    data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)

    server = resolve_server(data, 8, channel_id="c", game_id=1)

    assert server.host == "1.2.3.4"
    assert server.main_server_id == 8
    assert server.proxy_port is None


@pytest.mark.parametrize(
    ("payload", "message"),
    [
        (b"not-base64!", "Base64"),
        (base64.b64encode(b"wrong-sign"), "marker"),
        (base64.b64encode(b"4meJnPyl" + b"short"), "XXTEA"),
    ],
)
def test_decode_rejects_malformed_payload(payload, message):
    with pytest.raises(ServerListError, match=message):
        decode_detailed_server_list(payload)


def test_resolve_server_rejects_unknown_server_id():
    data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)

    with pytest.raises(ServerListError, match="server 999"):
        resolve_server(data, 999, channel_id="110585", game_id=206923)


@pytest.mark.asyncio
async def test_fetch_server_info_uses_channel_specific_endpoint():
    seen = []

    class Response:
        content = ENCRYPTED_SERVER_LIST

        def raise_for_status(self):
            return None

    class Client:
        async def get(self, url):
            seen.append(url)
            return Response()

    server = await fetch_server_info(
        7,
        channel_id="110585",
        game_id=206923,
        client=Client(),
    )

    assert seen == [
        "https://list-box.dhsf.xqhuyu.com/serverlist/110585/serverlist.txt"
    ]
    assert server.host == "s9-edge.example"