test_hub.py (11803B)
1 """Tests for inlite_ble hub module — ZoneState and notification safety.""" 2 3 import asyncio 4 5 from inlite_ble.hub import BLE_PACKET_PART_SIZE, STREAM_DATA_SIZE, InliteHub, ZoneState 6 from inlite_ble.protocol import ( 7 CHAR_CONTINUATION_UUID, 8 CHAR_WRITE_UUID, 9 OPCODE_IDENTIFY, 10 PKT_BLOCK_DATA_BLK, 11 PKT_BLOCK_FLUSH, 12 PKT_BLOCK_STREAM, 13 ) 14 15 16 class TestZoneState: 17 """Tests for ZoneState.""" 18 19 def test_is_on_when_mode_bit_set(self) -> None: 20 zs = ZoneState(output_id=0, output_mode=0x01, output_state=0x01) 21 assert zs.is_on is True 22 23 def test_is_off_when_mode_bit_clear(self) -> None: 24 zs = ZoneState(output_id=0, output_mode=0x00, output_state=0x00) 25 assert zs.is_on is False 26 27 def test_is_on_for_timer_even_without_permanent_mode(self) -> None: 28 # Confirmed by inlite-timer-on-off.btsnoop: mode=0x02, state=0x08. 29 zs = ZoneState(output_id=0, output_mode=0x02, output_state=0x08) 30 assert zs.is_on is True 31 32 def test_is_off_when_no_output_state_source_is_active(self) -> None: 33 zs = ZoneState(output_id=0, output_mode=0x03, output_state=0x00) 34 assert zs.is_on is False 35 36 def test_repr(self) -> None: 37 zs = ZoneState(output_id=1, output_mode=0x01, output_state=0x01) 38 r = repr(zs) 39 assert "id=1" in r 40 assert "ON" in r 41 42 43 class TestInliteHub: 44 """Tests for InliteHub initialization and properties.""" 45 46 def test_passphrase_required(self) -> None: 47 hub = InliteHub(device_id=0x1234, passphrase="test_pass") 48 assert hub.device_id == 0x1234 49 50 def test_not_connected_initially(self) -> None: 51 hub = InliteHub(device_id=1, passphrase="test") 52 assert hub.is_connected is False 53 54 def test_zone_states_empty_initially(self) -> None: 55 hub = InliteHub(device_id=1, passphrase="test") 56 assert hub.zone_states == {} 57 58 def test_smart_hub_75_hides_third_raw_zone(self) -> None: 59 async def run() -> InliteHub: 60 hub = InliteHub(device_id=1, passphrase="test") 61 62 async def get_info(_opcode: int, _data: bytes) -> bytes: 63 return bytes.fromhex( 64 "00000205000c1d380003" 65 "00010102000000" 66 "01010102000000" 67 "02010102000000" 68 ) 69 70 hub._send_acknowledged_command = get_info # type: ignore[method-assign] 71 await hub.query_zone_states() 72 return hub 73 74 hub = asyncio.run(run()) 75 assert set(hub.zone_states) == {0, 1} 76 assert hub.is_output_visible(2) is False 77 78 def test_loop_stored_on_connect(self) -> None: 79 """Verify _loop is set during connect (needed for thread-safe callbacks).""" 80 hub = InliteHub(device_id=1, passphrase="test") 81 assert hub._loop is None 82 83 def test_notification_uses_call_soon_threadsafe(self) -> None: 84 """Verify the notification handler references call_soon_threadsafe.""" 85 import inspect 86 source = inspect.getsource(InliteHub._on_notification) 87 assert "call_soon_threadsafe" in source 88 89 def test_stream_segments_are_reassembled_by_offset(self) -> None: 90 hub = InliteHub(device_id=1, passphrase="test") 91 assert hub._accept_stream_segment(b"\x00\x00abc") 92 assert hub._accept_stream_segment(b"\x03\x00de") 93 assert hub._last_stream_data == b"abcde" 94 95 def test_stream_segment_with_wrong_offset_is_rejected(self) -> None: 96 hub = InliteHub(device_id=1, passphrase="test") 97 assert hub._accept_stream_segment(b"\x00\x00abc") 98 assert not hub._accept_stream_segment(b"\x04\x00de") 99 assert hub._stream_invalid is True 100 101 def test_response_flush_is_queued_as_control_packet(self) -> None: 102 async def run() -> bytes: 103 hub = InliteHub(device_id=0x1234, passphrase="test") 104 hub._loop = asyncio.get_running_loop() 105 hub._crypto.decrypt_packet = lambda _packet: { # type: ignore[method-assign] 106 "pkt_type": PKT_BLOCK_FLUSH, 107 "src_id": hub.device_id, 108 "dest_id": hub._crypto.controller_address, 109 "data": b"\x00\x00", 110 } 111 hub.handle_notification(CHAR_WRITE_UUID, bytearray(b"packet")) 112 await asyncio.sleep(0) 113 return await hub._wait_ack() 114 115 assert asyncio.run(run()) == b"\x00\x00" 116 117 def test_identify_is_fire_and_forget(self) -> None: 118 async def run() -> list[tuple[int, int, bytes]]: 119 hub = InliteHub(device_id=0x1234, passphrase="test") 120 writes: list[tuple[int, int, bytes]] = [] 121 122 async def write_mesh(dest: int, packet_type: int, data: bytes) -> None: 123 writes.append((dest, packet_type, data)) 124 125 hub._write_mesh = write_mesh # type: ignore[method-assign] 126 assert await hub.identify() 127 return writes 128 129 assert asyncio.run(run()) == [(0x1234, PKT_BLOCK_DATA_BLK, b"\x01\x14\x00")] 130 131 def test_response_stream_uses_literal_completion_marker(self) -> None: 132 """The final ACK is count LE followed by ef, as captured from the app.""" 133 async def run() -> list[tuple[int, bytes]]: 134 hub = InliteHub(device_id=1, passphrase="test") 135 writes: list[tuple[int, bytes]] = [] 136 137 async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: 138 writes.append((packet_type, data)) 139 140 hub._write_mesh = write_mesh # type: ignore[method-assign] 141 hub._ack_queue.put_nowait(b"\x00\x00") 142 143 parts = iter( 144 [ 145 (PKT_BLOCK_STREAM, b"\x00\x00abc"), 146 (PKT_BLOCK_FLUSH, b"\x03\x00"), 147 ] 148 ) 149 150 async def wait_response_part() -> tuple[int, bytes]: 151 return next(parts) 152 153 hub._wait_response_part = wait_response_part # type: ignore[method-assign] 154 155 assert await hub._receive_response_stream() == b"abc" 156 return writes 157 158 writes = asyncio.run(run()) 159 assert writes == [(0x72, b"\x00\x00"), (0x72, b"\x03\x00"), (0x72, b"\x03\x00\xef")] 160 161 def test_response_stream_acknowledges_each_segment(self) -> None: 162 async def run() -> list[tuple[int, bytes]]: 163 hub = InliteHub(device_id=1, passphrase="test") 164 writes: list[tuple[int, bytes]] = [] 165 166 async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: 167 writes.append((packet_type, data)) 168 169 hub._write_mesh = write_mesh # type: ignore[method-assign] 170 hub._ack_queue.put_nowait(b"\x00\x00") 171 parts = iter( 172 [ 173 (PKT_BLOCK_STREAM, b"\x00\x00abc"), 174 (PKT_BLOCK_STREAM, b"\x03\x00de"), 175 (PKT_BLOCK_FLUSH, b"\x05\x00"), 176 ] 177 ) 178 179 async def wait_response_part() -> tuple[int, bytes]: 180 return next(parts) 181 182 hub._wait_response_part = wait_response_part # type: ignore[method-assign] 183 assert await hub._receive_response_stream() == b"abcde" 184 return writes 185 186 writes = asyncio.run(run()) 187 assert writes == [ 188 (0x72, b"\x00\x00"), 189 (0x72, b"\x03\x00"), 190 (0x72, b"\x05\x00"), 191 (0x72, b"\x05\x00\xef"), 192 ] 193 194 def test_response_stream_reacks_a_retransmitted_segment(self) -> None: 195 async def run() -> list[tuple[int, bytes]]: 196 hub = InliteHub(device_id=1, passphrase="test") 197 writes: list[tuple[int, bytes]] = [] 198 199 async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: 200 writes.append((packet_type, data)) 201 202 hub._write_mesh = write_mesh # type: ignore[method-assign] 203 hub._ack_queue.put_nowait(b"\x00\x00") 204 parts = iter( 205 [ 206 (PKT_BLOCK_STREAM, b"\x00\x00abc"), 207 (PKT_BLOCK_STREAM, b"\x00\x00abc"), 208 (PKT_BLOCK_FLUSH, b"\x03\x00"), 209 ] 210 ) 211 212 async def wait_response_part() -> tuple[int, bytes]: 213 return next(parts) 214 215 hub._wait_response_part = wait_response_part # type: ignore[method-assign] 216 assert await hub._receive_response_stream() == b"abc" 217 return writes 218 219 assert asyncio.run(run()) == [ 220 (0x72, b"\x00\x00"), 221 (0x72, b"\x03\x00"), 222 (0x72, b"\x03\x00"), 223 (0x72, b"\x03\x00\xef"), 224 ] 225 226 def test_complete_notification_reassembles_continuation(self) -> None: 227 received: list[dict] = [] 228 hub = InliteHub(device_id=1, passphrase="test") 229 loop = asyncio.new_event_loop() 230 try: 231 hub._loop = loop 232 hub._notification_callback = received.append 233 packet = hub._crypto.encrypt_packet(0, 0x73, b"x" * 100) 234 hub.handle_notification(CHAR_CONTINUATION_UUID, bytearray(packet[:78])) 235 assert received == [] 236 hub.handle_notification(CHAR_WRITE_UUID, bytearray(packet[78:])) 237 loop.call_soon(loop.stop) 238 loop.run_forever() 239 finally: 240 loop.close() 241 assert received[0]["data"] == b"x" * 100 242 243 def test_write_splits_large_ble_packet(self) -> None: 244 class Client: 245 is_connected = True 246 247 def __init__(self) -> None: 248 self.writes: list[tuple[str, bytes, bool]] = [] 249 250 async def write_gatt_char(self, char: str, data: bytes, response: bool) -> None: 251 self.writes.append((char, data, response)) 252 253 async def run() -> Client: 254 hub = InliteHub(device_id=1, passphrase="test") 255 client = Client() 256 hub._client = client # type: ignore[assignment] 257 await hub._write(b"x" * (BLE_PACKET_PART_SIZE + 1)) 258 return client 259 260 client = asyncio.run(run()) 261 assert client.writes == [ 262 (CHAR_CONTINUATION_UUID, b"x" * BLE_PACKET_PART_SIZE, True), 263 (CHAR_WRITE_UUID, b"x", True), 264 ] 265 266 def test_request_stream_is_segmented_and_offset_acknowledged(self) -> None: 267 async def run() -> list[tuple[int, bytes]]: 268 hub = InliteHub(device_id=1, passphrase="test") 269 writes: list[tuple[int, bytes]] = [] 270 271 async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: 272 writes.append((packet_type, data)) 273 274 acks = iter((b"\x00\x00", b"\x3e\x00", b"\x3f\x00", b"\x3f\x00\xef")) 275 276 async def wait_ack(_timeout: float = 0) -> bytes: 277 return next(acks) 278 279 hub._write_mesh = write_mesh # type: ignore[method-assign] 280 hub._wait_ack = wait_ack # type: ignore[method-assign] 281 assert await hub._send_raw_stream(b"\x00\x00" + b"x" * 63) 282 return writes 283 284 writes = asyncio.run(run()) 285 assert writes == [ 286 (0x70, b"\x00\x00"), 287 (0x71, b"\x00\x00" + b"x" * STREAM_DATA_SIZE), 288 (0x71, b"\x3e\x00x"), 289 (0x70, b"\x3f\x00"), 290 ] 291 292 def test_teach_in_tlv_is_forwarded(self) -> None: 293 received: list[tuple[int, int]] = [] 294 hub = InliteHub( 295 device_id=1, 296 passphrase="test", 297 on_teach_in_progress=lambda device_id, step: received.append((device_id, step)), 298 ) 299 loop = asyncio.new_event_loop() 300 try: 301 hub._loop = loop 302 hub._parse_tlv_notification(0x1234, b"\x54\x04\x01\xfa") 303 loop.call_soon(loop.stop) 304 loop.run_forever() 305 finally: 306 loop.close() 307 assert received == [(0x1234, 250)]