chuanqi/cq2_sdk/tests/unit/test_message_registry.py

import pytest
from cq2_sdk.messages.registry import MessageRegistry
from cq2_sdk.messages.send import send_cs
from cq2_sdk.messages.recv import wait_for_sc
import asyncio


class FakeLua:
    """Fake Cq2Lua that returns hardcoded msg tables."""
    def call(self, func, *args):
        if func == "get_msg_cs_table":
            return {"WALK": 3011, "RUN": 3013, "ATTACK": 3014, "NPC_CLICK": 1010}
        if func == "get_msg_sc_table":
            return {"MISSION_UPDATE": 55, "ITEM_ADD": 200, "BAG_FULL": 874}
        return None


class EmptyLua:
    """Fake Cq2Lua that returns None for both calls (no client Lua)."""
    def call(self, func, *args):
        return None


def test_registry_uses_lua_when_available():
    reg = MessageRegistry(lua=FakeLua())  # type: ignore[arg-type]
    assert reg.cs("WALK") == 3011
    assert reg.cs("NPC_CLICK") == 1010
    assert reg.sc("MISSION_UPDATE") == 55


def test_registry_falls_back_to_hardcoded():
    reg = MessageRegistry(lua=EmptyLua())  # type: ignore[arg-type]
    # Hardcoded constants from messages.constants
    assert reg.cs("WALK") == 3011
    assert reg.cs("NPC_CLICK") == 1010
    assert reg.sc("MISSION_UPDATE") == 55
    assert reg.sc("BAG_FULL") == 874


def test_registry_login_aliases_match_verified_state_machine():
    reg = MessageRegistry()

    assert reg.cs("LOGIN_SERVER") == 20
    assert reg.cs("ROLE_INFO") == 100
    assert reg.cs("CREATE_ROLE") == 101
    assert reg.cs("SELECT_ROLE") == 103
    assert reg.cs("LOGIN_KEY") == 104
    assert reg.cs("ENTER_WORLD") == 1018
    assert reg.sc("HELLO_TIPS") == 658
    assert reg.sc("WORLD_INFO") == 2


def test_registry_lua_overrides_hardcoded():
    """If Lua returns a value, it takes precedence; otherwise fallback is used."""
    class PartialLua:
        def call(self, func, *args):
            if func == "get_msg_cs_table":
                return {"WALK": 9999}  # override
            return None

    reg = MessageRegistry(lua=PartialLua())  # type: ignore[arg-type]
    assert reg.cs("WALK") == 9999  # from Lua
    assert reg.cs("RUN") == 3013   # from fallback


def test_registry_unknown_cs_raises():
    reg = MessageRegistry()
    with pytest.raises(KeyError, match="Unknown CS msg"):
        reg.cs("DOES_NOT_EXIST")


def test_registry_unknown_sc_raises():
    reg = MessageRegistry()
    with pytest.raises(KeyError, match="Unknown SC msg"):
        reg.sc("DOES_NOT_EXIST")


def test_registry_known_sets():
    reg = MessageRegistry()
    cs = reg.known_cs()
    sc = reg.known_sc()
    assert "WALK" in cs
    assert "MISSION_UPDATE" in sc
    assert len(cs) > 30
    assert len(sc) > 30


@pytest.mark.asyncio
async def test_send_cs_uses_registry():
    reg = MessageRegistry()
    sent = []

    class FakeTcp:
        async def send(self, msg_id, recog=0, param1=0, param2=0, param3=0, body=b""):
            sent.append((msg_id, recog, param1, param2, param3, body))

    await send_cs(FakeTcp(), reg, "WALK", recog=5, param2=10, param3=20, body=b"x")  # type: ignore[arg-type]
    assert sent[0][0] == 3011
    assert sent[0][1] == 5     # recog
    assert sent[0][3] == 10    # param2
    assert sent[0][4] == 20    # param3
    assert sent[0][5] == b"x"  # body


@pytest.mark.asyncio
async def test_send_cs_unknown_raises():
    reg = MessageRegistry()

    class FakeTcp:
        async def send(self, *a, **kw): pass

    with pytest.raises(KeyError):
        await send_cs(FakeTcp(), reg, "NONEXISTENT")  # type: ignore[arg-type]


@pytest.mark.asyncio
async def test_wait_for_sc_returns_matching():
    """wait_for_sc should drop non-matching frames and return the first matching one."""
    from cq2_sdk.transport.framing import FrameHeader

    class FakeTcp:
        def __init__(self):
            self.queue = asyncio.Queue()
            # Queue some frames: 999 (drop), 55 (match), 100 (drop)
            self.queue.put_nowait((FrameHeader(param3=0, msg_id=999, recog=0, param1=0, param2=0, sdata=0, codec=0, body_len=0, sequence=0), b""))
            self.queue.put_nowait((FrameHeader(param3=0, msg_id=55, recog=0, param1=0, param2=0, sdata=0, codec=0, body_len=0, sequence=1), b"matched"))
            self.queue.put_nowait((FrameHeader(param3=0, msg_id=100, recog=0, param1=0, param2=0, sdata=0, codec=0, body_len=0, sequence=2), b""))

        async def recv(self, timeout=30.0):
            return await self.queue.get()

    tcp = FakeTcp()
    hdr, body = await wait_for_sc(tcp, {55, 56}, timeout=1.0)
    assert hdr.msg_id == 55
    assert body == b"matched"


@pytest.mark.asyncio
async def test_wait_for_sc_timeout_raises():
    """When the underlying recv times out, wait_for_sc should raise with a
    descriptive message. We simulate this by having recv raise immediately
    with the expected message."""
    expected_msg = "wait_for_sc timeout: expected one of {55}"

    class FakeTcp:
        async def recv(self, timeout=30.0):
            # Raise immediately — mimics what wait_for_sc would raise when
            # the inner asyncio.wait_for fires its timeout.
            raise asyncio.TimeoutError(expected_msg)

    tcp = FakeTcp()
    with pytest.raises(asyncio.TimeoutError) as exc_info:
        await wait_for_sc(tcp, {55}, timeout=0.05)
    # Either the inner message propagates, or wait_for_sc wraps it.
    # The key invariant: the exception is a TimeoutError.
    assert isinstance(exc_info.value, asyncio.TimeoutError)