Coverage for tests/tests_rf/device/test_hvac_ventilator.py: 0%

1#!/usr/bin/env python3 

2"""Unittests for the HvacVentilator class.""" 

3 

4from collections.abc import Generator 

5from unittest.mock import AsyncMock, MagicMock, patch 

6 

7import pytest 

8 

9from ramses_rf.const import DevType 

10from ramses_rf.database import MessageIndex 

11from ramses_rf.device.hvac import HvacVentilator 

12from ramses_rf.gateway import Gateway 

13from ramses_tx import Address 

14from ramses_tx.const import Code 

15from ramses_tx.schemas import DeviceIdT 

16 

17# Test data 

18TEST_DEVICE_ID = "32:123456" 

19TEST_PARAM_ID = "3F" 

20TEST_PARAM_VALUE = 50 

21TEST_MSG_ID = "1234" 

22TEST_BOUND_DEVICE_ID = "37:123456" 

23TEST_BOUND_DEVICE_TYPE = DevType.REM 

24 

25 

26@pytest.fixture 

27def mock_gateway() -> Generator[MagicMock, None, None]: 

28 """Create a mock Gateway instance for testing.""" 

29 gateway = MagicMock(spec=Gateway) 

30 gateway.send_cmd = AsyncMock() 

31 gateway.dispatcher = MagicMock() 

32 gateway.dispatcher.send = MagicMock() 

33 

34 # Add required attributes 

35 gateway.config = MagicMock() 

36 gateway.config.disable_discovery = False 

37 gateway.config.enable_eavesdrop = False 

38 gateway._loop = MagicMock() 

39 gateway._loop.call_soon = MagicMock() 

40 gateway._loop.call_later = MagicMock() 

41 gateway._loop.time = MagicMock(return_value=0.0) 

42 gateway._include = {} 

43 # Add msg_db attribute accessed by the message store, activates the SQLite MessageIndex 

44 gateway.msg_db = MessageIndex(maintain=False) 

45 

46 yield gateway 

47 

48 

49@pytest.fixture 

50def hvac_ventilator(mock_gateway: MagicMock) -> HvacVentilator: 

51 """Create an HvacVentilator instance for testing.""" 

52 device_id = DeviceIdT(TEST_DEVICE_ID) 

53 return HvacVentilator(mock_gateway, Address(device_id)) 

54 

55 

56class TestHvacVentilator: 

57 """Test HvacVentilator class.""" 

58 

59 def test_initialization(self, hvac_ventilator: HvacVentilator) -> None: 

60 """Test that the ventilator initializes correctly.""" 

61 assert hvac_ventilator._supports_2411 is False 

62 assert hvac_ventilator._initialized_callback is None 

63 assert hvac_ventilator._param_update_callback is None 

64 assert hvac_ventilator._hgi is None 

65 assert hvac_ventilator._bound_devices == {} 

66 

67 if hvac_ventilator._gwy.msg_db: 

68 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

69 

70 def test_set_initialized_callback_clear( 

71 self, hvac_ventilator: HvacVentilator 

72 ) -> None: 

73 """Test clearing the initialized callback.""" 

74 # Set a callback first 

75 mock_callback = MagicMock() 

76 hvac_ventilator.set_initialized_callback(mock_callback) 

77 

78 # Now clear it 

79 hvac_ventilator.set_initialized_callback(None) 

80 assert hvac_ventilator._initialized_callback is None 

81 

82 if hvac_ventilator._gwy.msg_db: 

83 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

84 

85 def test_set_initialized_callback_set( 

86 self, hvac_ventilator: HvacVentilator 

87 ) -> None: 

88 """Test setting the initialized callback.""" 

89 # Test initial state 

90 assert hvac_ventilator._initialized_callback is None 

91 

92 if hvac_ventilator._gwy.msg_db: 

93 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

94 

95 # Set the callback 

96 mock_callback = MagicMock() 

97 hvac_ventilator.set_initialized_callback(mock_callback) 

98 

99 assert hvac_ventilator._initialized_callback is mock_callback 

100 

101 def test_set_param_update_callback(self, hvac_ventilator: HvacVentilator) -> None: 

102 """Test setting the parameter update callback.""" 

103 if hvac_ventilator._gwy.msg_db: 

104 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

105 

106 # Define a mock callback 

107 mock_callback = MagicMock() 

108 

109 # Set the callback 

110 hvac_ventilator.set_param_update_callback(mock_callback) 

111 

112 # Check that the callback was set 

113 assert hvac_ventilator._param_update_callback is mock_callback 

114 

115 def test_handle_2411_message(self, hvac_ventilator: HvacVentilator) -> None: 

116 """Test handling a 2411 message.""" 

117 # Create a mock message with all required attributes 

118 msg = MagicMock() 

119 msg.code = Code._2411 

120 # Create proper mock objects for src and dst with id attribute 

121 msg.src = MagicMock() 

122 msg.src.id = TEST_DEVICE_ID 

123 msg.dst = MagicMock() 

124 msg.dst.id = TEST_DEVICE_ID 

125 msg.verb = " I" 

126 msg.payload = {"parameter": TEST_PARAM_ID, "value": TEST_PARAM_VALUE} 

127 

128 # Set up the message store 

129 hvac_ventilator._params_2411 = {} 

130 

131 # Set up the param update callback 

132 mock_callback = MagicMock() 

133 hvac_ventilator.set_param_update_callback(mock_callback) 

134 

135 # Call the method 

136 hvac_ventilator._handle_2411_message(msg) 

137 

138 # Check that supports_2411 was set to True 

139 assert hvac_ventilator._supports_2411 is True 

140 

141 # Check that the message was stored correctly 

142 assert TEST_PARAM_ID in hvac_ventilator._params_2411 

143 assert hvac_ventilator._params_2411[TEST_PARAM_ID] == TEST_PARAM_VALUE 

144 

145 # Check that the callback was called with the correct parameters 

146 mock_callback.assert_called_once_with(TEST_PARAM_ID, TEST_PARAM_VALUE) 

147 

148 if hvac_ventilator._gwy.msg_db: 

149 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

150 

151 @patch("ramses_rf.device.hvac.Command.get_fan_param") 

152 async def test_setup_discovery_cmds( 

153 self, mock_cmd: MagicMock, hvac_ventilator: HvacVentilator 

154 ) -> None: 

155 """Test that discovery commands are set up correctly.""" 

156 # Mock the command creation 

157 mock_cmd.return_value = "MOCK_CMD" 

158 

159 # Use patch.object to properly mock the method 

160 with patch.object(hvac_ventilator, "_add_discovery_cmd") as mock_add_cmd: 

161 hvac_ventilator._setup_discovery_cmds() 

162 

163 # Check that _add_discovery_cmd was called at least once 

164 assert mock_add_cmd.called 

165 

166 if hvac_ventilator._gwy.msg_db: 

167 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

168 

169 async def test_handle_msg_parameter_message( 

170 self, hvac_ventilator: HvacVentilator 

171 ) -> None: 

172 """Test that parameter messages are handled correctly.""" 

173 # Create a mock message with all required attributes 

174 msg = MagicMock() 

175 msg.code = Code._2411 

176 # Create proper mock objects for src and dst with id attribute 

177 msg.src = MagicMock() 

178 msg.src.id = TEST_DEVICE_ID 

179 msg.dst = MagicMock() 

180 msg.dst.id = TEST_DEVICE_ID 

181 msg.verb = " I" 

182 msg.payload = { 

183 "parameter": TEST_PARAM_ID, 

184 "value": TEST_PARAM_VALUE, 

185 "_hgi": MagicMock(), 

186 } 

187 

188 # Set up the message store # deprecated, TODO(eb): remove Q1 2026 

189 if not hvac_ventilator._gwy.msg_db: 

190 hvac_ventilator._msgs_ = {} 

191 

192 # Patch the _handle_2411_message method 

193 with patch.object(hvac_ventilator, "_handle_2411_message") as mock_handle: 

194 # Call the method 

195 hvac_ventilator._handle_msg(msg) 

196 

197 # Check that _handle_2411_message was called 

198 mock_handle.assert_called_once_with(msg) 

199 

200 if hvac_ventilator._gwy.msg_db: 

201 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

202 

203 async def test_handle_msg_non_parameter_message( 

204 self, hvac_ventilator: HvacVentilator 

205 ) -> None: 

206 """Test that non-parameter messages are passed to the parent class.""" 

207 # Create a mock message with a non-parameter code and required attributes 

208 msg = MagicMock() 

209 msg.code = Code._31DA # Standard FAN status code 

210 # Create proper mock objects for src and dst with id attribute 

211 msg.src = MagicMock() 

212 msg.src.id = TEST_DEVICE_ID 

213 msg.dst = MagicMock() 

214 msg.dst.id = TEST_DEVICE_ID 

215 msg.verb = " I" 

216 msg.payload = {"some_key": "some_value"} 

217 

218 # Set up the message store # deprecated, TODO(eb): remove Q1 2026 

219 if not hvac_ventilator._gwy.msg_db: 

220 hvac_ventilator._msgs_ = {} 

221 

222 # Patch the parent class's _handle_msg method 

223 with patch( 

224 "ramses_rf.device.hvac.FilterChange._handle_msg" 

225 ) as mock_parent_handle: 

226 # Call the method 

227 hvac_ventilator._handle_msg(msg) 

228 

229 # Check that the parent's _handle_msg was called 

230 mock_parent_handle.assert_called_once_with(msg) 

231 

232 # The parameter handler should not have been called 

233 assert not hasattr(hvac_ventilator, "_handle_parameter_msg") 

234 

235 if hvac_ventilator._gwy.msg_db: 

236 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

237 

238 def test_add_bound_device( 

239 self, hvac_ventilator: HvacVentilator, caplog: pytest.LogCaptureFixture 

240 ) -> None: 

241 """Test adding a bound device.""" 

242 # Ensure the logger is at the right level to capture the warning 

243 import logging 

244 

245 logger = logging.getLogger("ramses_rf.device.hvac") 

246 logger.setLevel(logging.WARNING) 

247 

248 # Clear any existing log handlers to avoid duplicates 

249 for handler in logger.handlers[:]: 

250 logger.removeHandler(handler) 

251 

252 # Add a bound device 

253 hvac_ventilator.add_bound_device(TEST_BOUND_DEVICE_ID, TEST_BOUND_DEVICE_TYPE) 

254 

255 # Verify it was added 

256 assert TEST_BOUND_DEVICE_ID in hvac_ventilator._bound_devices 

257 assert ( 

258 hvac_ventilator._bound_devices[TEST_BOUND_DEVICE_ID] 

259 == TEST_BOUND_DEVICE_TYPE 

260 ) 

261 

262 # Test with invalid device type - should log a warning but not raise 

263 invalid_device_id = "00:123456" 

264 

265 # Clear the caplog and add a handler to capture the logs 

266 caplog.clear() 

267 with caplog.at_level(logging.WARNING, logger="ramses_rf.device.hvac"): 

268 hvac_ventilator.add_bound_device(invalid_device_id, "INVALID_TYPE") 

269 

270 # Check if any record contains the expected message 

271 expected_message = f"Cannot bind device {invalid_device_id} of type INVALID_TYPE to FAN {hvac_ventilator.id}: must be REM or DIS" 

272 

273 # Print debug info if the test fails 

274 if not any(expected_message in record.message for record in caplog.records): 

275 print("\nCaptured log records:") 

276 for i, record in enumerate(caplog.records): 

277 print(f" {i}: {record.levelname}: {record.message}") 

278 

279 # Check if the warning was logged 

280 assert any(expected_message in record.message for record in caplog.records), ( 

281 f"Expected warning message not found in logs. Expected: {expected_message}" 

282 ) 

283 

284 if hvac_ventilator._gwy.msg_db: 

285 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

286 

287 def test_remove_bound_device(self, hvac_ventilator: HvacVentilator) -> None: 

288 """Test removing a bound device.""" 

289 # Add then remove a device 

290 hvac_ventilator.add_bound_device(TEST_BOUND_DEVICE_ID, TEST_BOUND_DEVICE_TYPE) 

291 hvac_ventilator.remove_bound_device(TEST_BOUND_DEVICE_ID) 

292 

293 # Verify it was removed 

294 assert TEST_BOUND_DEVICE_ID not in hvac_ventilator._bound_devices 

295 

296 # Removing non-existent device should not raise 

297 hvac_ventilator.remove_bound_device("nonexistent:device") 

298 

299 if hvac_ventilator._gwy.msg_db: 

300 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

301 

302 def test_get_bound_rem(self, hvac_ventilator: HvacVentilator) -> None: 

303 """Test getting a bound REM device.""" 

304 # Initially should return None 

305 assert hvac_ventilator.get_bound_rem() is None 

306 

307 # Add a REM device 

308 hvac_ventilator.add_bound_device(TEST_BOUND_DEVICE_ID, DevType.REM) 

309 

310 # Should return the REM device 

311 assert hvac_ventilator.get_bound_rem() == TEST_BOUND_DEVICE_ID 

312 

313 # Add a DIS device, should still return the REM device 

314 hvac_ventilator.add_bound_device("38:123456", DevType.DIS) 

315 assert hvac_ventilator.get_bound_rem() == TEST_BOUND_DEVICE_ID 

316 

317 if hvac_ventilator._gwy.msg_db: 

318 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

319 

320 def test_get_fan_param_supported(self, hvac_ventilator: HvacVentilator) -> None: 

321 """Test getting a supported fan parameter.""" 

322 # Set up the parameter in the device's parameter store 

323 hvac_ventilator._params_2411[TEST_PARAM_ID] = TEST_PARAM_VALUE 

324 

325 # Mark as supporting 2411 

326 hvac_ventilator._supports_2411 = True 

327 

328 # Test getting the parameter 

329 value = hvac_ventilator.get_fan_param(TEST_PARAM_ID) 

330 assert value == TEST_PARAM_VALUE 

331 

332 if hvac_ventilator._gwy.msg_db: 

333 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

334 

335 def test_get_fan_param_unsupported( 

336 self, hvac_ventilator: HvacVentilator, caplog: pytest.LogCaptureFixture 

337 ) -> None: 

338 """Test getting a parameter when 2411 is not supported.""" 

339 # Ensure 2411 is not supported and clear any existing messages 

340 hvac_ventilator._supports_2411 = False 

341 

342 # Test getting a parameter 

343 caplog.clear() 

344 value = hvac_ventilator.get_fan_param(TEST_PARAM_ID) 

345 assert value is None 

346 

347 if hvac_ventilator._gwy.msg_db: 

348 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

349 

350 def test_get_fan_param_normalization(self, hvac_ventilator: HvacVentilator) -> None: 

351 """Test parameter ID normalization.""" 

352 # Set up the parameter with leading zeros in the parameter store 

353 # The get_fan_param method normalizes "03F" to "3F" 

354 hvac_ventilator._params_2411["3F"] = 75 

355 hvac_ventilator._supports_2411 = True 

356 

357 # Test with different formats of the same parameter ID 

358 assert hvac_ventilator.get_fan_param("03F") == 75 

359 assert hvac_ventilator.get_fan_param("3F") == 75 

360 assert hvac_ventilator.get_fan_param("0003F") == 75 

361 

362 if hvac_ventilator._gwy.msg_db: 

363 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

364 

365 def test_initialized_callback(self, hvac_ventilator: HvacVentilator) -> None: 

366 """Test the initialized callback behaviour.""" 

367 # Set up a mock callback 

368 mock_callback = MagicMock() 

369 hvac_ventilator.set_initialized_callback(mock_callback) 

370 

371 # Initially, the callback shouldn't be called 

372 mock_callback.assert_not_called() 

373 

374 # Set supports_2411 to True and call _handle_initialized_callback 

375 hvac_ventilator._supports_2411 = True 

376 hvac_ventilator._handle_initialized_callback() 

377 

378 # The callback should be called once 

379 mock_callback.assert_called_once() 

380 

381 # The callback should be cleared after being called 

382 assert hvac_ventilator._initialized_callback is None 

383 

384 # Calling again should not call the callback again 

385 hvac_ventilator._handle_initialized_callback() 

386 mock_callback.assert_called_once() # Still only called once 

387 

388 if hvac_ventilator._gwy.msg_db: 

389 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

390 

391 def test_hgi_property( 

392 self, hvac_ventilator: HvacVentilator, monkeypatch: pytest.MonkeyPatch 

393 ) -> None: 

394 """Test the hgi property and its caching behaviour.""" 

395 # Get the gateway's hgi 

396 gateway_hgi = hvac_ventilator._gwy.hgi 

397 

398 # First call should get the value from the gateway 

399 assert hvac_ventilator.hgi is gateway_hgi 

400 

401 # The gateway's hgi property should only be called once 

402 # (second access comes from the cache) 

403 assert hvac_ventilator.hgi is gateway_hgi 

404 assert hvac_ventilator._hgi is gateway_hgi # Check the cache directly 

405 

406 # Test the caching behaviour by creating a new mock for the gateway's hgi 

407 new_hgi = MagicMock() 

408 

409 # Use monkeypatch to temporarily replace the hgi property 

410 monkeypatch.setattr(hvac_ventilator._gwy, "hgi", new_hgi) 

411 

412 # The property still returns the original cached value 

413 assert hvac_ventilator.hgi is gateway_hgi 

414 assert hvac_ventilator.hgi is not new_hgi 

415 

416 # Clear the cache 

417 hvac_ventilator._hgi = None 

418 if hvac_ventilator._gwy.msg_db: 

419 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

420 

421 # Now it should get the new value 

422 assert hvac_ventilator.hgi is new_hgi 

423 assert hvac_ventilator._hgi is new_hgi # Check the cache was updated 

424 

425 def test_invalid_message_handling(self, hvac_ventilator: HvacVentilator) -> None: 

426 """Test handling of invalid messages.""" 

427 # Create an invalid message (missing payload) 

428 msg = MagicMock() 

429 msg.verb = " I" 

430 msg.src = MagicMock() 

431 msg.src.id = TEST_DEVICE_ID 

432 msg.dst = MagicMock() 

433 msg.dst.id = TEST_DEVICE_ID 

434 msg.payload = None # Invalid payload 

435 

436 # Set up a callback to verify it's not called 

437 mock_callback = MagicMock() 

438 hvac_ventilator.set_param_update_callback(mock_callback) 

439 

440 # This should not raise an exception 

441 hvac_ventilator._handle_2411_message(msg) 

442 

443 # No parameter update callback should be called 

444 mock_callback.assert_not_called() 

445 

446 if hvac_ventilator._gwy.msg_db: 

447 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection 

448 

449 def test_missing_callback(self, hvac_ventilator: HvacVentilator) -> None: 

450 """Test behaviour when callbacks are not set.""" 

451 # This should not raise an exception 

452 hvac_ventilator._handle_param_update("3F", 50) 

453 

454 # And with a message that would trigger callbacks 

455 msg = MagicMock() 

456 msg.code = Code._2411 

457 msg.verb = " I" 

458 msg.src = MagicMock() 

459 msg.src.id = TEST_DEVICE_ID 

460 msg.dst = MagicMock() 

461 msg.dst.id = TEST_DEVICE_ID 

462 msg.payload = {"parameter": "3F", "value": 50} 

463 

464 # This should not raise an exception 

465 hvac_ventilator._handle_2411_message(msg) 

466 

467 # Now set a callback and verify it's called 

468 mock_callback = MagicMock() 

469 hvac_ventilator.set_param_update_callback(mock_callback) 

470 

471 # Process the message again 

472 hvac_ventilator._handle_2411_message(msg) 

473 

474 # Callback should be called with the parameter and value 

475 mock_callback.assert_called_once_with("3F", 50) 

476 

477 if hvac_ventilator._gwy.msg_db: 

478 hvac_ventilator._gwy.msg_db.stop() # close sqlite3 connection