chuanqi/tests/test_wire.py

# tests/test_wire.py
import pytest
from apilot.wire import Frame, pack_frame, unpack_stream

def test_frame_roundtrip():
    f = Frame(recog=900000, msg_id=20, p1=123, p2=14481607, p3=8043,
              tail=7, body=b"*#acct/role/12345/9/1/0/110585/0/1/1/1/1")
    packed = pack_frame(f)
    frames, rest = unpack_stream(packed)
    assert rest == b""
    assert len(frames) == 1
    g = frames[0]
    assert (g.recog, g.msg_id, g.p1, g.p2, g.p3, g.tail, g.body) == \
           (f.recog, f.msg_id, f.p1, f.p2, f.p3, f.tail, f.body)

def test_unpack_partial_packet():
    f = Frame(recog=0, msg_id=529, p1=0, p2=0, p3=0, tail=1, body=b"key32abc")
    packed = pack_frame(f)
    frames, rest = unpack_stream(packed[:-10])   # 掐掉10字节
    assert frames == [] and len(rest) == len(packed) - 10

def test_pack_matches_reference_impl():
    """与 cq2_pure_enter.send() 的原始拼包逐字节一致(防字段序回归)。"""
    import struct, random
    sys_path_hack = __import__("sys"); sys_path_hack.path.insert(0, "protocol_bot")
    from encode6bitbuf_final import encode6
    random.seed(42)
    body = b"*#1186115037/role/1/2/3/1/0/110585/0/1/1/1/1"
    ref = encode6(struct.pack("<9i", 0, 20, 900000, 111, 14481607, 8043, 0, len(body), 9)) + body
    mine = pack_frame(Frame(recog=900000, msg_id=20, p1=111, p2=14481607, p3=8043, tail=9, body=body))
    assert mine == ref

def test_recv_buffered_across_calls():
    """帧跨两次 recv() 到达不能丢(Wire 自持 buffer,rest 留存到下次)。"""
    import socket, struct, threading, time
    sys_path_hack = __import__("sys"); sys_path_hack.path.insert(0, "protocol_bot")
    from encode6bitbuf_final import encode6
    from apilot.wire import Wire

    srv = socket.socket(); srv.bind(("127.0.0.1", 0)); srv.listen(1)
    port = srv.getsockname()[1]

    body = b"key32/session123"
    head = struct.pack("<9i", 0, 529, 1, 0, 0, 0, 0, len(body), 1)
    wire_bytes = encode6(head) + body
    mid = len(wire_bytes) // 2

    def gateway():
        conn, _ = srv.accept()
        try:
            conn.setblocking(False)
            try:
                conn.recv(65536)
            except (BlockingIOError, OSError):
                pass
            conn.setblocking(True)
            conn.sendall(wire_bytes[:mid])  # 第一半
            time.sleep(0.2)
            conn.sendall(wire_bytes[mid:])  # 第二半
            time.sleep(0.2)
        finally:
            conn.close()
            srv.close()

    threading.Thread(target=gateway, daemon=True).start()

    w = Wire("127.0.0.1", port, timeout=2.0)
    f1 = w.recv(0.3)            # 只能拿到第一半
    f2 = w.recv(0.3)            # 第二半在此次补齐
    w.close()
    all_frames = f1 + f2
    assert len([f for f in all_frames if f.msg_id == 529]) == 1, f"got {all_frames}"
    g = next(f for f in all_frames if f.msg_id == 529)
    assert g.body == body