"""Offline HA boundary doubles; these do not validate the HA runtime itself.""" import importlib.util import sys import types import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import AsyncMock, Mock, patch ROOT = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc' class CoordinatorBase: @classmethod def __class_getitem__(cls, item): return cls def __init__(self, hass, *args, **kwargs): self.hass = hass async def async_config_entry_first_refresh(self): self.data = await self._async_update_data() class Store: def __init__(self, *args): self.async_load = AsyncMock(return_value=None) self.async_save = AsyncMock() self.async_remove = AsyncMock() def load_integration(): modules = {name: types.ModuleType(name) for name in ( 'homeassistant', 'homeassistant.const', 'homeassistant.core', 'homeassistant.config_entries', 'homeassistant.helpers', 'homeassistant.helpers.storage', 'homeassistant.helpers.update_coordinator', 'homeassistant.helpers.entity_registry')} modules['homeassistant.const'].Platform = types.SimpleNamespace(SENSOR='sensor', BINARY_SENSOR='binary_sensor') modules['homeassistant.core'].HomeAssistant = object modules['homeassistant.config_entries'].ConfigEntry = object modules['homeassistant.helpers.storage'].Store = Store modules['homeassistant.helpers.update_coordinator'].DataUpdateCoordinator = CoordinatorBase registry = modules['homeassistant.helpers.entity_registry'] registry.async_get = lambda hass: hass.registry registry.async_entries_for_config_entry = lambda reg, entry: reg.entries name = 'audit_heating_mpc' spec = importlib.util.spec_from_file_location(name, ROOT / '__init__.py', submodule_search_locations=[str(ROOT)]) module = importlib.util.module_from_spec(spec) sys.modules[name] = module with patch.dict(sys.modules, modules): spec.loader.exec_module(module) return module, sys.modules[name + '.coordinator'] integration, coordinator = load_integration() class BoundaryTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.now = datetime.now(timezone.utc) self.states = {} self.hass = types.SimpleNamespace( data={}, states=types.SimpleNamespace(get=self.states.get), config=types.SimpleNamespace(units=types.SimpleNamespace(temperature_unit='°C')), services=types.SimpleNamespace(async_call=Mock(side_effect=AssertionError('No service calls permitted'))), registry=types.SimpleNamespace(entries=[], async_remove=Mock()), config_entries=types.SimpleNamespace( async_forward_entry_setups=AsyncMock(), async_unload_platforms=AsyncMock(return_value=True), async_reload=AsyncMock(), async_update_entry=Mock()), ) async def execute(job): return job() self.hass.async_add_executor_job = execute self.entry = types.SimpleNamespace(entry_id='entry', version=2, data={'room_entities': ['sensor.room'], 'heater_entity': 'switch.boiler', 'climate_entity': 'climate.reference', 'outside_temperature_sources': ['sensor.outside'], 'horizon_minutes': 60}, options={}, async_on_unload=Mock(), add_update_listener=Mock()) self.set_state('sensor.room', '20', unit_of_measurement='°C') self.set_state('sensor.outside', '5', unit_of_measurement='°C') self.set_state('switch.boiler', 'off') self.set_state('climate.reference', 'heat', temperature=22) self.c = coordinator.HeatingMPCCoordinator(self.hass, self.entry) def set_state(self, entity, value, **attrs): self.states[entity] = types.SimpleNamespace(state=value, attributes=attrs, last_changed=self.now - timedelta(hours=1), last_reported=self.now) async def sample(self, seconds=300): with patch.object(coordinator, 'datetime') as clock: clock.now.return_value = self.now data = await self.c._async_update_data() self.now += timedelta(seconds=seconds) return data def ready(self): m = self.c.learners['sensor.room'] m.count, m.on_count, m.off_count = 200, 100, 100 self.c._update_counts() async def test_learning_persistence_and_restart(self): await self.sample() await self.sample() self.assertEqual(self.c.learners['sensor.room'].count, 1) payload = self.c._store.async_save.call_args.args[0] self.assertEqual(payload['schema_version'], 2) new = coordinator.HeatingMPCCoordinator(self.hass, self.entry) new._store.async_load.return_value = payload await new.async_restore_model() self.assertEqual(new._off_samples, 1) self.assertIsNone(new._previous) async def test_transition_and_round_trip_switch_are_skipped(self): for new_state in ('on', 'off'): self.c._previous = None await self.sample() self.states['switch.boiler'].state = new_state self.states['switch.boiler'].last_changed = self.now - timedelta(seconds=60) await self.sample() self.assertEqual(self.c.learners['sensor.room'].count, 0) async def test_gaps_backward_clock_and_excessive_jitter_skipped(self): for seconds in (0, -300, 100, 420, 3600): self.c._previous = None await self.sample(seconds) await self.sample() self.assertEqual(self.c.learners['sensor.room'].count, 0) async def test_missing_sensor_breaks_training_continuity(self): await self.sample() self.states['sensor.room'].state = 'unavailable' self.assertEqual((await self.sample())['status'], 'waiting_for_sensors') self.states['sensor.room'].state = '20' await self.sample() self.assertEqual(self.c.learners['sensor.room'].count, 0) async def test_weather_fallback_skips_transition(self): await self.sample() self.c.config['outside_temperature_sources'].append('sensor.alternate') self.set_state('sensor.alternate', '7', unit_of_measurement='°C') self.states['sensor.outside'].state = 'unavailable' data = await self.sample() self.assertEqual(data['outside_temperature'], 7) self.assertEqual(data['samples'], 0) async def test_outlier_does_not_count_towards_readiness(self): await self.sample() self.states['sensor.room'].state = '25' data = await self.sample() self.assertEqual(data['samples'], 0) self.assertEqual(data['off_samples'], 0) async def test_fahrenheit_target_and_manual_off(self): self.ready() self.hass.config.units.temperature_unit = '°F' self.states['climate.reference'].attributes['temperature'] = 71.6 data = await self.sample() self.assertEqual(data['target'], 22) self.assertEqual(data['status'], 'ready') self.states['climate.reference'].state = 'off' data = await self.sample() self.assertEqual(data['status'], 'paused') self.assertIsNone(data['proposal']) self.assertEqual(self.states['climate.reference'].attributes['temperature'], 71.6) self.hass.services.async_call.assert_not_called() async def test_invalid_target_and_optimizer_failure(self): self.ready() self.states['climate.reference'].attributes['temperature'] = 'nan' self.assertEqual((await self.sample())['status'], 'waiting_for_target') self.states['climate.reference'].attributes['temperature'] = 22 with patch.object(coordinator, 'optimize', side_effect=RuntimeError('test')): data = await self.sample() self.assertEqual(data['status'], 'prediction_error') self.assertIsNone(data['proposal']) async def test_corrupt_storage_and_legacy_counters(self): for payload in ([], 'broken', {'schema_version': 999}, {'learners': []}, {'on_samples': 'bad', 'off_samples': float('inf')}): self.c._store.async_load.return_value = payload await self.c.async_restore_model() self.assertEqual(self.c._on_samples, 0) old = self.c.learners['sensor.room'].as_dict() old.update(count=500) old.pop('on_count'); old.pop('off_count') self.c._store.async_load.return_value = {'learners': {'sensor.room': old}, 'on_samples': 250} await self.c.async_restore_model() self.assertEqual(self.c.learners['sensor.room'].count, 500) self.assertEqual(self.c._on_samples, 0) async def test_replaced_heater_invalidates_model(self): self.c._store.async_load.return_value = {'heater_entity': 'switch.other', 'learners': {'sensor.room': {'count': 999}}} await self.c.async_restore_model() self.assertEqual(self.c.learners['sensor.room'].count, 0) async def test_setup_reload_unload_and_registry_cleanup(self): obsolete = types.SimpleNamespace(platform='heating_mpc', unique_id='entry_sensor.old_predicted_60min', entity_id='sensor.old_forecast') kept = types.SimpleNamespace(platform='heating_mpc', unique_id='entry_sensor.room_predicted_60min', entity_id='sensor.forecast') self.hass.registry.entries = [obsolete, kept] self.assertTrue(await integration.async_setup_entry(self.hass, self.entry)) self.hass.registry.async_remove.assert_called_once_with('sensor.old_forecast') await integration.async_reload_entry(self.hass, self.entry) self.hass.config_entries.async_reload.assert_awaited_once_with('entry') self.hass.config_entries.async_unload_platforms.return_value = False self.assertFalse(await integration.async_unload_entry(self.hass, self.entry)) self.assertIn('entry', self.hass.data['heating_mpc']) self.hass.config_entries.async_unload_platforms.return_value = True self.assertTrue(await integration.async_unload_entry(self.hass, self.entry)) self.assertNotIn('entry', self.hass.data['heating_mpc']) self.hass.services.async_call.assert_not_called() async def test_failed_platform_setup_removes_coordinator(self): self.hass.config_entries.async_forward_entry_setups.side_effect = RuntimeError('test') with self.assertRaises(RuntimeError): await integration.async_setup_entry(self.hass, self.entry) self.assertNotIn('entry', self.hass.data['heating_mpc']) async def test_uninstall_removes_storage(self): store = Store() with patch.object(integration, 'Store', return_value=store): await integration.async_remove_entry(self.hass, self.entry) store.async_remove.assert_awaited_once() async def test_config_migration_preserves_options_precedence(self): self.entry.version = 1 self.entry.data['outside_entity'] = 'sensor.old' self.entry.options = {'outside_entity': 'sensor.override'} self.assertTrue(await integration.async_migrate_entry(self.hass, self.entry)) kwargs = self.hass.config_entries.async_update_entry.call_args.kwargs self.assertEqual(kwargs['options']['outside_temperature_sources'], ['sensor.override']) self.assertEqual(kwargs['data']['outside_temperature_sources'], ['sensor.outside']) self.assertNotIn('outside_entity', kwargs['data']) self.assertEqual(kwargs['version'], 2) self.entry.version = 3 self.assertFalse(await integration.async_migrate_entry(self.hass, self.entry)) async def test_changed_inputs_during_optimization_discard_result(self): self.ready() async def changed(job): output = job() self.set_state('climate.reference', 'off', temperature=22) return output self.hass.async_add_executor_job = changed data = await self.sample() self.assertEqual(data['status'], 'inputs_changed') self.assertIsNone(data['proposal']) async def test_invalid_weights_do_not_publish_proposal(self): for weight in (0, -1, 'bad', float('nan'), float('inf')): self.c.config['weights'] = {'sensor.room': weight} data = await self.sample() self.assertEqual(data['status'], 'invalid_weights') self.assertIsNone(data['proposal']) async def test_added_room_needs_its_own_excitation_evidence(self): model = coordinator.Learner(count=500, on_count=250, off_count=250) self.entry.options['room_entities'] = ['sensor.room', 'sensor.new'] self.c = coordinator.HeatingMPCCoordinator(self.hass, self.entry) self.c._store.async_load.return_value = {'learners': {'sensor.room': model.as_dict()}} await self.c.async_restore_model() self.assertEqual(self.c.learners['sensor.room'].count, 500) self.assertEqual(self.c._on_samples, 0) self.assertEqual(self.c._off_samples, 0) async def test_changed_inputs_during_save_discard_result(self): self.ready() await self.sample() async def changed(payload): self.set_state('climate.reference', 'off', temperature=22) self.c._store.async_save.side_effect = changed data = await self.sample() self.assertEqual(data['status'], 'inputs_changed') self.assertIsNone(data['proposal']) async def test_stale_room_pauses_and_recovery_skips_gap(self): await self.sample() self.states['sensor.room'].last_reported = self.now - timedelta(minutes=31) data = await self.sample() self.assertEqual(data['status'], 'waiting_for_sensors') self.assertEqual(data['input_quality']['rooms']['sensor.room']['status'], 'stale') self.assertIsNone(data['proposal']) self.set_state('sensor.room', '20', unit_of_measurement='°C') data = await self.sample() self.assertEqual(data['samples'], 0) self.assertEqual(data['status'], 'learning') async def test_stale_weather_fallback_and_optional_outage(self): self.c.config['outside_temperature_sources'].append('sensor.alternate') self.set_state('sensor.alternate', '7', unit_of_measurement='°C') self.states['sensor.outside'].last_reported = self.now - timedelta(minutes=121) self.c.config['outside_wind_sources'] = ['sensor.missing'] data = await self.sample() self.assertEqual(data['outside_temperature'], 7) self.assertEqual(data['status'], 'learning') quality = data['input_quality']['weather'] self.assertEqual(quality['temperature']['sensor.outside']['status'], 'stale') self.assertEqual(quality['wind']['sensor.missing']['status'], 'missing') async def test_fresh_unchanged_reading_is_valid(self): self.states['sensor.room'].last_changed = self.now - timedelta(days=2) self.assertEqual((await self.sample())['status'], 'learning') async def test_configured_age_limit_applied(self): self.c.config['room_max_age_minutes'] = 5 self.states['sensor.room'].last_reported = self.now - timedelta(minutes=6) self.assertEqual((await self.sample())['status'], 'waiting_for_sensors') async def test_invalid_unit_diagnosed_by_metric(self): self.states['sensor.outside'].attributes['unit_of_measurement'] = 'km/h' data = await self.sample() self.assertEqual(data['input_quality']['weather']['temperature']['sensor.outside']['status'], 'invalid_value_or_unit')