test_protocol.py (15764B)
1 """Tests built from BTT-101 firmware 5.0.1.4 observations.""" 2 3 import unittest 4 from datetime import date, datetime, timezone 5 6 from pyhunterbtt.const import CONTROLLER_COMMAND_UUID, ZONE1_CONFIG_UUID 7 from pyhunterbtt.device import HunterBTTState 8 from pyhunterbtt.models import ( 9 ControllerClock, 10 RuntimeActivity, 11 ZoneActivity, 12 ZoneMode, 13 ) 14 from pyhunterbtt.protocol import ( 15 ProtocolError, 16 channel_count_from_firmware_revision, 17 decode_battery_level, 18 decode_additional_start_times, 19 decode_clock, 20 decode_controller_command, 21 decode_controller_state, 22 decode_cycling_schedule, 23 decode_firmware_revision, 24 decode_latest_watering_log, 25 decode_runtime, 26 decode_schedule_dates, 27 decode_schedule_state, 28 decode_timer_schedule, 29 decode_watering_log_count, 30 decode_watering_log_records, 31 decode_zone_config, 32 encode_clock, 33 encode_controller_command, 34 encode_watering_log_query, 35 encode_zone_config, 36 plan_manual_start_zone1, 37 plan_manual_stop_zone1, 38 ) 39 40 IDLE_STATE = bytes.fromhex("01 00 01 01 00 00 01 01 00 00 01 01 80 00") 41 RUNNING_STATE = bytes.fromhex("01 00 01 01 01 00 01 01 00 00 05 01 80 00") 42 INITIAL_COMMAND = bytes.fromhex("01 00 01 01 00 01 01 00 00 1e 00 00") 43 START_COMMAND = bytes.fromhex("01 00 01 01 01 01 01 00 00 1e 00 00") 44 STOP_COMMAND = bytes.fromhex("01 00 02 01 00 02 01 00 00 1e 00 00") 45 ZONE1_INITIAL = bytes.fromhex("04 01 7f ff ff ff ff 00 ff ff ff 02 00 00 00 1e 00") 46 ZONE1_TEN_SECONDS = bytes.fromhex("04 01 7f ff ff ff ff 00 ff ff ff 00 00 0a 00 1e 00") 47 48 49 class ScheduleDateTests(unittest.TestCase): 50 """Exercise FFA4's observed unset-date sentinel.""" 51 52 def test_zero_tuple_is_an_unset_date(self) -> None: 53 dates = decode_schedule_dates(bytes(16)) 54 55 self.assertIsNone(dates.zone1_timer) 56 self.assertIsNone(dates.zone1_cycling) 57 self.assertIsNone(dates.zone2_timer) 58 self.assertIsNone(dates.zone2_cycling) 59 60 def test_populated_date_is_decoded(self) -> None: 61 dates = decode_schedule_dates(bytes.fromhex("07 ea 08 1e") + bytes(12)) 62 63 self.assertEqual(dates.zone1_timer, date(2026, 8, 30)) 64 65 66 class ScheduleTimeTests(unittest.TestCase): 67 """Exercise the observed FF:FF:FF unset-time sentinel.""" 68 69 def test_unset_timer_starts_are_none(self) -> None: 70 timer = decode_timer_schedule(bytes.fromhex("ff ff ff") * 5) 71 72 self.assertEqual(timer.start_seconds, (None, None, None, None)) 73 self.assertIsNone(timer.run_seconds) 74 75 def test_unset_cycling_windows_and_additional_starts_are_none(self) -> None: 76 cycling = decode_cycling_schedule(bytes.fromhex("ff ff ff") * 6) 77 78 self.assertIsNone(cycling.window1_start_seconds) 79 self.assertIsNone(cycling.window2_end_seconds) 80 self.assertIsNone(cycling.run_seconds) 81 self.assertIsNone(cycling.soak_seconds) 82 self.assertEqual( 83 decode_additional_start_times(bytes.fromhex("ff ff ff") * 4), (None,) * 4 84 ) 85 86 87 class ControllerStateTests(unittest.TestCase): 88 """Exercise state and runtime observations.""" 89 90 def test_idle_state(self) -> None: 91 state = decode_controller_state(IDLE_STATE) 92 self.assertTrue(state.enabled) 93 self.assertFalse(state.zone1.enable_manual) 94 self.assertEqual(state.zone1.state, 1) 95 self.assertEqual(state.zone1.conflict, 0x80) 96 self.assertFalse(state.zone1.timer_schedule_conflict) 97 self.assertFalse(state.zone1.cycling_schedule_conflict) 98 self.assertTrue(state.zone1.has_pending_conflict_details) 99 100 def test_running_state(self) -> None: 101 state = decode_controller_state(RUNNING_STATE) 102 self.assertTrue(state.zone1.enable_manual) 103 self.assertEqual(state.zone1.state, 5) 104 self.assertIs(state.zone1.activity, ZoneActivity.APPLICATION_MANUAL) 105 self.assertFalse(state.zone2.enable_manual) 106 107 def test_conflict_bit_properties(self) -> None: 108 payload = bytearray(IDLE_STATE) 109 payload[12] = 0x03 110 state = decode_controller_state(bytes(payload)) 111 self.assertTrue(state.zone1.timer_schedule_conflict) 112 self.assertTrue(state.zone1.cycling_schedule_conflict) 113 self.assertFalse(state.zone1.has_pending_conflict_details) 114 115 def test_runtime_counts_down(self) -> None: 116 runtime = decode_runtime( 117 bytes.fromhex("08 00 00 00 00 00 00 00 00 00 00 00 09 00 00 00") 118 ) 119 self.assertEqual(runtime.watering_type, 8) 120 self.assertIs(runtime.activity, RuntimeActivity.APPLICATION_MANUAL) 121 self.assertEqual(runtime.manual_seconds, 9) 122 123 def test_runtime_resets_after_completion(self) -> None: 124 state = decode_controller_state(IDLE_STATE) 125 runtime = decode_runtime( 126 bytes.fromhex("08 00 00 00 00 00 00 00 00 00 00 00 0a 00 00 00") 127 ) 128 self.assertFalse(state.zone1.enable_manual) 129 self.assertEqual(runtime.manual_seconds, 10) 130 131 def test_aggregate_state_hides_reset_duration_while_idle(self) -> None: 132 device = HunterBTTState() 133 self.assertIsNone(device.zone1_is_manual_watering) 134 135 device.update( 136 "0000ff8a-0000-1000-8000-00805f9b34fb", 137 bytes.fromhex("08 00 00 00 00 00 00 00 00 00 00 00 0a 00 00 00"), 138 ) 139 device.update("0000ff82-0000-1000-8000-00805f9b34fb", IDLE_STATE) 140 self.assertFalse(device.zone1_is_manual_watering) 141 self.assertFalse(device.zone1_is_watering) 142 self.assertIsNone(device.zone1_manual_remaining_seconds) 143 144 def test_aggregate_state_exposes_active_countdown(self) -> None: 145 device = HunterBTTState() 146 device.update("0000ff82-0000-1000-8000-00805f9b34fb", RUNNING_STATE) 147 device.update( 148 "0000ff8a-0000-1000-8000-00805f9b34fb", 149 bytes.fromhex("08 00 00 00 00 00 00 00 00 00 00 00 09 00 00 00"), 150 ) 151 self.assertTrue(device.zone1_is_manual_watering) 152 self.assertTrue(device.zone1_is_watering) 153 self.assertEqual(device.zone1_manual_remaining_seconds, 9) 154 155 def test_aggregate_state_decodes_channel_count(self) -> None: 156 device = HunterBTTState() 157 self.assertIsNone(device.channel_count) 158 self.assertTrue( 159 device.update( 160 "00002a26-0000-1000-8000-00805f9b34fb", 161 b"Version 5.0.1.5 1C", 162 ) 163 ) 164 self.assertEqual(device.firmware_revision, "Version 5.0.1.5 1C") 165 self.assertEqual(device.channel_count, 1) 166 167 def test_known_zone_activity_values(self) -> None: 168 device = HunterBTTState() 169 for value in (2, 5, 9, 17): 170 payload = bytearray(IDLE_STATE) 171 payload[10] = value 172 device.update( 173 "0000ff82-0000-1000-8000-00805f9b34fb", 174 bytes(payload), 175 ) 176 self.assertTrue(device.zone1_is_watering) 177 178 payload[10] = 3 179 device.update( 180 "0000ff82-0000-1000-8000-00805f9b34fb", 181 bytes(payload), 182 ) 183 self.assertIsNone(device.zone1_is_watering) 184 185 def test_known_zone_modes(self) -> None: 186 state = decode_controller_state(IDLE_STATE) 187 self.assertIs(state.zone1.mode_type, ZoneMode.TIMER) 188 189 payload = bytearray(IDLE_STATE) 190 payload[3] = 2 191 state = decode_controller_state(bytes(payload)) 192 self.assertIs(state.zone1.mode_type, ZoneMode.CYCLING) 193 194 payload[3] = 3 195 state = decode_controller_state(bytes(payload)) 196 self.assertIsNone(state.zone1.mode_type) 197 198 def test_known_runtime_activity_values(self) -> None: 199 for value in (1, 2, 4, 6, 8, 16, 32): 200 payload = bytes((value,)) + bytes(15) 201 runtime = decode_runtime(payload) 202 self.assertIs(runtime.activity, RuntimeActivity(value)) 203 204 runtime = decode_runtime(bytes((3,)) + bytes(15)) 205 self.assertIsNone(runtime.activity) 206 207 208 class CodecTests(unittest.TestCase): 209 """Verify lossless codecs for captured fields.""" 210 211 def test_command_round_trip(self) -> None: 212 command = decode_controller_command(INITIAL_COMMAND) 213 self.assertEqual(command.run_all_seconds, 30 * 60) 214 self.assertEqual(encode_controller_command(command), INITIAL_COMMAND) 215 216 def test_zone_config_round_trip(self) -> None: 217 config = decode_zone_config(ZONE1_INITIAL) 218 self.assertEqual(config.operation_index, 4) 219 self.assertEqual(config.manual_seconds, 2 * 60 * 60) 220 self.assertEqual(config.external_manual_seconds, 30 * 60) 221 self.assertEqual(encode_zone_config(config), ZONE1_INITIAL) 222 223 def test_clock_round_trip(self) -> None: 224 payload = bytes.fromhex("07 ea 08 1e 0a 1f 20 40") 225 clock = decode_clock(payload) 226 self.assertEqual(clock.date_time, datetime(2026, 8, 30, 10, 31, 32)) 227 self.assertEqual(clock.weekday_mask, 0x40) 228 self.assertEqual(encode_clock(clock), payload) 229 230 def test_clock_rejects_invalid_date(self) -> None: 231 with self.assertRaises(ProtocolError): 232 decode_clock(bytes.fromhex("07 ea 02 1f 0a 1f 20 40")) 233 234 def test_clock_rejects_large_weekday_mask(self) -> None: 235 clock = ControllerClock(datetime(2026, 8, 30), 0x100) 236 with self.assertRaises(ProtocolError): 237 encode_clock(clock) 238 239 def test_clock_rejects_timezone_aware_value(self) -> None: 240 clock = ControllerClock(datetime(2026, 8, 30, tzinfo=timezone.utc), 0) 241 242 with self.assertRaisesRegex(ProtocolError, "timezone-naive"): 243 encode_clock(clock) 244 245 def test_battery(self) -> None: 246 self.assertEqual(decode_battery_level(b"\x64"), 100) 247 with self.assertRaises(ProtocolError): 248 decode_battery_level(b"\x65") 249 250 def test_firmware_revision_and_channel_count(self) -> None: 251 revision = decode_firmware_revision(b"Version 5.0.1.5 2C\x00") 252 self.assertEqual(revision, "Version 5.0.1.5 2C") 253 self.assertEqual(channel_count_from_firmware_revision(revision), 2) 254 self.assertEqual(channel_count_from_firmware_revision("4.0 1c"), 1) 255 self.assertIsNone(channel_count_from_firmware_revision("Version 2.3")) 256 257 def test_firmware_revision_rejects_invalid_text(self) -> None: 258 with self.assertRaises(ProtocolError): 259 decode_firmware_revision(b"\xff") 260 with self.assertRaises(ProtocolError): 261 decode_firmware_revision(b"\x00") 262 263 def test_schedule_state(self) -> None: 264 payload = bytes.fromhex("01 82 50 d8 a5 80 cc 00 00 00 00 00 00 85 04 83 02") 265 schedule = decode_schedule_state(payload) 266 self.assertTrue(schedule.has_conflicts) 267 self.assertEqual( 268 schedule.zone1_timer_conflicts.out_of_order_pairs, 269 ((1, 2),), 270 ) 271 self.assertEqual( 272 schedule.zone1_timer_conflicts.overlapping_pairs, 273 ((1, 2), (2, 3), (3, 4)), 274 ) 275 self.assertEqual( 276 schedule.zone1_timer_conflicts.affected_start_times, 277 (1, 2, 4), 278 ) 279 cycling = schedule.zone1_cycling_conflicts 280 self.assertTrue(cycling.window1_end_before_start) 281 self.assertTrue(cycling.windows_overlap) 282 self.assertTrue(cycling.window1_too_short_for_run) 283 self.assertTrue(cycling.window2_too_short_for_run) 284 self.assertTrue(cycling.window2_too_short_for_soak) 285 self.assertEqual(schedule.zone1_timer_interval.day_counter, 5) 286 self.assertTrue(schedule.zone1_timer_interval.initial_day_starts_elapsed) 287 self.assertEqual(schedule.zone1_cycling_interval.day_counter, 4) 288 self.assertFalse(schedule.zone1_cycling_interval.initial_day_starts_elapsed) 289 290 def test_size_validation(self) -> None: 291 with self.assertRaisesRegex(ProtocolError, "14 bytes"): 292 decode_controller_state(b"\x00") 293 294 def test_latest_watering_log(self) -> None: 295 payload = bytes.fromhex("02 00 00 00 01 12 ab cd 34 56") 296 record = decode_latest_watering_log(payload) 297 self.assertEqual(record.zone, 2) 298 self.assertEqual(record.started_at, datetime(2000, 1, 1, 0, 0, 1)) 299 self.assertEqual(record.infiltrate_seconds, 0x1ABCD) 300 self.assertEqual(record.run_seconds, 0x23456) 301 302 def test_watering_log_page_ignores_empty_slot(self) -> None: 303 record = bytes.fromhex("02 00 00 00 01 12 ab cd 34 56") 304 records = decode_watering_log_records(record + bytes(10)) 305 self.assertEqual(len(records), 1) 306 self.assertEqual(records[0].zone, 2) 307 308 def test_watering_log_count(self) -> None: 309 payload = bytes.fromhex("00 01 02 03") 310 self.assertEqual(decode_watering_log_count(payload), 0x10203) 311 312 def test_watering_log_query_uses_zero_based_month_and_day(self) -> None: 313 start = datetime(2024, 1, 1, 2, 3, 4) 314 end = datetime(2024, 12, 31, 22, 23, 24) 315 self.assertEqual( 316 encode_watering_log_query(start, end), 317 bytes.fromhex("07 e8 00 00 02 03 04 07 e8 0b 1e 16 17 18"), 318 ) 319 320 def test_watering_log_query_rejects_timezone_aware_values(self) -> None: 321 start = datetime(2024, 1, 1, tzinfo=timezone.utc) 322 end = datetime(2024, 1, 2, tzinfo=timezone.utc) 323 324 with self.assertRaisesRegex(ProtocolError, "timezone-naive"): 325 encode_watering_log_query(start, end) 326 327 def test_watering_log_query_rejects_reversed_range(self) -> None: 328 with self.assertRaises(ProtocolError): 329 encode_watering_log_query( 330 datetime(2024, 1, 2), 331 datetime(2024, 1, 1), 332 ) 333 334 335 class ManualWateringPlanTests(unittest.TestCase): 336 """Verify the exact writes observed in the trace.""" 337 338 def test_start_ten_seconds(self) -> None: 339 command = decode_controller_command(INITIAL_COMMAND) 340 config = decode_zone_config(ZONE1_INITIAL) 341 plan = plan_manual_start_zone1(command, config, 10) 342 343 self.assertEqual(len(plan.writes), 2) 344 self.assertEqual(plan.writes[0].uuid, ZONE1_CONFIG_UUID) 345 self.assertEqual(plan.writes[0].payload, ZONE1_TEN_SECONDS) 346 self.assertEqual(plan.writes[1].uuid, CONTROLLER_COMMAND_UUID) 347 self.assertEqual(plan.writes[1].payload, START_COMMAND) 348 349 def test_start_preserves_other_state(self) -> None: 350 command = decode_controller_command(INITIAL_COMMAND) 351 config = decode_zone_config(ZONE1_INITIAL) 352 result = decode_controller_command( 353 plan_manual_start_zone1(command, config, 10).writes[1].payload 354 ) 355 self.assertEqual(result.zone1_mode, command.zone1_mode) 356 self.assertEqual(result.zone2_mode, command.zone2_mode) 357 self.assertEqual(result.zone2_enable_manual, command.zone2_enable_manual) 358 self.assertEqual(result.run_all_seconds, command.run_all_seconds) 359 360 def test_stop_matches_observed_write(self) -> None: 361 command = decode_controller_command(START_COMMAND) 362 plan = plan_manual_stop_zone1(command) 363 self.assertEqual(plan.writes[0].payload, STOP_COMMAND) 364 stopped = decode_controller_command(plan.writes[0].payload) 365 366 self.assertFalse(stopped.zone1_enable_manual) 367 self.assertEqual(stopped.zone1_enabled, 2) 368 self.assertEqual(stopped.zone1_mode, command.zone1_mode) 369 self.assertFalse(stopped.zone2_enable_manual) 370 self.assertEqual(stopped.zone2_enabled, 2) 371 self.assertEqual(stopped.run_all_seconds, command.run_all_seconds) 372 373 def test_start_has_three_hour_safety_limit(self) -> None: 374 command = decode_controller_command(INITIAL_COMMAND) 375 config = decode_zone_config(ZONE1_INITIAL) 376 with self.assertRaises(ProtocolError): 377 plan_manual_start_zone1(command, config, 3 * 60 * 60 + 1) 378 379 380 if __name__ == "__main__": 381 unittest.main()