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