chuanqi/tests/test_network_stubs.py

# tests/test_network_stubs.py
"""NetClientStub + NetworkCtlStub: 真实环回 socket 模拟网关。"""
import socket
import struct
import sys
import threading

import lupa.luajit21 as lj

sys.path.insert(0, "protocol_bot")
from encode6bitbuf_final import encode6  # type: ignore  # noqa: E402

from apilot.stubs.netclient import NetClientStub  # noqa: E402
from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame  # noqa: E402


def _build_frame(msg_id: int, body: bytes) -> bytes:
    """encode6(9×i32 header) + body。字段序对齐 wire.pack_frame。"""
    head = struct.pack("<9i", 0, msg_id, 1, 0, 0, 0, 0, len(body), 1)
    return encode6(head) + body


def _spawn_gateway(handler):
    """启一个临时监听,handler(server_sock) 在调用线程同步完成一次会话。"""
    srv = socket.socket()
    srv.bind(("127.0.0.1", 0))
    srv.listen(1)
    port = srv.getsockname()[1]
    threading.Thread(target=handler, args=(srv,), daemon=True).start()
    return port


def _fake_gateway_echo529(server_sock):
    """收一帧, 回一个 529 帧。body 在线缆上是 raw bytes(仅 header 编码)。"""
    conn, _ = server_sock.accept()
    try:
        conn.recv(65536)  # drain client frame
        conn.sendall(_build_frame(529, b"key32/session123"))
    finally:
        conn.close()
        server_sock.close()


def test_sendmsg_roundtrip_and_dispatch():
    port = _spawn_gateway(_fake_gateway_echo529)

    nc = NetClientStub()
    nc.connect("127.0.0.1", port)
    ctl = NetworkCtlStub(nc)
    ctl.send_msg(2001, 123456, 1, 2, 3, b"uid/key/ch/1/0", 19)

    frames = nc.tick(1.0)
    assert any(f.msg_id == 529 for f in frames), f"expected 529 in {frames}"

    rt = lj.LuaRuntime()
    rt.execute("""
got = nil
function on_529(msg)
    local h = msg:GetHeader()
    local data = msg:GetData()
    got = h.msgId .. ":" .. data:ReadString(5)
end
""")
    ctl.register_lua_handler(529, rt.eval("on_529"))
    target = next(f for f in frames if f.msg_id == 529)
    dispatch_frame(rt, ctl, target)
    assert rt.eval("got") == "529:key32"
    nc.disconnect()


def test_register_unregister():
    """register_lua_handler 覆盖, unregister_all 清空。"""
    nc = NetClientStub()
    ctl = NetworkCtlStub(nc)
    rt = lj.LuaRuntime()
    rt.execute("function h(msg) end")
    cb = rt.eval("h")
    ctl.register_lua_handler(10, cb)
    ctl.register_lua_handler(10, cb)  # 覆盖
    assert ctl.handlers[10] is cb
    ctl.unregister_all()
    assert ctl.handlers == {}


def test_dispatch_no_handler_is_noop():
    """无 handler 的帧不抛异常。"""
    nc = NetClientStub()
    ctl = NetworkCtlStub(nc)
    from apilot.wire import Frame
    rt = lj.LuaRuntime()
    dispatch_frame(rt, ctl, Frame(recog=0, msg_id=99999, p1=0, p2=0, p3=0, tail=1, body=b""))


def test_is_connected_and_disconnect():
    nc = NetClientStub()
    assert not nc.is_connected()
    port = _spawn_gateway(_fake_gateway_echo529)
    nc.connect("127.0.0.1", port)
    assert nc.is_connected()
    nc.disconnect()
    assert not nc.is_connected()