feat(sdk): add Python SDK with QUIC and FFI transport backends
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.
This commit is contained in:
89
sdks/python/tests/test_proto.py
Normal file
89
sdks/python/tests/test_proto.py
Normal file
@@ -0,0 +1,89 @@
|
||||
"""Tests for the manual protobuf encode/decode layer."""
|
||||
|
||||
from quicproquo.proto import (
|
||||
encode_health,
|
||||
decode_health_response,
|
||||
encode_resolve_user,
|
||||
decode_resolve_user_response,
|
||||
encode_enqueue,
|
||||
decode_enqueue_response,
|
||||
encode_create_channel,
|
||||
decode_create_channel_response,
|
||||
encode_opaque_login_start,
|
||||
decode_opaque_login_start_response,
|
||||
encode_opaque_login_finish,
|
||||
decode_opaque_login_finish_response,
|
||||
_encode_string_field,
|
||||
_encode_bytes_field,
|
||||
_encode_varint_field,
|
||||
)
|
||||
|
||||
|
||||
def test_health_empty():
|
||||
assert encode_health() == b""
|
||||
|
||||
|
||||
def test_health_response_decode():
|
||||
# Manually build a HealthResponse: status="ok" (field 1), version="1.0" (field 3)
|
||||
data = _encode_string_field(1, "ok") + _encode_string_field(3, "1.0")
|
||||
info = decode_health_response(data)
|
||||
assert info["status"] == "ok"
|
||||
assert info["version"] == "1.0"
|
||||
assert info["uptime_secs"] == 0
|
||||
|
||||
|
||||
def test_resolve_user_roundtrip():
|
||||
encoded = encode_resolve_user("alice")
|
||||
# The encoded form is a string field (field 1).
|
||||
assert b"alice" in encoded
|
||||
|
||||
# Build a fake response: identity_key = b"\x01\x02\x03"
|
||||
resp = _encode_bytes_field(1, b"\x01\x02\x03")
|
||||
key, proof = decode_resolve_user_response(resp)
|
||||
assert key == b"\x01\x02\x03"
|
||||
assert proof == b""
|
||||
|
||||
|
||||
def test_enqueue_roundtrip():
|
||||
encoded = encode_enqueue(
|
||||
recipient_key=b"\xaa\xbb",
|
||||
payload=b"hello",
|
||||
ttl_secs=60,
|
||||
)
|
||||
assert b"hello" in encoded
|
||||
|
||||
# Build a fake response: seq=42, delivery_proof=b"\xff"
|
||||
resp = _encode_varint_field(1, 42) + _encode_bytes_field(2, b"\xff")
|
||||
seq, proof, dup = decode_enqueue_response(resp)
|
||||
assert seq == 42
|
||||
assert proof == b"\xff"
|
||||
assert dup is False
|
||||
|
||||
|
||||
def test_create_channel_roundtrip():
|
||||
encoded = encode_create_channel(b"\xde\xad")
|
||||
assert b"\xde\xad" in encoded
|
||||
|
||||
# Build response: channel_id=b"\xca\xfe", was_new=true
|
||||
resp = _encode_bytes_field(1, b"\xca\xfe") + _encode_varint_field(2, 1)
|
||||
ch_id, was_new = decode_create_channel_response(resp)
|
||||
assert ch_id == b"\xca\xfe"
|
||||
assert was_new is True
|
||||
|
||||
|
||||
def test_opaque_login_flow():
|
||||
# Encode login start
|
||||
start_req = encode_opaque_login_start("bob", b"\x01\x02")
|
||||
assert b"bob" in start_req
|
||||
|
||||
# Decode login start response
|
||||
start_resp = _encode_bytes_field(1, b"\xaa\xbb")
|
||||
assert decode_opaque_login_start_response(start_resp) == b"\xaa\xbb"
|
||||
|
||||
# Encode login finish
|
||||
finish_req = encode_opaque_login_finish("bob", b"\x03\x04", b"\x05\x06")
|
||||
assert b"bob" in finish_req
|
||||
|
||||
# Decode login finish response (session_token)
|
||||
finish_resp = _encode_bytes_field(1, b"\xde\xad\xbe\xef")
|
||||
assert decode_opaque_login_finish_response(finish_resp) == b"\xde\xad\xbe\xef"
|
||||
Reference in New Issue
Block a user