chuanqi/protocol_bot/tests/test_codec.py

"""codec 层验证 — 编码向量来自 backup pcap 实测, 帧布局来自 cq2_pure_enter.py。"""

import random
import struct
from pathlib import Path

import pytest

from pbot.codec.bitbuf import decode6, default_table_path, encode6
from pbot.codec.frame import Frame, FrameSplitter, build_frame, parse_frame
from pbot.errors import CodecError
from pbot.types import pack_point

# pcap 心跳帧向量(backup self-test 100% 匹配): 密文如下, 明文构造见 _plain_heartbeat()
_HEARTBEAT_ENC = bytes.fromhex(
    "5d5452415d4c67445d545078495e55525d5452415d5452415d54507a5d5452415d5452415d5452415d54527b5d545241")


def _plain_heartbeat() -> bytes:
    return struct.pack("<8i", 0, 60000, 1789709744, 0, 0, 149, 0, 0) + bytes.fromhex("70000000")


def test_table_exists() -> None:
    """Given: 仓库资产 When: 按默认路径找替换表 Then: 存在且 512B。"""
    path = default_table_path()
    assert Path(path).exists(), f"替换表不在 {path}"
    assert Path(path).stat().st_size >= 512


def test_encode6_matches_pcap_vector() -> None:
    """Given: 实测心跳明文 When: encode6 Then: 与 pcap 抓包字节完全一致。"""
    assert encode6(_plain_heartbeat()) == _HEARTBEAT_ENC


def test_decode6_roundtrip_pcap_vector() -> None:
    """Given: pcap 密文 When: decode6 Then: 还原心跳明文。"""
    assert decode6(_HEARTBEAT_ENC) == _plain_heartbeat()


def test_roundtrip_random_bytes() -> None:
    """Given: 任意字节流 When: encode→decode Then: 恒等。"""
    rng = random.Random(20260923)  # noqa: S311 - 非密码用途, 固定种子
    src = bytes(rng.randrange(256) for _ in range(1000))
    assert decode6(encode6(src)) == src


def test_build_and_parse_frame_roundtrip() -> None:
    """Given: CS3011 走路帧参数 When: build→parse Then: 全字段还原。"""
    wire = build_frame(3011, pack_point(23, 27), 0, 5, pack_point(23, 27),
                       body="等待地图".encode(), tail=7)
    f = parse_frame(wire)
    assert f.msg_id == 3011
    assert f.p1 == 0 and f.p2 == 5 and f.tail == 7
    assert f.body == "等待地图".encode()
    assert f.pre == 0


def test_splitter_handles_tcp_fragmentation() -> None:
    """Given: 两帧粘包并被任意切碎 When: 逐块 feed Then: 恰好产出两帧。"""
    f1 = build_frame(10, 0, 1, 2, 3, b"x" * 10, tail=1)
    f2 = build_frame(51, 0, 4, 5, 6, b"", tail=2)
    blob = f1 + f2
    splitter = FrameSplitter()
    frames: list[Frame] = []
    for i in range(len(blob)):
        frames += splitter.feed(blob[i:i + 1])
    assert [(f.msg_id, f.tail) for f in frames] == [(10, 1), (51, 2)]


def test_splitter_resyncs_on_garbage() -> None:
    """Given: 帧前混入垃圾字节 When: feed Then: 垃圾被跳过, 帧仍解析出。"""
    wire = build_frame(60000, 0, 0, 0, 0, b"", tail=3)
    splitter = FrameSplitter()
    frames = splitter.feed(b"\x00garbage" + wire)
    assert [fr.msg_id for fr in frames] == [60000]
    assert frames[0].tail == 3


def test_parse_frame_rejects_short_input() -> None:
    """Given: 不足 48 字符 When: parse_frame Then: CodecError。"""
    with pytest.raises(CodecError):
        _ = parse_frame(b"ab")