chuanqi/cq2_sdk/tests/unit/test_server_list.py

import base64
from dataclasses import FrozenInstanceError

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=="
)

INVALID_PADDED_LENGTH_SERVER_LIST = b"NG1lSm5QeWxWnBdj3dyvl1nAc+J5GmVq"


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.state == 1
    assert server.url_suffix == "v"
    assert server.tag == 1
    assert server.gm_url_suffix == ""
    assert server.explicit_ip == ""
    assert server.proxy_port == 9000
    assert server.beta is False
    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.explicit_ip == "1.2.3.4"
    assert server.main_server_id == 8
    assert server.proxy_port is None


def test_resolve_server_expands_long_form_aliases():
    server = resolve_server(
        {
            "srvRoot": "edge.example",
            "serverData": [
                {
                    "srvid": "17",
                    "port": "8041",
                    "state": "4",
                    "urlsuffix": "stable-v2",
                    "mainsrvid": "19",
                    "tag": "featured",
                    "srvname": "Seventeen",
                    "gmurlsuffix": "gm-v2",
                    "srvip": "10.0.0.17",
                    "proxyport": "9001",
                    "beta": "1",
                }
            ],
        },
        17,
        channel_id="channel",
        game_id=42,
    )

    assert server.server_id == 17
    assert server.port == 8041
    assert server.state == 4
    assert server.url_suffix == "stable-v2"
    assert server.main_server_id == 19
    assert server.tag == "featured"
    assert server.name == "Seventeen"
    assert server.gm_url_suffix == "gm-v2"
    assert server.explicit_ip == "10.0.0.17"
    assert server.host == "10.0.0.17"
    assert server.proxy_port == 9001
    assert server.beta is True


@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_decode_rejects_more_than_three_padding_bytes():
    with pytest.raises(ServerListError, match="plaintext length"):
        decode_detailed_server_list(INVALID_PADDED_LENGTH_SERVER_LIST)


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"


@pytest.mark.asyncio
async def test_fetch_server_info_does_not_disable_tls_verification(monkeypatch):
    constructed = []

    class Response:
        content = ENCRYPTED_SERVER_LIST

        def raise_for_status(self):
            return None

    class Client:
        def __init__(self, **kwargs):
            constructed.append(kwargs)

        async def get(self, _url):
            return Response()

        async def aclose(self):
            return None

    monkeypatch.setattr("cq2_sdk.login.server_list.httpx.AsyncClient", Client)

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

    assert constructed[0].get("verify", True) is True


def test_server_info_is_immutable_and_copies_config_versions():
    source_versions = {"cfg_stditem": "hash-1"}
    server = resolve_server(
        {
            "srvRoot": "edge.example",
            "serverData": [{"d": 7, "p": 8041, "s": 1}],
        },
        7,
        channel_id="110585",
        game_id=206923,
    )
    server_with_config = type(server)(
        server_id=server.server_id,
        name=server.name,
        host=server.host,
        port=server.port,
        config_versions=source_versions,
    )

    source_versions["cfg_stditem"] = "mutated"
    with pytest.raises(FrozenInstanceError):
        server_with_config.port = 9000
    with pytest.raises(TypeError):
        server_with_config.config_versions["new"] = "value"
    assert dict(server_with_config.config_versions) == {"cfg_stditem": "hash-1"}