chuanqi/cq2_sdk/tests/unit/test_client.py

"""Tests for Cq2Client + LauncherFacade using the REAL protocol (all mocked)."""
import asyncio

import httpx
import pytest

from cq2_sdk.client import Cq2Client, FlowResult, LauncherFacade
from cq2_sdk.login import (
    Box996Login,
    BoxSession,
    CaptchaRequiredError,
    GameSession,
    ServerInfo,
    TodayOpenServer,
)

LOGIN_200 = {
    "code": 200,
    "msg": "登录成功",
    "data": {
        "token": "eyJfake.jwt",
        "userInfo": {"id": 1186115037, "account": "longwu988"},
    },
    "success": True,
}
TODAY_OPEN = {
    "code": 200,
    "msg": "ok",
    "data": [
        {"server_name": "霸业10服", "serverid": 14414736, "open_time": 1,
         "section": " 00:00", "state": 1, "main_serverid": "14414736"},
        {"server_name": "霸业12服", "serverid": 14414738, "open_time": 2,
         "section": " 10:00", "state": 1, "main_serverid": "14414738"},
    ],
}


def make_box() -> Box996Login:
    def handler(request: httpx.Request) -> httpx.Response:
        url = str(request.url)
        if "byPwd" in url:
            return httpx.Response(200, json=LOGIN_200)
        if "getTodayOpen" in url:
            return httpx.Response(200, json=TODAY_OPEN)
        if "EnterGameClient" in url:
            return httpx.Response(200, json={"code": 200, "msg": "", "data": {"download_info": {}}})
        return httpx.Response(404, json={"code": -1, "msg": "no fixture"})

    box = Box996Login()
    box._client = httpx.AsyncClient(transport=httpx.MockTransport(handler), timeout=10)
    return box


# ---- LauncherFacade ----


@pytest.mark.asyncio
async def test_launcher_get_today_servers():
    facade = LauncherFacade(make_box())
    servers = await facade.get_today_servers()
    assert len(servers) == 2
    assert servers[1].server_name == "霸业12服"


@pytest.mark.asyncio
async def test_launcher_enter_game():
    captured = {}

    def handler(request: httpx.Request) -> httpx.Response:
        if "EnterGameClient" in str(request.url):
            captured["body"] = request.content.decode()
            return httpx.Response(200, json={"code": 200, "msg": "", "data": {"ok": 1}})
        return httpx.Response(404, json={})

    box = Box996Login()
    box._client = httpx.AsyncClient(transport=httpx.MockTransport(handler), timeout=10)
    facade = LauncherFacade(box)
    data = await facade.enter_game(14414738)
    assert data == {"ok": 1}
    assert "14414738" in captured["body"]


@pytest.mark.asyncio
async def test_run_until_enter_game_orchestrates_real_login(monkeypatch):
    events = []
    ready = []
    resolved = ServerInfo(
        server_id=14414736,
        main_server_id=14414736,
        name="霸业10服",
        host="game.example",
        port=8041,
        channel_id="110585",
        game_id=206923,
    )
    game_session = GameSession(
        char_id="role-1",
        char_name="Hero",
        session_id=149,
        user_id="user-1",
        world_ready=True,
    )

    async def fake_resolve(server_id, *, channel_id, game_id):
        events.append(("resolve", server_id, channel_id, game_id))
        return resolved

    class FakeTcp:
        def __init__(self, host, port):
            events.append(("tcp", host, port))

        async def connect(self):
            events.append("connect")

        async def start_heartbeat(self):
            events.append("heartbeat")

        async def close(self):
            events.append("close")

    class FakeGameLogin:
        def __init__(self, tcp):
            self.tcp = tcp

        async def login(self, session, server, **kwargs):
            events.append(("login", session.uid, server.server_id, kwargs))
            return game_session

    async def on_ready(tcp, server, result):
        ready.append((tcp, server, result))

    monkeypatch.setattr("cq2_sdk.client.fetch_server_info", fake_resolve)
    monkeypatch.setattr("cq2_sdk.client.TcpClient", FakeTcp)
    monkeypatch.setattr("cq2_sdk.client.GameLogin", FakeGameLogin)
    box = make_box()
    facade = LauncherFacade(
        box,
        BoxSession(uid="user-1", token="token", sdkid="", signkey=""),
        on_ready=on_ready,
    )

    result = await facade.run_until_enter_game(
        zone_index=0,
        channel_id="110585",
        char_name="Hero",
        job=1,
        sex=0,
        timeout=3.0,
    )
    await box.close()

    assert result is game_session
    assert events == [
        ("resolve", 14414736, "110585", 206923),
        ("tcp", "game.example", 8041),
        "connect",
        (
            "login",
            "user-1",
            14414736,
            {"char_name": "Hero", "job": 1, "sex": 0, "timeout": 3.0},
        ),
        "heartbeat",
    ]
    assert ready == [(facade.tcp, resolved, game_session)]
    assert facade.game_session is game_session


@pytest.mark.asyncio
async def test_run_until_enter_game_requires_authenticated_session():
    box = make_box()
    facade = LauncherFacade(box)
    try:
        with pytest.raises(RuntimeError, match="authenticated BoxSession"):
            await facade.run_until_enter_game()
    finally:
        await box.close()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("server_name", "zone_index", "expected_today_calls"),
    [("unsupported-game", 0, 0), ("玖扬复古176", 2, 1)],
)
async def test_invalid_launch_selection_has_no_enter_game_or_tcp_side_effects(
    monkeypatch, server_name, zone_index, expected_today_calls
):
    calls = []

    class Box:
        async def get_today_open(self, game_id):
            calls.append(("today", game_id))
            return [
                TodayOpenServer(
                    server_name="One",
                    serverid=1,
                    open_time=1,
                    section="00:00",
                    state=1,
                    main_serverid="1",
                )
            ]

        async def enter_game_client(self, **kwargs):
            calls.append(("enter", kwargs))

    def forbidden_tcp(*args, **kwargs):
        calls.append(("tcp", args, kwargs))
        raise AssertionError("TCP setup must not run")

    monkeypatch.setattr("cq2_sdk.client.TcpClient", forbidden_tcp)
    facade = LauncherFacade(
        Box(),
        BoxSession(uid="u", token="t", sdkid="", signkey=""),
    )

    with pytest.raises((ValueError, IndexError)):
        await facade.run_until_enter_game(
            server_name=server_name,
            zone_index=zone_index,
        )

    assert sum(call[0] == "today" for call in calls) == expected_today_calls
    assert not any(call[0] in {"enter", "tcp"} for call in calls)


@pytest.mark.asyncio
async def test_initial_tcp_connect_is_bounded_by_login_timeout(monkeypatch):
    server = ServerInfo(server_id=1, name="One", host="game.example", port=8041)
    connect_started = asyncio.Event()
    instances = []

    async def fake_resolve(*args, **kwargs):
        return server

    class StalledTcp:
        def __init__(self, host, port):
            self.cancelled = False
            self.closed = False
            instances.append(self)

        async def connect(self):
            connect_started.set()
            try:
                await asyncio.Event().wait()
            except asyncio.CancelledError:
                self.cancelled = True
                raise

        async def close(self):
            self.closed = True

    monkeypatch.setattr("cq2_sdk.client.fetch_server_info", fake_resolve)
    monkeypatch.setattr("cq2_sdk.client.TcpClient", StalledTcp)
    box = make_box()
    facade = LauncherFacade(
        box,
        BoxSession(uid="u", token="t", sdkid="", signkey=""),
    )
    loop = asyncio.get_running_loop()
    began = loop.time()
    try:
        with pytest.raises(TimeoutError):
            await asyncio.wait_for(
                facade.run_until_enter_game(timeout=0.02),
                timeout=0.2,
            )
    finally:
        await box.close()

    assert connect_started.is_set()
    assert loop.time() - began < 0.15
    assert instances[0].cancelled is True
    assert instances[0].closed is True


@pytest.mark.asyncio
async def test_second_launch_is_rejected_before_external_side_effects(monkeypatch):
    events = []
    server = ServerInfo(server_id=1, name="One", host="game.example", port=8041)
    game_session = GameSession(
        char_id="r1",
        char_name="Hero",
        session_id=149,
        user_id="u1",
        world_ready=True,
    )

    class Box:
        async def get_today_open(self, game_id):
            events.append("today")
            return [
                TodayOpenServer(
                    server_name="One",
                    serverid=1,
                    open_time=1,
                    section="00:00",
                    state=1,
                    main_serverid="1",
                )
            ]

        async def enter_game_client(self, **kwargs):
            events.append("enter")

    async def fake_resolve(*args, **kwargs):
        events.append("resolve")
        return server

    class FakeTcp:
        def __init__(self, host, port):
            events.append("tcp")

        async def connect(self):
            events.append("connect")

        async def start_heartbeat(self):
            events.append("heartbeat")

        async def close(self):
            events.append("close")

    class FakeGameLogin:
        def __init__(self, tcp):
            pass

        async def login(self, *args, **kwargs):
            events.append("login")
            return game_session

    monkeypatch.setattr("cq2_sdk.client.fetch_server_info", fake_resolve)
    monkeypatch.setattr("cq2_sdk.client.TcpClient", FakeTcp)
    monkeypatch.setattr("cq2_sdk.client.GameLogin", FakeGameLogin)
    facade = LauncherFacade(
        Box(),
        BoxSession(uid="u", token="t", sdkid="", signkey=""),
    )

    assert await facade.run_until_enter_game() is game_session
    after_first_launch = list(events)
    with pytest.raises(RuntimeError, match="already owns"):
        await facade.run_until_enter_game()

    assert events == after_first_launch


@pytest.mark.asyncio
async def test_run_until_enter_game_closes_tcp_when_ready_binding_fails(monkeypatch):
    events = []
    server = ServerInfo(
        server_id=1,
        name="One",
        host="game.example",
        port=8041,
    )
    result = GameSession(
        char_id="r1",
        char_name="Hero",
        session_id=1,
        user_id="u1",
        world_ready=True,
    )

    async def fake_resolve(*args, **kwargs):
        return server

    class FakeTcp:
        def __init__(self, host, port):
            pass

        async def connect(self):
            pass

        async def start_heartbeat(self):
            pass

        async def close(self):
            events.append("close")

    class FakeGameLogin:
        def __init__(self, tcp):
            pass

        async def login(self, *args, **kwargs):
            return result

    async def failing_ready(*args):
        raise RuntimeError("bind failed")

    monkeypatch.setattr("cq2_sdk.client.fetch_server_info", fake_resolve)
    monkeypatch.setattr("cq2_sdk.client.TcpClient", FakeTcp)
    monkeypatch.setattr("cq2_sdk.client.GameLogin", FakeGameLogin)
    box = make_box()
    facade = LauncherFacade(
        box,
        BoxSession(uid="u1", token="token", sdkid="", signkey=""),
        on_ready=failing_ready,
    )
    try:
        with pytest.raises(RuntimeError, match="bind failed"):
            await facade.run_until_enter_game()
    finally:
        await box.close()

    assert events == ["close"]
    assert facade.tcp is None


# ---- Cq2Client ----


@pytest.mark.asyncio
async def test_client_password_login_full_flow(monkeypatch):
    box = make_box()
    monkeypatch.setattr(
        "cq2_sdk.client.Box996Login", lambda **kw: box
    )
    async with Cq2Client(account="longwu988", password="Longwu1988") as cli:
        assert cli.box_session is not None
        assert cli.box_session.uid == "1186115037"
        assert cli.launcher is not None
        assert cli.launcher.session is cli.box_session
        servers = await cli.launcher.get_today_servers()
        assert len(servers) == 2


@pytest.mark.asyncio
async def test_client_ready_callback_builds_gameplay_context():
    cli = Cq2Client(account="account", password="password")
    tcp = object()
    server = ServerInfo(
        server_id=7,
        name="Seven",
        host="game.example",
        port=8041,
    )
    game_session = GameSession(
        char_id="role-1",
        char_name="Hero",
        session_id=149,
        user_id="user-1",
        world_ready=True,
    )
    cli.box_session = BoxSession(
        uid="user-1", token="token", sdkid="", signkey=""
    )

    await cli._bind_game_session(tcp, server, game_session)

    assert cli.tcp is tcp
    assert cli.server_info is server
    assert cli.game_session is game_session
    assert cli.task is not None
    assert cli.task.ctx.tcp is tcp
    assert cli.task.ctx.box_session is cli.box_session
    assert cli.task.ctx.server is server
    assert cli.task.ctx.game_session is game_session
    assert cli.task.ctx.movement.tcp is tcp
    assert cli.task.ctx.combat.tcp is tcp
    assert cli.task.ctx.missions.tcp is tcp


def test_client_requires_credentials():
    with pytest.raises(RuntimeError, match="password"):
        import asyncio
        asyncio.run(Cq2Client(account="x").__aenter__())


def test_client_legacy_phone_param():
    cli = Cq2Client(phone="13800138000", password="pw")
    assert cli.account == "13800138000"


def test_flow_result_defaults():
    r = FlowResult()
    assert r.completed_tasks == []
    assert r.elapsed == 0.0