""" Unit tests for forecast-related coordinator methods. Covers: _w_at_lookahead, _get_raw_forecast_w, _get_adjusted_forecast_w, and the forecast profile branches of _get_solar_excess_w. """ from __future__ import annotations from datetime import datetime, timedelta, timezone import pytest from freezegun import freeze_time from tests.conftest import FakeCoordinator, make_state from custom_components.easee_solar_charging.const import ( CHARGING_PROFILE_ALL_SURPLUS, CHARGING_PROFILE_CONSERVATIVE, CHARGING_PROFILE_FORECAST, CHARGING_PROFILE_FORECAST_CONSERVATIVE, CONF_CHARGING_PROFILE, CONF_FORECAST_CONFIDENCE, CONF_FORECAST_LOOKAHEAD_MINUTES, CONF_FORECAST_SENSOR, DEFAULT_FORECAST_CONFIDENCE, DEFAULT_FORECAST_LOOKAHEAD_MINUTES, MIN_CHARGER_CURRENT_A, PHASES, VOLTAGE_V, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _FREEZE = "2024-06-15 10:00:00" # UTC; used by freezegun-based tests _FREEZE_DT = datetime(2024, 6, 15, 10, 0, 0, tzinfo=timezone.utc) _FORECAST_ENTITY = "sensor.forecast_solar_power_production_now" def _coord(options: dict | None = None) -> FakeCoordinator: return FakeCoordinator(options=options) def _utc_str(dt: datetime) -> str: """ISO string with explicit UTC offset.""" return dt.strftime("%Y-%m-%dT%H:%M:%S+00:00") def _naive_str(dt: datetime) -> str: """Naive local-time string — the raw Forecast.Solar API format.""" return dt.strftime("%Y-%m-%d %H:%M:%S") def _set_rolling( coord: FakeCoordinator, production: float | None, house: float = 0.0, ev: float = 0.0, ) -> None: """Patch _rolling_average on coord with fixed values for all three sensors.""" sensor_map = { coord._production_sensor: production, coord._house_load_sensor: house, coord._ev_charging_sensor: ev, } coord._rolling_average = lambda entity_id: sensor_map.get(entity_id) def _set_forecast_state(coord: FakeCoordinator, value: str, attributes: dict | None = None) -> None: """Make coord.hass.states.get return the given state for the forecast sensor.""" coord.hass.states.get.return_value = make_state(value, attributes) # --------------------------------------------------------------------------- # _w_at_lookahead # --------------------------------------------------------------------------- class TestWAtLookahead: """Timestamp-nearest lookup in a Forecast.Solar watts dict.""" @freeze_time(_FREEZE) def test_exact_match_returns_correct_watts(self): coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) watts = { _utc_str(_FREEZE_DT): 1000.0, _utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0, _utc_str(_FREEZE_DT + timedelta(minutes=60)): 7000.0, } assert coord._w_at_lookahead(watts) == 5000.0 @freeze_time(_FREEZE) def test_picks_entry_nearest_to_target(self): """When no exact match, the closest timestamp wins.""" coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) # +20 min is 10 min before target; +50 min is 20 min after — +20 wins watts = { _utc_str(_FREEZE_DT + timedelta(minutes=20)): 3000.0, _utc_str(_FREEZE_DT + timedelta(minutes=50)): 6000.0, } assert coord._w_at_lookahead(watts) == 3000.0 @freeze_time(_FREEZE) def test_naive_string_keys_treated_as_utc(self): """Naive 'YYYY-MM-DD HH:MM:SS' strings get the local timezone (UTC here).""" coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) target = _FREEZE_DT + timedelta(minutes=30) watts = { _naive_str(target - timedelta(hours=1)): 1000.0, _naive_str(target): 5000.0, _naive_str(target + timedelta(hours=1)): 3000.0, } assert coord._w_at_lookahead(watts) == 5000.0 @freeze_time(_FREEZE) def test_datetime_object_keys(self): """datetime objects as dict keys are handled directly.""" coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) target = _FREEZE_DT + timedelta(minutes=30) watts = { _FREEZE_DT: 1000.0, target: 8000.0, _FREEZE_DT + timedelta(hours=1): 5000.0, } assert coord._w_at_lookahead(watts) == 8000.0 @freeze_time(_FREEZE) def test_empty_dict_returns_none(self): coord = _coord() assert coord._w_at_lookahead({}) is None @freeze_time(_FREEZE) def test_malformed_keys_are_skipped(self): """Unparseable timestamp strings are silently skipped.""" coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) watts = { "not-a-timestamp": 9999.0, _utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0, } assert coord._w_at_lookahead(watts) == 5000.0 @freeze_time(_FREEZE) def test_respects_lookahead_minutes_option(self): """Different lookahead values select different entries.""" coord_30 = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) coord_60 = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 60}) watts = { _utc_str(_FREEZE_DT + timedelta(minutes=30)): 4000.0, _utc_str(_FREEZE_DT + timedelta(minutes=60)): 8000.0, } assert coord_30._w_at_lookahead(watts) == 4000.0 assert coord_60._w_at_lookahead(watts) == 8000.0 @freeze_time(_FREEZE) def test_default_lookahead_is_30_minutes(self): """With no option set, the default lookahead is 30 minutes.""" coord = _coord(options={}) watts = { _utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0, _utc_str(_FREEZE_DT + timedelta(minutes=60)): 9000.0, } assert coord._w_at_lookahead(watts) == 5000.0 @freeze_time(_FREEZE) def test_invalid_timezone_falls_back_to_utc(self): """An unresolvable timezone string falls back to UTC without crashing.""" coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) coord.hass.config.time_zone = "Not/AReal_Timezone" watts = {_utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0} assert coord._w_at_lookahead(watts) == 5000.0 @freeze_time(_FREEZE) def test_single_entry_is_always_returned(self): """A dict with only one entry always returns that entry.""" coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30}) watts = {_utc_str(_FREEZE_DT + timedelta(hours=5)): 2500.0} assert coord._w_at_lookahead(watts) == 2500.0 # --------------------------------------------------------------------------- # _get_raw_forecast_w # --------------------------------------------------------------------------- class TestGetRawForecastW: """Reading the raw (pre-confidence) forecast value from a HA sensor.""" def test_returns_none_when_no_sensor_configured(self): coord = _coord(options={}) assert coord._get_raw_forecast_w() is None def test_returns_none_when_sensor_state_unavailable(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "unavailable") assert coord._get_raw_forecast_w() is None def test_returns_none_when_sensor_state_unknown(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "unknown") assert coord._get_raw_forecast_w() is None def test_returns_none_when_sensor_entity_missing(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) coord.hass.states.get.return_value = None assert coord._get_raw_forecast_w() is None @freeze_time(_FREEZE) def test_uses_watts_attribute_for_lookahead(self): """When the sensor has a 'watts' attribute, uses it for the lookahead lookup.""" coord = _coord(options={ CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_LOOKAHEAD_MINUTES: 30, }) watts_attr = {_utc_str(_FREEZE_DT + timedelta(minutes=30)): 6500.0} _set_forecast_state(coord, "3000", {"watts": watts_attr}) # Should return the lookahead value (6500), not the sensor state (3000) assert coord._get_raw_forecast_w() == 6500.0 def test_falls_back_to_sensor_state_when_no_watts_attribute(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "4200.5") assert coord._get_raw_forecast_w() == 4200.5 def test_falls_back_to_sensor_state_when_watts_attribute_is_empty(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "3000.0", {"watts": {}}) assert coord._get_raw_forecast_w() == 3000.0 def test_returns_none_for_non_numeric_state_with_no_watts_attr(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "banana") assert coord._get_raw_forecast_w() is None def test_ignores_non_dict_watts_attribute(self): """A 'watts' attribute that is not a dict is ignored; state is used instead.""" coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "5000.0", {"watts": "not-a-dict"}) assert coord._get_raw_forecast_w() == 5000.0 def test_state_value_is_returned_as_float(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "7000") result = coord._get_raw_forecast_w() assert result == 7000.0 assert isinstance(result, float) # --------------------------------------------------------------------------- # _get_adjusted_forecast_w # --------------------------------------------------------------------------- class TestGetAdjustedForecastW: """Confidence factor is applied to the raw forecast value.""" def test_default_confidence_is_80_percent(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "10000.0") # 10 000 × 0.80 = 8 000 assert coord._get_adjusted_forecast_w() == pytest.approx(8000.0) def test_custom_confidence_is_applied(self): coord = _coord(options={ CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 60, }) _set_forecast_state(coord, "5000.0") # 5 000 × 0.60 = 3 000 assert coord._get_adjusted_forecast_w() == pytest.approx(3000.0) def test_100_percent_confidence_returns_raw_value(self): coord = _coord(options={ CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_forecast_state(coord, "7000.0") assert coord._get_adjusted_forecast_w() == pytest.approx(7000.0) def test_returns_none_when_sensor_unavailable(self): coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY}) _set_forecast_state(coord, "unavailable") assert coord._get_adjusted_forecast_w() is None def test_returns_none_when_no_sensor_configured(self): coord = _coord(options={}) assert coord._get_adjusted_forecast_w() is None def test_50_percent_confidence_halves_the_value(self): coord = _coord(options={ CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 50, }) _set_forecast_state(coord, "8000.0") assert coord._get_adjusted_forecast_w() == pytest.approx(4000.0) # --------------------------------------------------------------------------- # _get_solar_excess_w — forecast profile branches # --------------------------------------------------------------------------- class TestForecastLookaheadProfile: """CHARGING_PROFILE_FORECAST: use forecast when it exceeds actual surplus.""" def test_uses_forecast_when_higher_than_actual(self): """Forecast beats actual → forecast value used as surplus.""" coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=2000.0, house=500.0, ev=0.0) _set_forecast_state(coord, "8000.0") # 8000 > actual (1500) → wins surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(8000.0) def test_uses_actual_when_higher_than_forecast(self): """Actual surplus beats forecast → actual used (avoids reducing charge).""" coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=9000.0, house=500.0, ev=0.0) _set_forecast_state(coord, "3000.0") # 3000 < actual (8500) surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(8500.0) def test_falls_back_to_actual_when_no_sensor_configured(self): """Without a forecast sensor, behaves identically to Conservative.""" coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST}) _set_rolling(coord, production=5000.0, house=1000.0, ev=500.0) coord.hass.states.get.return_value = None surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(3500.0) def test_confidence_reduces_forecast_before_comparison(self): coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 80, }) _set_rolling(coord, production=1000.0, house=0.0, ev=0.0) _set_forecast_state(coord, "10000.0") # 10 000 × 0.80 = 8 000 > actual (1 000) surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(8000.0) def test_when_both_below_threshold_actual_is_returned(self): """Below-threshold values flow to hysteresis logic — surplus is still the max.""" coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=500.0, house=0.0, ev=0.0) _set_forecast_state(coord, "300.0") # forecast (300) < actual (500) surplus, *_ = coord._get_solar_excess_w() # actual wins because 500 > 300 assert surplus == pytest.approx(500.0) def test_returns_full_four_tuple(self): coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=6000.0, house=1000.0, ev=200.0) _set_forecast_state(coord, "4000.0") surplus, production, house, ev = coord._get_solar_excess_w() assert production == pytest.approx(6000.0) assert house == pytest.approx(1000.0) assert ev == pytest.approx(200.0) # actual (4800) > forecast (4000) → actual assert surplus == pytest.approx(4800.0) class TestForecastSurplusProfile: """CHARGING_PROFILE_FORECAST_CONSERVATIVE: forecast_adj − house_load − ev_charging.""" def test_subtracts_loads_from_forecast(self): coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=2000.0, house=1500.0, ev=500.0) _set_forecast_state(coord, "8000.0") surplus, *_ = coord._get_solar_excess_w() # 8 000 − 1 500 − 500 = 6 000 assert surplus == pytest.approx(6000.0) def test_falls_back_to_conservative_when_no_sensor(self): coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE}) _set_rolling(coord, production=5000.0, house=1000.0, ev=200.0) coord.hass.states.get.return_value = None surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(3800.0) def test_confidence_applied_before_subtracting_loads(self): coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 80, }) _set_rolling(coord, production=1000.0, house=1000.0, ev=0.0) _set_forecast_state(coord, "10000.0") surplus, *_ = coord._get_solar_excess_w() # (10 000 × 0.80) − 1 000 − 0 = 7 000 assert surplus == pytest.approx(7000.0) def test_surplus_can_be_negative(self): """A large house load can push forecast surplus below zero.""" coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=1000.0, house=5000.0, ev=0.0) _set_forecast_state(coord, "4000.0") surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(-1000.0) def test_returns_full_four_tuple(self): coord = _coord(options={ CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE, CONF_FORECAST_SENSOR: _FORECAST_ENTITY, CONF_FORECAST_CONFIDENCE: 100, }) _set_rolling(coord, production=6000.0, house=1500.0, ev=300.0) _set_forecast_state(coord, "9000.0") surplus, production, house, ev = coord._get_solar_excess_w() assert production == pytest.approx(6000.0) assert house == pytest.approx(1500.0) assert ev == pytest.approx(300.0) assert surplus == pytest.approx(7200.0) # 9 000 − 1 500 − 300 # --------------------------------------------------------------------------- # _get_solar_excess_w — baseline profiles (sanity checks) # --------------------------------------------------------------------------- class TestBaselineProfiles: """Conservative and All Surplus profiles still work correctly.""" def test_conservative_subtracts_all_loads(self): coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_CONSERVATIVE}) _set_rolling(coord, production=8000.0, house=2000.0, ev=1000.0) surplus, production, house, ev = coord._get_solar_excess_w() assert surplus == pytest.approx(5000.0) assert production == pytest.approx(8000.0) def test_all_surplus_ignores_loads(self): coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_ALL_SURPLUS}) _set_rolling(coord, production=8000.0, house=3000.0, ev=2000.0) surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(8000.0) def test_unknown_profile_falls_back_to_conservative(self): coord = _coord(options={CONF_CHARGING_PROFILE: "not_a_real_profile"}) _set_rolling(coord, production=7000.0, house=1000.0, ev=500.0) surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(5500.0) def test_production_unavailable_returns_none_surplus(self): """When the production sensor is unavailable, surplus is None.""" coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_CONSERVATIVE}) _set_rolling(coord, production=None, house=1000.0, ev=200.0) surplus, production, *_ = coord._get_solar_excess_w() assert surplus is None assert production is None def test_default_profile_is_conservative(self): """With no charging_profile option, Conservative is used.""" coord = _coord(options={}) _set_rolling(coord, production=5000.0, house=1500.0, ev=0.0) surplus, *_ = coord._get_solar_excess_w() assert surplus == pytest.approx(3500.0)