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