Private
Public Access
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""tests/test_packet.py"""
|
|
|
|
import pytest
|
|
|
|
from tuya_re import packet
|
|
from tuya_re.const import Code, SECURITY_FLAG_LOGIN
|
|
|
|
|
|
def test_varint_roundtrip():
|
|
for value in (0, 1, 0x7F, 0x80, 300, 0xFFFF, 2**28 - 1):
|
|
encoded = packet.pack_varint(value)
|
|
decoded, pos = packet.unpack_varint(encoded, 0)
|
|
assert decoded == value
|
|
assert pos == len(encoded)
|
|
|
|
|
|
def test_varint_single_byte_under_128():
|
|
assert packet.pack_varint(5) == b"\x05"
|
|
assert packet.pack_varint(300) == bytes([0xAC, 0x02]) # 300 = 0b10_0101100
|
|
|
|
|
|
def test_build_frame_layout():
|
|
frame = packet.build_frame(
|
|
seq_num=1, response_to=0, code=Code.SENDER_DEVICE_INFO, payload=b""
|
|
)
|
|
# header(12) + crc(2) = 14, zero-padded to 16
|
|
assert len(frame) == 16
|
|
assert frame[:12] == bytes([0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0])
|
|
|
|
|
|
def test_parse_frame_roundtrip():
|
|
frame = packet.build_frame(3, 2, Code.SENDER_DPS, b"\x01\x01\x01\x01")
|
|
seq, resp_to, code, payload = packet.parse_frame(frame)
|
|
assert (seq, resp_to, code, payload) == (
|
|
3,
|
|
2,
|
|
Code.SENDER_DPS.value,
|
|
b"\x01\x01\x01\x01",
|
|
)
|
|
|
|
|
|
def test_parse_frame_bad_crc_raises():
|
|
frame = bytearray(packet.build_frame(1, 0, Code.SENDER_DPS, b"\x05"))
|
|
frame[12] ^= 0xFF # corrupt payload byte
|
|
with pytest.raises(packet.FrameError):
|
|
packet.parse_frame(bytes(frame))
|
|
|
|
|
|
def test_encrypt_decrypt_frame_roundtrip():
|
|
key = bytes(range(16))
|
|
frame = packet.build_frame(1, 0, Code.SENDER_DEVICE_INFO, b"")
|
|
blob = packet.encrypt_frame(frame, key, SECURITY_FLAG_LOGIN, iv=bytes(16))
|
|
assert blob[0] == SECURITY_FLAG_LOGIN
|
|
assert blob[1:17] == bytes(16)
|
|
out = packet.decrypt_frame(blob, lambda flag: key)
|
|
assert out == frame
|
|
|
|
|
|
def test_split_chunks_fit_mtu_and_roundtrip():
|
|
blob = bytes(range(256)) * 2 # 512 bytes, forces many chunks
|
|
chunks = packet.split_packets(blob, protocol_version=3)
|
|
assert all(len(c) <= 20 for c in chunks)
|
|
# chunk 0 carries total length and version byte
|
|
_, pos = packet.unpack_varint(chunks[0], 0)
|
|
total, pos = packet.unpack_varint(chunks[0], pos)
|
|
assert total == len(blob)
|
|
assert chunks[0][pos] == 3 << 4
|
|
|
|
ras = packet.Reassembler()
|
|
results = [ras.feed(c) for c in chunks]
|
|
assert results[:-1] == [None] * (len(chunks) - 1)
|
|
assert results[-1] == blob
|
|
|
|
|
|
def test_reassembler_resets_on_packet_zero():
|
|
blob = bytes(range(40)) # 40 bytes -> 3 chunks (17 + 19 + 4 payload)
|
|
chunks = packet.split_packets(blob)
|
|
assert len(chunks) == 3
|
|
ras = packet.Reassembler()
|
|
ras.feed(chunks[0]) # start once
|
|
assert ras.feed(chunks[0]) is None # restart mid-stream is tolerated
|
|
assert ras.feed(chunks[1]) is None
|
|
assert ras.feed(chunks[2]) == blob
|