Christian Nennemann
49e8e066d7
Implements quicproquo-py with two transport backends: - Async QUIC transport via aioquic with v2 protobuf wire format - Synchronous Rust FFI transport via CFFI wrapping libquicproquo_ffi Includes manual protobuf encode/decode (no codegen), full RPC coverage (auth, delivery, channels, users, keys, health), PyPI-ready packaging, async echo bot and FFI demo examples, and 15 passing unit tests.
42 lines
927 B
Python
42 lines
927 B
Python
"""Tests for the v2 wire format encoder/decoder."""
|
|
|
|
from quicproquo.wire import (
|
|
HEADER_SIZE,
|
|
encode_frame,
|
|
decode_header,
|
|
HEALTH,
|
|
ENQUEUE,
|
|
)
|
|
|
|
|
|
def test_header_size():
|
|
assert HEADER_SIZE == 10
|
|
|
|
|
|
def test_encode_decode_roundtrip():
|
|
payload = b"\x0a\x05hello"
|
|
frame = encode_frame(HEALTH, 42, payload)
|
|
|
|
assert len(frame) == HEADER_SIZE + len(payload)
|
|
|
|
method_id, req_id, length = decode_header(frame)
|
|
assert method_id == HEALTH
|
|
assert req_id == 42
|
|
assert length == len(payload)
|
|
assert frame[HEADER_SIZE:] == payload
|
|
|
|
|
|
def test_empty_payload():
|
|
frame = encode_frame(ENQUEUE, 1, b"")
|
|
method_id, req_id, length = decode_header(frame)
|
|
assert method_id == ENQUEUE
|
|
assert req_id == 1
|
|
assert length == 0
|
|
|
|
|
|
def test_decode_header_too_short():
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError, match="header too short"):
|
|
decode_header(b"\x00\x01")
|