diff --git a/pyhap/hds_protocol.py b/pyhap/hds_protocol.py index 31a859c3..f0e42658 100644 --- a/pyhap/hds_protocol.py +++ b/pyhap/hds_protocol.py @@ -76,73 +76,97 @@ class Int64(int): _DICT_TERMINATED = 0xEF -def _encode_int(value: int) -> bytes: +def _encode_int(value: int, out: bytearray) -> None: if isinstance(value, Int64): - return bytes([_INT64]) + struct.pack(" bytes: + value: bytes, + out: bytearray, + base: int, + len8: int, + len16: int, + len32: int, + len64: int, +) -> None: length = len(value) if length <= 32: - return bytes([base + length]) + value - if length <= 0xFF: - return bytes([len8]) + struct.pack(" bytes: - """Serialize a python value into the HDS payload format.""" + out.append(base + length) + elif length <= 0xFF: + out.append(len8) + out.append(length) + elif length <= 0xFFFF: + out += struct.pack(" None: + """Append the HDS serialization of a python value to ``out``.""" if value is True: - return bytes([_TRUE]) - if value is False: - return bytes([_FALSE]) - if value is None: - return bytes([_NULL]) - if isinstance(value, int): - return _encode_int(value) - if isinstance(value, float): - return bytes([_FLOAT64]) + struct.pack(" 14 + out.append(_ARRAY_TERMINATED if terminated else _ARRAY_0 + len(value)) + for item in value: + _encode_into(item, out) + if terminated: + out.append(_TERMINATOR) + elif isinstance(value, dict): + terminated = len(value) > 14 + out.append(_DICT_TERMINATED if terminated else _DICT_0 + len(value)) + for key, item in value.items(): + _encode_into(key, out) + _encode_into(item, out) + if terminated: + out.append(_TERMINATOR) + else: + raise TypeError(f"Cannot encode {type(value).__name__} into an HDS payload") + + +def encode(value: Any) -> bytes: + """Serialize a python value into the HDS payload format.""" + out = bytearray() + _encode_into(value, out) + return bytes(out) # Real HDS payloads nest only a handful of levels; bound recursion well above diff --git a/tests/test_hds_protocol.py b/tests/test_hds_protocol.py new file mode 100644 index 00000000..513e293a --- /dev/null +++ b/tests/test_hds_protocol.py @@ -0,0 +1,64 @@ +"""Tests for the HDS payload encoder and decoder.""" + +from typing import Any + +import pytest + +from pyhap.hds_protocol import Int64, decode, encode + +VALUES = [ + True, + False, + None, + -1, + 0, + 38, + 39, + -128, + 127, + -32768, + 32767, + -(2**31), + 2**31 - 1, + 2**40, + Int64(5), + 1.5, + "", + "a" * 32, + "a" * 33, + "a" * 0x100, + b"x" * 32, + b"x" * 0xFF, + b"x" * 0x10000, + bytearray(b"ab"), + [1, "two", b"3"], + list(range(15)), + {"a": 1, "b": [True, None]}, + {str(i): i for i in range(15)}, +] + + +@pytest.mark.parametrize("value", VALUES) +def test_round_trip(value: Any) -> None: + """Encoded values decode back to the same value.""" + result = decode(encode(value)) + if isinstance(value, bytearray): + value = bytes(value) + assert result == value + + +def test_encode_int_size() -> None: + """Each integer uses the smallest encoding, Int64 always uses 8 bytes.""" + assert len(encode(-1)) == 1 + assert len(encode(38)) == 1 + assert len(encode(-128)) == 2 + assert len(encode(32767)) == 3 + assert len(encode(2**31 - 1)) == 5 + assert len(encode(2**40)) == 9 + assert len(encode(Int64(5))) == 9 + + +def test_encode_unsupported_type() -> None: + """An unsupported type raises TypeError.""" + with pytest.raises(TypeError): + encode(object())