chuanqi/cq2_sdk/tests/unit/test_framing.py

import pytest
from cq2_sdk.transport.framing import (
    HEADER_SIZE,
    PLAIN_HEADER_SIZE,
    WIRE_HEADER_SIZE,
    FrameHeader,
    decode_frame,
    decode_header,
    encode_frame,
)
from cq2_sdk.transport.encode6 import decode6, encode6
from cq2_sdk.messages import constants as C


HEARTBEAT_HEADER = bytes.fromhex(
    "0000000060ea0000b0cdac6a000000000000000095000000000000000000000070000000"
)
HEARTBEAT_WIRE_HEADER = bytes.fromhex(
    "5d5452415d4c67445d545078495e55525d5452415d545241"
    "5d54507a5d5452415d5452415d5452415d54527b5d545241"
)


def test_encode6_matches_captured_heartbeat_header():
    assert encode6(HEARTBEAT_HEADER) == HEARTBEAT_WIRE_HEADER


def test_decode6_restores_captured_heartbeat_header():
    assert decode6(HEARTBEAT_WIRE_HEADER, expected_size=36) == HEARTBEAT_HEADER


def test_decode_captured_heartbeat_uses_verified_header_field_order():
    header = decode_header(HEARTBEAT_WIRE_HEADER)

    assert header.param3 == 0
    assert header.msg_id == C.CS_60000
    assert header.recog == 1789709744
    assert header.param1 == 0
    assert header.param2 == 0
    assert header.sdata == 149
    assert header.codec == 0
    assert header.body_len == 0
    assert header.sequence == 112


def test_decode6_rejects_non_wire_character():
    with pytest.raises(ValueError, match="outside 6-bit wire range"):
        decode6(b"<~")


def test_header_is_48_bytes_on_wire():
    raw = encode_frame(1, 0, 0, 0, 0, b"")
    assert len(raw) == WIRE_HEADER_SIZE
    assert PLAIN_HEADER_SIZE == 36
    assert WIRE_HEADER_SIZE == 48
    assert HEADER_SIZE == WIRE_HEADER_SIZE


def test_encode_decode_roundtrip():
    body = b"hello"
    raw = encode_frame(
        C.CS_2001,
        recog=900000,
        param1=12345,
        param2=0,
        param3=7,
        body=body,
        sdata=11,
        sequence=42,
    )
    assert len(raw) == WIRE_HEADER_SIZE + len(body)
    hdr, decoded_body = decode_frame(raw)
    assert hdr.param3 == 7
    assert hdr.msg_id == C.CS_2001
    assert hdr.recog == 900000
    assert hdr.param1 == 12345
    assert hdr.param2 == 0
    assert hdr.sdata == 11
    assert hdr.body_len == 5
    assert hdr.codec == 0
    assert hdr.sequence == 42
    assert decoded_body == body


def test_body_length_field_matches():
    body = b"x" * 100
    raw = encode_frame(1, 0, 0, 0, 0, body)
    hdr, _ = decode_frame(raw)
    assert hdr.body_len == 100


def test_empty_body():
    raw = encode_frame(C.CS_100, 0, 0, 0, 0, b"")
    hdr, body = decode_frame(raw)
    assert body == b""


def test_truncated_body_raises():
    raw = encode_frame(1, 0, 0, 0, 0, b"hello world")
    truncated = raw[:WIRE_HEADER_SIZE + 4]  # cut off in the middle of the body
    with pytest.raises(ValueError, match="truncated"):
        decode_frame(truncated)


def test_too_short_raises():
    with pytest.raises(ValueError, match="too short"):
        decode_header(b"\x00" * 10)


def test_negative_recog_allowed():
    # recog is signed int; can be negative for error codes
    raw = encode_frame(1, recog=-1, param1=0, param2=0, param3=0, body=b"")
    hdr, _ = decode_frame(raw)
    assert hdr.recog == -1


def test_unsupported_body_codec_raises():
    raw = encode_frame(1, 0, 0, 0, 0, b"compressed", codec=1)
    with pytest.raises(ValueError, match="unsupported body codec: 1"):
        decode_frame(raw)


def test_decode_header_reads_the_first_wire_header_from_a_complete_frame():
    raw = encode_frame(529, 1, 0, 0, 0, b"response", sequence=8)
    header = decode_header(raw)
    assert header.msg_id == 529
    assert header.sequence == 8


def test_constants_loaded():
    # Smoke check: a few known constants are importable and distinct
    assert C.CS_2001 == 2001
    assert C.SC_529 == 529
    assert C.SC_530 == 530
    assert C.ALL_TASKS == [f"t{n:02d}" for n in range(1, 14)]