import json from datetime import datetime, timedelta, timezone import sys import tempfile from pathlib import Path from types import SimpleNamespace import unittest from unittest.mock import Mock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps')) sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps' / 'ad_toolbox')) from boilerregulator import BoilerRegulator, setmode from boilerclimate import BoilerClimate class RegulatorTests(unittest.TestCase): def test_equilibrium_retains_heat_at_target(self): r = BoilerRegulator() v = r.update(18, 18, 5, 60) self.assertTrue(v.enabled) self.assertEqual(v.flow, 47.5) def test_colder_outdoors_and_room_increase_water_request(self): mild = BoilerRegulator().update(18, 18, 10, 0) cold = BoilerRegulator().update(18, 18, 5, 0) cold_room = BoilerRegulator().update(18, 17, 10, 0) self.assertGreater(cold.flow, mild.flow) self.assertGreater(cold_room.flow, mild.flow) def test_persistent_error_integrates_in_water_degrees(self): r = BoilerRegulator(max_flow=65) v = r.update(18, 17.5, 10, 3600) self.assertAlmostEqual(v.integral, 0.1) def test_upper_saturation_does_not_wind_up(self): r = BoilerRegulator() for _ in range(1000): v = r.update(18, 12, -10, 60) self.assertEqual(v.flow, 50) self.assertEqual(v.integral, 0) def test_overheat_and_low_demand_stop_with_hysteresis(self): r = BoilerRegulator() self.assertTrue(r.update(18, 18, 5, 0).enabled) self.assertFalse(r.update(18, 18.6, 5, 60).enabled) self.assertFalse(r.update(18, 18.2, 5, 60).enabled) self.assertTrue(r.update(18, 18, 5, 60).enabled) self.assertFalse(r.update(18, 18, 20, 60).enabled) def test_payload_preserves_sanitary_fields(self): a, b = setmode(40, True), setmode(0, False) self.assertEqual(a['disablehc'], 0) self.assertEqual(b['disablehc'], 1) for p in [a, b]: self.assertIsNone(p['hwctempdesired']) self.assertIsNone(p['hwcflowtempdesired']) self.assertEqual(p['disablehwctapping'], 0) self.assertEqual(p['disablehwcload'], 0) def test_rejects_nonfinite_inputs(self): for value in [float('nan'), float('inf')]: with self.assertRaises(ValueError): BoilerRegulator().update(18, value, 5, 60) class ClimateTests(unittest.TestCase): def app(self): # Test the actual app callbacks without starting an AppDaemon server. app = object.__new__(BoilerClimate) app.base, app.boiler = 'appdaemon/test', 'boiler' app.mode, app.target = 'heat', 18 app.indoor, app.outdoor = 'sensor.indoor', 'sensor.outdoor' app.min_target, app.max_target = 10, 23 app.interval = 60 app.controller = BoilerRegulator() app.send_commands = True app.last_payload = None app.last_send = app.previous_time = 1000 app.last_command_at = app.last_echo_at = None app.telemetry, app.last_status_time = {}, 0 app._publish = Mock() app._save = Mock() app.log = Mock() app.initialize_logger_interface(app, ascii_encode=False) readings = {app.indoor: 18, app.outdoor: 10} app._sensor_value = lambda entity: readings[entity] return app def writes(self, app): return [c for c in app._publish.call_args_list if c.args[0] == 'boiler/bai/SetMode/set'] def test_single_change_and_periodic_refresh_only(self): a = self.app() with patch('boilerclimate.time.monotonic', return_value=1000): a._control() self.assertEqual(len(self.writes(a)), 1) with patch('boilerclimate.time.monotonic', return_value=1030): a._control() self.assertEqual(len(self.writes(a)), 1) with patch('boilerclimate.time.monotonic', return_value=1060): a._control() self.assertEqual(len(self.writes(a)), 2) for call in self.writes(a): self.assertFalse(call.args[2]) def test_invalid_sensor_explicitly_inhibits_heat(self): a = self.app() a._sensor_value = Mock(side_effect=ValueError('unavailable')) a._control() self.assertEqual(self.writes(a)[0].args[1]['disablehc'], 1) self.assertEqual(a.controller.integral, 0) def test_observation_mode_never_writes_boiler(self): a = self.app() a.send_commands = False a._control() self.assertEqual(self.writes(a), []) def test_room_target_rejects_nan_and_out_of_bounds(self): a = self.app() for p in ['nan', 'inf', '30', 'garbage']: a._mqtt_event('', dict(topic=a.base + '/command/temperature', payload=p), {}) self.assertEqual(a.target, 18) self.assertFalse(a._save.called) self.assertEqual(self.writes(a), []) def test_off_command_sends_stop(self): a = self.app() a._mqtt_event('', dict(topic=a.base + '/command/mode', payload='off'), {}) self.assertEqual(self.writes(a)[0].args[1]['disablehc'], 1) def test_native_ha_template_payload_is_accepted(self): a = self.app() a._mqtt_event('', dict(topic='boiler/bai/Statenumber', payload={'value': 30}), {}) self.assertEqual(a.telemetry['Statenumber'], {'value': 30}) self.assertEqual(self.writes(a), []) def reading_app(self, state='18', age=0): a = self.app() stamp = (datetime.now(timezone.utc) - timedelta(seconds=age)).isoformat() a.get_state = Mock(return_value={ 'state': state, 'attributes': {'unit_of_measurement': '°C', 'friendly_name': 'Couloir Température'}, 'last_updated': stamp, }) return a def test_unchanged_temperature_is_valid_even_after_days(self): a = self.reading_app(age=86400 * 3) self.assertEqual(BoilerClimate._sensor_value(a, a.indoor), 18) def test_unavailable_is_invalid_without_waiting_two_hours(self): a = self.reading_app(state='unavailable', age=1) with self.assertRaisesRegex(ValueError, r'sensor.indoor.*invalid state.*unavailable'): BoilerClimate._sensor_value(a, a.indoor) def test_missing_timestamp_does_not_invalidate_temperature(self): a = self.reading_app() a.get_state.return_value.pop('last_updated') self.assertEqual(BoilerClimate._sensor_value(a, a.indoor), 18) def test_fault_logs_error_once_then_logs_recovery(self): a = self.app() a._sensor_value = Mock(side_effect=ValueError('sensor.indoor (Couloir): unavailable')) a._control() a._control() errors = [c for c in a.log.call_args_list if c.kwargs.get('level') == 'ERROR' and c.kwargs.get('log') == 'main_log'] self.assertEqual(len(errors), 1) self.assertIn('Couloir', errors[0].args[0]) self.assertFalse(errors[0].kwargs['ascii_encode']) self.assertTrue(any(c.kwargs.get('log') == 'error_log' for c in a.log.call_args_list)) self.assertIn('sensor.indoor', a.sensor_error) a._sensor_value = lambda entity: 18 if entity == a.indoor else 10 a._control() self.assertIsNone(a.sensor_error) self.assertTrue(any('recovered' in c.args[0] for c in a.log.call_args_list)) def test_telemetry_does_not_refresh_command(self): a = self.app() a._mqtt_event('', dict(topic='boiler/bai/SetMode', payload=json.dumps(setmode(40, True))), {}) self.assertEqual(self.writes(a), []) def test_save_failure_does_not_block_off(self): a = self.app() a._save.side_effect = OSError('disk full') a._mqtt_event('', dict(topic=a.base + '/command/mode', payload='off'), {}) self.assertEqual(self.writes(a)[0].args[1]['disablehc'], 1) self.assertTrue(any('disk full' in c.args[0] for c in a.log.call_args_list)) def test_save_failure_does_not_block_shutdown(self): a = self.app() a._save.side_effect = OSError('disk full') a.terminate() self.assertEqual(self.writes(a)[0].args[1]['disablehc'], 1) a._publish.assert_any_call(a.base + '/availability', 'offline', True) def test_shutdown_publish_failure_still_saves_and_marks_offline(self): a = self.app() def publish(topic, *args): if topic == 'boiler/bai/SetMode/set': raise RuntimeError('service unavailable') a._publish.side_effect = publish a.terminate() a._save.assert_called_once() a._publish.assert_any_call(a.base + '/availability', 'offline', True) def test_invalid_persisted_state_starts_off(self): with tempfile.TemporaryDirectory() as folder: for saved in [{'mode': 'off', 'target': 'invalid'}, {'mode': 'heat', 'target': 30}, {'target': 18}, [], None]: with self.subTest(saved=saved): a = self.app() a.state_file = Path(folder) / 'state.json' a.state_file.write_text(json.dumps(saved)) a._restore() self.assertEqual(a.mode, 'off') self.assertEqual(a.target, 18) self.assertTrue(any(c.kwargs.get('level') == 'WARNING' for c in a.log.call_args_list)) def test_restore_valid_state_and_first_start(self): with tempfile.TemporaryDirectory() as folder: a = self.app() a.state_file = Path(folder) / 'state.json' a._restore() self.assertEqual(a.mode, 'heat') a.state_file.write_text(json.dumps({'mode': 'off', 'target': 19.5})) a._restore() self.assertEqual((a.mode, a.target), ('off', 19.5)) def test_corrupt_or_unreadable_state_starts_off(self): for contents in ['{broken', OSError('permission denied')]: with self.subTest(contents=contents): a = self.app() a.state_file = Mock() if isinstance(contents, OSError): a.state_file.read_text.side_effect = contents else: a.state_file.read_text.return_value = contents a._restore() self.assertEqual(a.mode, 'off') def test_standard_log_pipeline_preserves_unicode_and_error_routing(self): a = self.app() a.log_info('Demande à 40 °C') a.log_warning('État invalide') a.log_error('Échec chaudière') a.log.assert_any_call('Demande à 40 °C', level='INFO', log='main_log', ascii_encode=False) a.log.assert_any_call('État invalide', level='WARNING', log='main_log', ascii_encode=False) a.log.assert_any_call('Échec chaudière', level='ERROR', log='error_log', ascii_encode=False) def test_boiler_status_updates_action_without_advancing_regulation(self): a = self.app() a._control() a._publish.reset_mock() previous_time, integral = a.previous_time, a.controller.integral for status, action in [(4, 'heating'), (14, 'idle'), (0, 'idle')]: a._mqtt_event('', dict(topic='boiler/bai/Statenumber', payload={'value': status}), {}) states = [c.args[1] for c in a._publish.call_args_list if c.args[0] == a.base + '/state'] self.assertEqual(states[-1]['action'], action) self.assertEqual(self.writes(a), []) self.assertEqual(a.previous_time, previous_time) self.assertEqual(a.controller.integral, integral) if __name__ == '__main__': unittest.main()