lda
Verified
038e152bbc
Checks / tests (ubuntu-latest, 3.11) (push) Canceled after 0s
Checks / tests (ubuntu-latest, 3.14) (push) Canceled after 0s
Checks / tests (windows-latest, 3.11) (push) Canceled after 0s
Checks / tests (windows-latest, 3.14) (push) Canceled after 0s
Checks / package (push) Canceled after 0s
83 lines
3.5 KiB
Python
83 lines
3.5 KiB
Python
"""Packet fixtures and stream boundaries independent of physical Bluetooth."""
|
|
|
|
import unittest
|
|
|
|
from controller.protocol import Decoder, Encoder
|
|
|
|
|
|
class ProtocolTests(unittest.TestCase):
|
|
def test_captured_mx_t40_connection_reply(self):
|
|
raw = bytes.fromhex("00 bb 01 1d 00 0b 02 10 01 34 30 04 01 02 04 05 01 b0")
|
|
frames = Decoder().feed(raw)
|
|
self.assertEqual(len(frames), 1)
|
|
self.assertEqual(frames[0].sequence, 0x1D)
|
|
self.assertEqual(frames[0].payload, bytes.fromhex("02 10 01 34 30 04 01 02 04 05 01"))
|
|
|
|
def test_captured_ble_source_and_eq_replies(self):
|
|
# These two consecutive hardware replies demonstrate both checksum
|
|
# variants: source excludes type, EQ includes it.
|
|
decoder = Decoder()
|
|
frames = decoder.feed(
|
|
bytes.fromhex("00 bb 01 6d 00 02 31 01 a1 00 bb 01 6e 00 04 41 03 02 00 b9")
|
|
)
|
|
self.assertEqual([frame.payload for frame in frames], [b"\x31\x01", b"\x41\x03\x02\x00"])
|
|
|
|
def test_captured_lighting_checksum_quirk(self):
|
|
for raw, mode in [
|
|
("00 bb 01 91 00 03 51 03 03 e9", 3),
|
|
("00 bb 01 ab 00 03 51 03 05 03", 5),
|
|
("00 bb 01 b0 00 03 51 03 06 08", 6),
|
|
("00 bb 01 b1 00 03 51 03 01 09", 1),
|
|
]:
|
|
with self.subTest(raw=raw):
|
|
frame = Decoder().feed(bytes.fromhex(raw))[0]
|
|
self.assertEqual(frame.payload, bytes((0x51, 3, mode)))
|
|
self.assertTrue(frame.checksum_valid)
|
|
|
|
def test_unknown_checksum_is_flagged_in_compatibility_mode(self):
|
|
raw = bytes.fromhex("00 bb 01 91 00 03 51 03 03 20")
|
|
self.assertEqual(Decoder().feed(raw), [])
|
|
frame = Decoder(strict_checksums=False).feed(raw)[0]
|
|
self.assertFalse(frame.checksum_valid)
|
|
|
|
def test_captured_play_pause_event_checksums(self):
|
|
for raw in ("00 bb 01 cd 00 02 70 11 40", "00 bb 01 ce 00 02 70 11 41"):
|
|
with self.subTest(raw=raw):
|
|
frame = Decoder().feed(bytes.fromhex(raw))[0]
|
|
self.assertEqual(frame.payload, b"\x70\x11")
|
|
self.assertTrue(frame.checksum_valid)
|
|
|
|
def test_app_connection_fixture(self):
|
|
self.assertEqual(Encoder().encode(b"\x01"), bytes.fromhex("00 bb 01 01 00 01 01 04"))
|
|
|
|
def test_signed_java_sequence_rollover(self):
|
|
encoder = Encoder()
|
|
for _ in range(127):
|
|
encoder.encode(b"\x01")
|
|
self.assertEqual(encoder.sequence, 127)
|
|
self.assertEqual(encoder.encode(b"\x01")[3], 1)
|
|
|
|
def test_every_split_and_multiple_frames(self):
|
|
# Peer replies use a different checksum rule than Encoder's commands.
|
|
first = bytes.fromhex("00 bb 01 01 00 04 41 03 00 02 4b")
|
|
second = bytes.fromhex("00 bb 01 02 00 03 51 03 01 5a")
|
|
for split in range(len(first) + 1):
|
|
decoder = Decoder()
|
|
frames = decoder.feed(first[:split]) + decoder.feed(first[split:] + second)
|
|
self.assertEqual([f.payload for f in frames], [b"\x41\x03\x00\x02", b"\x51\x03\x01"])
|
|
|
|
def test_corruption_and_noise_recovery(self):
|
|
bad = bytearray.fromhex("00 bb 01 01 00 01 01 03")
|
|
bad[-1] ^= 1
|
|
good = bytes.fromhex("00 bb 01 02 00 01 04 07")
|
|
frames = Decoder().feed(b"noise" + bad + good)
|
|
self.assertEqual([f.payload for f in frames], [b"\x04"])
|
|
|
|
def test_empty_ack(self):
|
|
frame = Decoder().feed(bytes.fromhex("00 bb 00 02 00 00 02"))[0]
|
|
self.assertEqual((frame.kind, frame.sequence, frame.payload), (0, 2, b""))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|