test_transport.py (11924B)
1 """Tests for the read-only Bleak transport boundary.""" 2 3 from collections.abc import Callable 4 from dataclasses import dataclass 5 from typing import cast 6 import unittest 7 8 from bleak import BleakClient 9 from bleak.backends.characteristic import BleakGATTCharacteristic 10 11 from pyhunterbtt.const import ( 12 AUTHENTICATION_UUID, 13 BATTERY_LEVEL_UUID, 14 CLIENT_IDENTIFIER_UUID, 15 CLOCK_UUID, 16 CONTROLLER_COMMAND_UUID, 17 CONTROLLER_STATE_UUID, 18 FIRMWARE_REVISION_UUID, 19 RUNTIME_UUID, 20 SCHEDULE_STATE_UUID, 21 ZONE1_CONFIG_UUID, 22 ) 23 from pyhunterbtt.device import HunterBTTState 24 from pyhunterbtt.exceptions import ( 25 AuthenticationIdentifierError, 26 AuthenticationVerificationError, 27 AuthenticationWriteError, 28 ConfigurationReadError, 29 ManualWateringError, 30 NotAuthenticatedError, 31 StatusReadError, 32 ) 33 from pyhunterbtt.session import HunterBTTSession 34 from pyhunterbtt.transport import HunterBTTTransport 35 36 IDLE_STATE = bytes.fromhex("01 00 01 01 00 00 01 01 00 00 01 01 80 00") 37 RUNNING_STATE = bytes.fromhex("01 00 01 01 01 00 01 01 00 00 05 01 80 00") 38 RUNTIME = bytes.fromhex("08 00 00 00 00 00 00 00 00 00 00 00 09 00 00 00") 39 CONTROLLER_COMMAND = bytes(12) 40 CLOCK = bytes.fromhex("07 ea 01 02 03 04 05 00") 41 ZONE_CONFIG = bytes(17) 42 TIMER_SCHEDULE = bytes(15) 43 CYCLING_SCHEDULE = bytes(18) 44 ADDITIONAL_STARTS = bytes(12) 45 SEASONAL_ADJUSTMENT = bytes((0,)) + bytes((100,)) * 12 46 SCHEDULE_DATES = bytes.fromhex("07 ea 01 02") * 4 47 48 49 @dataclass(frozen=True, slots=True) 50 class FakeCharacteristic: 51 """A minimal notification sender with a Bleak-compatible UUID attribute.""" 52 53 uuid: str 54 55 56 class FakeBleakClient: 57 """Minimal async BLE client used without accessing a Bluetooth adapter.""" 58 59 def __init__(self, values: dict[str, bytes]) -> None: 60 self.values = values 61 self.reads: list[str] = [] 62 self.started: list[str] = [] 63 self.stopped: list[str] = [] 64 self.writes: list[tuple[str, bytes, bool]] = [] 65 self.read_errors: dict[str, Exception] = {} 66 self.write_errors: dict[str, Exception] = {} 67 self.callbacks: dict[ 68 str, Callable[[BleakGATTCharacteristic, bytearray], None] 69 ] = {} 70 71 async def read_gatt_char(self, uuid: str) -> bytearray: 72 """Return a configured characteristic value.""" 73 self.reads.append(uuid) 74 if error := self.read_errors.get(uuid): 75 raise error 76 return bytearray(self.values[uuid]) 77 78 async def start_notify( 79 self, 80 uuid: str, 81 callback: Callable[[BleakGATTCharacteristic, bytearray], None], 82 ) -> None: 83 """Register a notification callback.""" 84 self.started.append(uuid) 85 self.callbacks[uuid] = callback 86 87 async def stop_notify(self, uuid: str) -> None: 88 """Record notification shutdown.""" 89 self.stopped.append(uuid) 90 self.callbacks.pop(uuid, None) 91 92 async def write_gatt_char( 93 self, 94 uuid: str, 95 payload: bytes, 96 *, 97 response: bool, 98 ) -> None: 99 """Record an outgoing write without contacting BLE hardware.""" 100 self.writes.append((uuid, payload, response)) 101 if error := self.write_errors.get(uuid): 102 raise error 103 104 def notify(self, uuid: str, payload: bytes) -> None: 105 """Deliver a synthetic notification.""" 106 callback = self.callbacks[uuid] 107 callback( 108 cast(BleakGATTCharacteristic, FakeCharacteristic(uuid)), 109 bytearray(payload), 110 ) 111 112 113 class HunterBTTTransportTests(unittest.IsolatedAsyncioTestCase): 114 """Verify that the transport does not require live BLE access.""" 115 116 def setUp(self) -> None: 117 self.client = FakeBleakClient( 118 { 119 FIRMWARE_REVISION_UUID: b"Version 5.0.1.5 1C", 120 BATTERY_LEVEL_UUID: b"\x64", 121 CONTROLLER_STATE_UUID: IDLE_STATE, 122 SCHEDULE_STATE_UUID: bytes(17), 123 } 124 ) 125 self.transport = HunterBTTTransport(cast(BleakClient, self.client)) 126 127 async def test_read_snapshot_avoids_authentication_material(self) -> None: 128 state = await self.transport.async_read_snapshot() 129 130 self.assertEqual( 131 self.client.reads, 132 [ 133 FIRMWARE_REVISION_UUID, 134 BATTERY_LEVEL_UUID, 135 CONTROLLER_STATE_UUID, 136 SCHEDULE_STATE_UUID, 137 ], 138 ) 139 self.assertNotIn(AUTHENTICATION_UUID, self.client.reads) 140 self.assertNotIn(CLIENT_IDENTIFIER_UUID, self.client.reads) 141 self.assertEqual(state.channel_count, 1) 142 self.assertEqual(state.battery_level, 100) 143 self.assertTrue(state.controller is not None) 144 self.assertTrue(state.schedule_state is not None) 145 146 async def test_notifications_update_the_shared_state(self) -> None: 147 updates: list[HunterBTTState] = [] 148 await self.transport.async_start_notifications(updates.append) 149 150 self.client.notify(CONTROLLER_STATE_UUID, IDLE_STATE) 151 self.client.notify(RUNTIME_UUID, RUNTIME) 152 153 self.assertEqual(self.client.started, [CONTROLLER_STATE_UUID, RUNTIME_UUID]) 154 self.assertEqual(len(updates), 2) 155 self.assertTrue(self.transport.state.controller is not None) 156 self.assertTrue(self.transport.state.runtime is not None) 157 self.assertEqual(self.transport.state.runtime.manual_seconds, 9) 158 159 await self.transport.async_stop_notifications() 160 self.assertEqual(self.client.stopped, [RUNTIME_UUID, CONTROLLER_STATE_UUID]) 161 162 async def test_notifications_cannot_be_started_twice(self) -> None: 163 await self.transport.async_start_notifications(lambda state: None) 164 165 with self.assertRaisesRegex(RuntimeError, "already active"): 166 await self.transport.async_start_notifications(lambda state: None) 167 168 169 class HunterBTTSessionTests(unittest.IsolatedAsyncioTestCase): 170 """Verify authentication and configuration reads using synthetic values.""" 171 172 def setUp(self) -> None: 173 self.client = FakeBleakClient( 174 { 175 CLIENT_IDENTIFIER_UUID: bytes.fromhex("01 02 03 04 05 06"), 176 FIRMWARE_REVISION_UUID: b"Version 5.0.1.4 1C", 177 BATTERY_LEVEL_UUID: b"\x64", 178 CONTROLLER_STATE_UUID: IDLE_STATE, 179 SCHEDULE_STATE_UUID: bytes(17), 180 "0000ff83-0000-1000-8000-00805f9b34fb": CONTROLLER_COMMAND, 181 "0000ff84-0000-1000-8000-00805f9b34fb": CLOCK, 182 "0000ff86-0000-1000-8000-00805f9b34fb": ZONE_CONFIG, 183 "0000ff87-0000-1000-8000-00805f9b34fb": TIMER_SCHEDULE, 184 "0000ff88-0000-1000-8000-00805f9b34fb": CYCLING_SCHEDULE, 185 "0000ffa0-0000-1000-8000-00805f9b34fb": ADDITIONAL_STARTS, 186 "0000ff9c-0000-1000-8000-00805f9b34fb": SEASONAL_ADJUSTMENT, 187 "0000ffa4-0000-1000-8000-00805f9b34fb": SCHEDULE_DATES, 188 } 189 ) 190 self.session = HunterBTTSession(cast(BleakClient, self.client)) 191 192 async def test_configuration_requires_authenticated_session(self) -> None: 193 with self.assertRaises(NotAuthenticatedError): 194 await self.session.async_read_configuration() 195 196 async def test_authentication_and_configuration_are_whitelisted(self) -> None: 197 await self.session.async_authenticate("2468") 198 configuration = await self.session.async_read_configuration() 199 200 self.assertTrue(self.session.is_authenticated) 201 self.assertEqual(len(self.client.writes), 1) 202 self.assertEqual(self.client.writes[0][0], AUTHENTICATION_UUID) 203 self.assertEqual(len(self.client.writes[0][1]), 14) 204 self.assertTrue(self.client.writes[0][2]) 205 self.assertEqual(configuration.clock.date_time.year, 2026) 206 self.assertEqual(configuration.zone1.timer.run_seconds, 0) 207 self.assertEqual(configuration.zone1.additional_start_seconds, (0, 0, 0, 0)) 208 self.assertEqual( 209 configuration.seasonal_adjustment.monthly_percentages, (100,) * 12 210 ) 211 self.assertIsNone(configuration.zone2) 212 213 async def test_active_status_reads_runtime_once(self) -> None: 214 self.client.values[CONTROLLER_STATE_UUID] = RUNNING_STATE 215 self.client.values[RUNTIME_UUID] = RUNTIME 216 217 state = await self.session.async_authenticate("2468") 218 219 self.assertIn(RUNTIME_UUID, self.client.reads) 220 self.assertIsNotNone(state.runtime) 221 self.assertEqual(state.runtime.manual_seconds, 9) 222 223 async def test_identifier_read_failure_is_typed_and_not_authenticated(self) -> None: 224 self.client.read_errors[CLIENT_IDENTIFIER_UUID] = OSError("synthetic") 225 226 with self.assertRaises(AuthenticationIdentifierError) as context: 227 await self.session.async_authenticate("2468") 228 229 self.assertEqual( 230 str(context.exception), 231 "could not read the transient authentication identifier: synthetic", 232 ) 233 self.assertIsInstance(context.exception.__cause__, OSError) 234 self.assertFalse(self.session.is_authenticated) 235 self.assertEqual(self.client.writes, []) 236 237 async def test_authentication_write_failure_is_typed_and_not_authenticated( 238 self, 239 ) -> None: 240 self.client.write_errors[AUTHENTICATION_UUID] = OSError("synthetic") 241 242 with self.assertRaises(AuthenticationWriteError): 243 await self.session.async_authenticate("2468") 244 245 self.assertFalse(self.session.is_authenticated) 246 247 async def test_verification_failure_clears_authentication(self) -> None: 248 self.client.read_errors[FIRMWARE_REVISION_UUID] = OSError("synthetic") 249 250 with self.assertRaises(AuthenticationVerificationError) as context: 251 await self.session.async_authenticate("2468") 252 253 self.assertEqual( 254 str(context.exception), 255 "could not verify session authentication: " 256 "could not read controller status: synthetic", 257 ) 258 self.assertIsInstance(context.exception.__cause__, StatusReadError) 259 self.assertFalse(self.session.is_authenticated) 260 261 async def test_status_read_failure_is_typed(self) -> None: 262 await self.session.async_authenticate("2468") 263 self.client.read_errors[CONTROLLER_STATE_UUID] = OSError("synthetic") 264 265 with self.assertRaises(StatusReadError): 266 await self.session.async_read_status() 267 268 async def test_configuration_read_failure_is_typed(self) -> None: 269 await self.session.async_authenticate("2468") 270 self.client.read_errors[CLOCK_UUID] = OSError("synthetic") 271 272 with self.assertRaises(ConfigurationReadError): 273 await self.session.async_read_configuration() 274 275 async def test_manual_start_writes_duration_before_command(self) -> None: 276 await self.session.async_authenticate("2468") 277 278 await self.session.async_start_manual_zone1(10) 279 280 writes = self.client.writes[1:] 281 self.assertEqual( 282 [write[0] for write in writes], [ZONE1_CONFIG_UUID, CONTROLLER_COMMAND_UUID] 283 ) 284 self.assertEqual(writes[0][1][11:14], bytes((0, 0, 10))) 285 self.assertTrue(all(write[2] for write in writes)) 286 287 async def test_manual_stop_writes_observed_command(self) -> None: 288 await self.session.async_authenticate("2468") 289 290 await self.session.async_stop_manual_zone1() 291 292 self.assertEqual(self.client.writes[-1][0], CONTROLLER_COMMAND_UUID) 293 self.assertEqual(self.client.writes[-1][1][2], 2) 294 self.assertFalse(self.client.writes[-1][1][4]) 295 296 async def test_manual_start_write_failure_is_typed(self) -> None: 297 await self.session.async_authenticate("2468") 298 self.client.write_errors[ZONE1_CONFIG_UUID] = OSError("synthetic") 299 300 with self.assertRaises(ManualWateringError): 301 await self.session.async_start_manual_zone1(10) 302 303 304 if __name__ == "__main__": 305 unittest.main()