240 lines
9.0 KiB
Python
240 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from freezegun import freeze_time
|
|
|
|
from tests.conftest import FakeCoordinator, make_state
|
|
from custom_components.easee_solar_charging.const import (
|
|
CONF_STOP_GRACE_MINUTES,
|
|
DEFAULT_STOP_GRACE_MINUTES,
|
|
EASEE_CONTROLLABLE_STATES,
|
|
EASEE_SLEEPING_STATES,
|
|
HYSTERESIS_A,
|
|
MIN_CHARGER_CURRENT_A,
|
|
)
|
|
|
|
_NOW = "2024-06-01 12:00:00"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _should_send
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestShouldSend:
|
|
def test_first_run_always_sends(self, coord):
|
|
assert coord._should_send(8) is True
|
|
|
|
def test_same_value_no_send(self, coord):
|
|
coord._last_sent_current = 8
|
|
assert coord._should_send(8) is False
|
|
|
|
def test_zero_to_nonzero_sends(self, coord):
|
|
coord._last_sent_current = 0
|
|
assert coord._should_send(8) is True
|
|
|
|
def test_nonzero_to_zero_sends(self, coord):
|
|
coord._last_sent_current = 8
|
|
assert coord._should_send(0) is True
|
|
|
|
def test_outside_deadband_sends_immediately(self, coord):
|
|
coord._last_sent_current = 8
|
|
assert coord._should_send(8 + HYSTERESIS_A) is True
|
|
|
|
def test_inside_deadband_suppressed_when_recent(self, coord):
|
|
coord._last_sent_current = 8
|
|
coord._last_sent_at = datetime.now(timezone.utc)
|
|
# 1 A delta — inside dead-band, timestamp is fresh
|
|
assert coord._should_send(9) is False
|
|
|
|
def test_inside_deadband_sends_after_one_minute(self, coord):
|
|
coord._last_sent_current = 8
|
|
coord._last_sent_at = datetime.now(timezone.utc) - timedelta(minutes=1, seconds=1)
|
|
assert coord._should_send(9) is True
|
|
|
|
def test_inside_deadband_sends_when_no_timestamp(self, coord):
|
|
coord._last_sent_current = 8
|
|
coord._last_sent_at = None
|
|
assert coord._should_send(9) is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _apply_stop_hysteresis
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestApplyStopHysteresis:
|
|
def test_nonzero_target_passes_through(self, coord):
|
|
assert coord._apply_stop_hysteresis(10) == 10
|
|
|
|
def test_nonzero_target_clears_timer(self, coord):
|
|
coord._below_threshold_since = datetime.now(timezone.utc)
|
|
coord._apply_stop_hysteresis(10)
|
|
assert coord._below_threshold_since is None
|
|
|
|
def test_zero_starts_grace_period(self, coord):
|
|
coord._last_sent_current = 10
|
|
with freeze_time(_NOW):
|
|
result = coord._apply_stop_hysteresis(0)
|
|
assert result == 10
|
|
assert coord._below_threshold_since is not None
|
|
|
|
def test_zero_held_during_grace(self, coord):
|
|
coord._last_sent_current = 8
|
|
with freeze_time(_NOW):
|
|
coord._apply_stop_hysteresis(0)
|
|
with freeze_time("2024-06-01 12:04:00"):
|
|
result = coord._apply_stop_hysteresis(0)
|
|
assert result == 8
|
|
|
|
def test_zero_applied_after_grace_expires(self, coord):
|
|
coord._last_sent_current = 8
|
|
with freeze_time(_NOW):
|
|
coord._apply_stop_hysteresis(0)
|
|
with freeze_time("2024-06-01 12:05:01"):
|
|
result = coord._apply_stop_hysteresis(0)
|
|
assert result == 0
|
|
|
|
def test_recovery_before_grace_expires_cancels_timer(self, coord):
|
|
coord._last_sent_current = 8
|
|
with freeze_time(_NOW):
|
|
coord._apply_stop_hysteresis(0)
|
|
with freeze_time("2024-06-01 12:02:00"):
|
|
result = coord._apply_stop_hysteresis(10)
|
|
assert result == 10
|
|
assert coord._below_threshold_since is None
|
|
|
|
def test_disabled_passes_through_immediately(self, coord):
|
|
coord.stop_grace_enabled = False
|
|
coord._last_sent_current = 8
|
|
assert coord._apply_stop_hysteresis(0) == 0
|
|
assert coord._below_threshold_since is None
|
|
|
|
def test_disabled_clears_running_timer(self, coord):
|
|
coord._below_threshold_since = datetime.now(timezone.utc)
|
|
coord.stop_grace_enabled = False
|
|
coord._apply_stop_hysteresis(0)
|
|
assert coord._below_threshold_since is None
|
|
|
|
def test_respects_configured_grace_duration(self):
|
|
coord = FakeCoordinator(options={CONF_STOP_GRACE_MINUTES: 2})
|
|
coord._last_sent_current = 8
|
|
with freeze_time(_NOW):
|
|
coord._apply_stop_hysteresis(0)
|
|
with freeze_time("2024-06-01 12:02:01"):
|
|
result = coord._apply_stop_hysteresis(0)
|
|
assert result == 0
|
|
|
|
def test_no_last_sent_during_grace_returns_zero(self, coord):
|
|
# Edge case: grace period started but charger was never commanded
|
|
coord._last_sent_current = None
|
|
with freeze_time(_NOW):
|
|
result = coord._apply_stop_hysteresis(0)
|
|
assert result == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_solar_excess_w
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetSolarExcessW:
|
|
def test_valid_numeric_state(self, coord):
|
|
coord.hass.states.get.return_value = make_state("3500.5")
|
|
assert coord._get_solar_excess_w() == 3500.5
|
|
|
|
def test_integer_state(self, coord):
|
|
coord.hass.states.get.return_value = make_state("5000")
|
|
assert coord._get_solar_excess_w() == 5000.0
|
|
|
|
def test_unavailable_returns_none(self, coord):
|
|
coord.hass.states.get.return_value = make_state("unavailable")
|
|
assert coord._get_solar_excess_w() is None
|
|
|
|
def test_unknown_returns_none(self, coord):
|
|
coord.hass.states.get.return_value = make_state("unknown")
|
|
assert coord._get_solar_excess_w() is None
|
|
|
|
def test_missing_entity_returns_none(self, coord):
|
|
coord.hass.states.get.return_value = None
|
|
assert coord._get_solar_excess_w() is None
|
|
|
|
def test_non_numeric_returns_none(self, coord):
|
|
coord.hass.states.get.return_value = make_state("not_a_number")
|
|
assert coord._get_solar_excess_w() is None
|
|
|
|
def test_queries_correct_entity(self, coord):
|
|
coord.hass.states.get.return_value = make_state("1000")
|
|
coord._get_solar_excess_w()
|
|
coord.hass.states.get.assert_called_once_with(coord._solar_excess_sensor)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _charger_allows_control
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestChargerAllowsControl:
|
|
@pytest.mark.parametrize("state", sorted(EASEE_CONTROLLABLE_STATES))
|
|
def test_allows_controllable_states(self, coord, state):
|
|
coord.hass.states.get.return_value = make_state(state)
|
|
assert coord._charger_allows_control() is True
|
|
|
|
def test_rejects_disconnected(self, coord):
|
|
coord.hass.states.get.return_value = make_state("disconnected")
|
|
assert coord._charger_allows_control() is False
|
|
|
|
def test_rejects_error_state(self, coord):
|
|
coord.hass.states.get.return_value = make_state("error")
|
|
assert coord._charger_allows_control() is False
|
|
|
|
def test_rejects_missing_entity(self, coord):
|
|
coord.hass.states.get.return_value = None
|
|
assert coord._charger_allows_control() is False
|
|
|
|
def test_queries_correct_status_entity(self, coord):
|
|
coord.hass.states.get.return_value = make_state("charging")
|
|
coord._charger_allows_control()
|
|
coord.hass.states.get.assert_called_once_with(
|
|
f"sensor.{coord.charger_id.lower()}_status"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cancel_stop_grace
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _charger_is_sleeping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestChargerIsSleeping:
|
|
@pytest.mark.parametrize("state", sorted(EASEE_SLEEPING_STATES))
|
|
def test_detects_sleeping_states(self, coord, state):
|
|
coord.hass.states.get.return_value = make_state(state)
|
|
assert coord._charger_is_sleeping() is True
|
|
|
|
@pytest.mark.parametrize("state", sorted(EASEE_CONTROLLABLE_STATES))
|
|
def test_not_sleeping_when_controllable(self, coord, state):
|
|
coord.hass.states.get.return_value = make_state(state)
|
|
assert coord._charger_is_sleeping() is False
|
|
|
|
def test_not_sleeping_when_entity_missing(self, coord):
|
|
coord.hass.states.get.return_value = None
|
|
assert coord._charger_is_sleeping() is False
|
|
|
|
def test_not_sleeping_when_disconnected(self, coord):
|
|
coord.hass.states.get.return_value = make_state("disconnected")
|
|
assert coord._charger_is_sleeping() is False
|
|
|
|
|
|
def test_cancel_stop_grace_clears_timer(coord):
|
|
coord._below_threshold_since = datetime.now(timezone.utc)
|
|
coord.cancel_stop_grace()
|
|
assert coord._below_threshold_since is None
|
|
|
|
|
|
def test_cancel_stop_grace_is_idempotent(coord):
|
|
coord._below_threshold_since = None
|
|
coord.cancel_stop_grace()
|
|
assert coord._below_threshold_since is None
|