From 3454d65275d7353c6658bd702da492aba3a2c4ef Mon Sep 17 00:00:00 2001 From: Marcel van der Laan Date: Sat, 24 Jan 2026 15:42:28 +0100 Subject: [PATCH 1/3] Adding initial Wallbox implementation --- .../evse_load_balancer/chargers/__init__.py | 3 + .../chargers/wallbox_charger.py | 180 ++++++++++++++++++ custom_components/evse_load_balancer/const.py | 1 + tests/chargers/test_wallbox_charger.py | 172 +++++++++++++++++ 4 files changed, 356 insertions(+) create mode 100644 custom_components/evse_load_balancer/chargers/wallbox_charger.py create mode 100644 tests/chargers/test_wallbox_charger.py diff --git a/custom_components/evse_load_balancer/chargers/__init__.py b/custom_components/evse_load_balancer/chargers/__init__.py index 3ef25fd..93941e2 100644 --- a/custom_components/evse_load_balancer/chargers/__init__.py +++ b/custom_components/evse_load_balancer/chargers/__init__.py @@ -12,6 +12,8 @@ from .keba_charger import KebaCharger from .lektrico_charger import LektricoCharger from .zaptec_charger import ZaptecCharger +from .wallbox_charger import WallboxCharger + if TYPE_CHECKING: from homeassistant.helpers.device_registry import DeviceEntry @@ -34,6 +36,7 @@ async def charger_factory( ZaptecCharger, KebaCharger, LektricoCharger, + WallboxCharger, ]: if charger_cls.is_charger_device(device): return charger_cls(hass, config_entry, device) diff --git a/custom_components/evse_load_balancer/chargers/wallbox_charger.py b/custom_components/evse_load_balancer/chargers/wallbox_charger.py new file mode 100644 index 0000000..8f08241 --- /dev/null +++ b/custom_components/evse_load_balancer/chargers/wallbox_charger.py @@ -0,0 +1,180 @@ +"""Wallbox Charger implementation.""" + +import logging + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceEntry + +from ..const import CHARGER_DOMAIN_WALLBOX, Phase +from ..ha_device import HaDevice +from .charger import Charger, PhaseMode + +_LOGGER = logging.getLogger(__name__) + + +class WallboxEntityMap: + """ + Map Wallbox entities to their respective translation keys. + + These translation keys correspond with what the official Wallbox + integration registers (see its `number` and `sensor` platforms). + """ + + DynamicChargerLimit = "maximum_charging_current" + MaxChargerLimit = "maximum_charging_current" + Status = "status_description" + + +class WallboxStatusMap: + """Normalized status values used by the adapter. + + Values mirror the Wallbox integration `ChargerStatus` strings. + See Home Assistant `wallbox.const.ChargerStatus` for full list. + """ + + Charging = "Charging" + Discharging = "Discharging" + Paused = "Paused" + Scheduled = "Scheduled" + WaitingForCarDemand = "Waiting for car demand" + Waiting = "Waiting" + Disconnected = "Disconnected" + Error = "Error" + Ready = "Ready" + Locked = "Locked" + LockedCarConnected = "Locked, car connected" + Updating = "Updating" + WaitingInQueuePowerSharing = "Waiting in queue by Power Sharing" + WaitingInQueuePowerBoost = "Waiting in queue by Power Boost" + WaitingMidFailed = "Waiting MID failed" + WaitingMidSafety = "Waiting MID safety margin exceeded" + WaitingInQueueEcoSmart = "Waiting in queue by Eco-Smart" + Unknown = "Unknown" + + +class WallboxCharger(HaDevice, Charger): + """Implementation of the Charger class for Wallbox chargers.""" + + def __init__( + self, hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry + ) -> None: + """Initialize the Wallbox charger.""" + HaDevice.__init__(self, hass, device_entry) + Charger.__init__(self, hass, config_entry, device_entry) + self.refresh_entities() + + @staticmethod + def is_charger_device(device: DeviceEntry) -> bool: + """Check if the given device is a Wallbox charger. + + The Wallbox integration registers devices with the domain "wallbox" + as the identifier in the device registry. + """ + return any(id_domain == CHARGER_DOMAIN_WALLBOX for id_domain, _ in device.identifiers) + + async def async_setup(self) -> None: + """Set up the charger (no-op).""" + + def set_phase_mode(self, mode: PhaseMode, _phase: Phase | None = None) -> None: + """Wallbox number entity controls global charging current; phase mode is not managed here.""" + if mode not in PhaseMode: + raise ValueError("Invalid mode. Must be 'single' or 'multi'.") + + def has_synced_phase_limits(self) -> bool: + """Wallbox number entity exposes a single max current value (global).""" + return False + + async def set_current_limit(self, limit: dict[Phase, int]) -> None: + """Set the charger current limit. + + The official Wallbox integration exposes the charging current as a + `number` entity. We update that entity using the common `number.set_value` + service with the entity_id and a numeric `value`. + """ + # Wallbox exposes a single configurable current; be conservative and use lowest phase value + amps = int(min(limit.values())) + + try: + entity_id = self._get_entity_id_by_translation_key( + WallboxEntityMap.DynamicChargerLimit + ) + except ValueError: + _LOGGER.error( + "Wallbox dynamic limit entity not found for device %s", self.device_entry.id + ) + return + + await self.hass.services.async_call( + domain="number", + service="set_value", + service_data={"entity_id": entity_id, "value": amps}, + blocking=True, + ) + + def get_current_limit(self) -> dict[Phase, int] | None: + """Return the currently configured charging limit (from the `number` entity).""" + state = self._get_entity_state_by_translation_key( + WallboxEntityMap.DynamicChargerLimit + ) + if state is None: + _LOGGER.warning( + "Wallbox dynamic charger limit not available for device %s", + self.device_entry.id, + ) + return None + try: + return dict.fromkeys(Phase, int(float(state))) + except (ValueError, TypeError): + _LOGGER.warning("Unable to parse Wallbox dynamic limit state: %s", state) + return None + + def get_max_current_limit(self) -> dict[Phase, int] | None: + """Return the configured maximum charging current.""" + state = self._get_entity_state_by_translation_key(WallboxEntityMap.MaxChargerLimit) + if state is None: + _LOGGER.warning( + "Wallbox max charger limit not available for device %s", + self.device_entry.id, + ) + return None + try: + return dict.fromkeys(Phase, int(float(state))) + except (ValueError, TypeError): + _LOGGER.warning("Unable to parse Wallbox max limit state: %s", state) + return None + + def _get_status(self) -> str | None: + return self._get_entity_state_by_translation_key(WallboxEntityMap.Status) + + def car_connected(self) -> bool: + status = self._get_status() + # `Ready` explicitly means no car connected (confirmed by user). + # Consider the car connected in statuses that indicate a vehicle + # is physically present even if it is not actively charging. + return status in ( + WallboxStatusMap.Charging, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.Paused, + WallboxStatusMap.Discharging, + ) + + def can_charge(self) -> bool: + status = self._get_status() + # The charger can accept or deliver charge when it is actively + # charging or when a car is connected and waiting for demand. + # We treat `READY` as no car connected and therefore not able to + # charge until a car is present. + return status in ( + WallboxStatusMap.Charging, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Paused, + ) + + def is_charging(self) -> bool: + status = self._get_status() + return status == WallboxStatusMap.Charging + + async def async_unload(self) -> None: + """Unload the Wallbox charger (no-op).""" diff --git a/custom_components/evse_load_balancer/const.py b/custom_components/evse_load_balancer/const.py index 8dd828d..5fb7317 100644 --- a/custom_components/evse_load_balancer/const.py +++ b/custom_components/evse_load_balancer/const.py @@ -8,6 +8,7 @@ CHARGER_DOMAIN_ZAPTEC = "zaptec" CHARGER_DOMAIN_LEKTRICO = "lektrico" CHARGER_DOMAIN_KEBA = "keba" +CHARGER_DOMAIN_WALLBOX = "wallbox" HA_INTEGRATION_DOMAIN_MQTT = "mqtt" Z2M_DEVICE_IDENTIFIER_DOMAIN = "zigbee2mqtt" diff --git a/tests/chargers/test_wallbox_charger.py b/tests/chargers/test_wallbox_charger.py new file mode 100644 index 0000000..0040802 --- /dev/null +++ b/tests/chargers/test_wallbox_charger.py @@ -0,0 +1,172 @@ +"""Tests for the Wallbox charger implementation.""" + +from unittest.mock import MagicMock, patch, AsyncMock + +import pytest +from homeassistant.helpers.device_registry import DeviceEntry +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.evse_load_balancer.meters.meter import Phase +from custom_components.evse_load_balancer.chargers.wallbox_charger import ( + WallboxCharger, + WallboxEntityMap, + WallboxStatusMap, + PhaseMode, +) + + +@pytest.fixture +def mock_hass(): + """Create a mock HomeAssistant instance for testing.""" + hass = MagicMock() + hass.services = MagicMock() + hass.services.async_call = AsyncMock() + return hass + + +@pytest.fixture +def mock_config_entry(): + """Create a mock ConfigEntry for the tests.""" + return MockConfigEntry( + domain="evse_load_balancer", + title="Wallbox Test Charger", + data={"charger_type": "wallbox"}, + unique_id="test_wallbox_charger", + ) + + +@pytest.fixture +def mock_device_entry(): + """Create a mock DeviceEntry object for testing.""" + device_entry = MagicMock(spec=DeviceEntry) + device_entry.id = "test_device_id" + device_entry.identifiers = {("wallbox", "test_charger")} + return device_entry + + +@pytest.fixture +def wallbox_charger(mock_hass, mock_config_entry, mock_device_entry): + """Create a WallboxCharger instance for testing.""" + with patch( + "custom_components.evse_load_balancer.chargers.wallbox_charger.WallboxCharger.refresh_entities" + ): + charger = WallboxCharger( + hass=mock_hass, config_entry=mock_config_entry, device_entry=mock_device_entry + ) + # Mock entity lookup/state helpers + charger._get_entity_state_by_translation_key = MagicMock() + charger._get_entity_id_by_translation_key = MagicMock() + return charger + + +async def test_set_current_limit(wallbox_charger, mock_hass): + """Test setting current limits on the Wallbox charger.""" + test_limits = {Phase.L1: 16, Phase.L2: 14, Phase.L3: 15} + entity_id = "number.wallbox_test_maximum_charging_current" + + wallbox_charger._get_entity_id_by_translation_key.return_value = entity_id + + await wallbox_charger.set_current_limit(test_limits) + + mock_hass.services.async_call.assert_called_once_with( + domain="number", + service="set_value", + service_data={"entity_id": entity_id, "value": 14}, + blocking=True, + ) + + +def test_get_current_limit_success(wallbox_charger): + """Test retrieving the current limit when entity exists.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "16" + + result = wallbox_charger.get_current_limit() + assert result == {Phase.L1: 16, Phase.L2: 16, Phase.L3: 16} + wallbox_charger._get_entity_state_by_translation_key.assert_called_with( + WallboxEntityMap.DynamicChargerLimit + ) + + +def test_get_current_limit_missing_entity(wallbox_charger): + """Test retrieving the current limit when entity doesn't exist.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = None + + result = wallbox_charger.get_current_limit() + assert result is None + wallbox_charger._get_entity_state_by_translation_key.assert_called_with( + WallboxEntityMap.DynamicChargerLimit + ) + + +def test_get_max_current_limit_success(wallbox_charger): + """Test retrieving the max current limit when entity exists.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "32" + + result = wallbox_charger.get_max_current_limit() + assert result == {Phase.L1: 32, Phase.L2: 32, Phase.L3: 32} + wallbox_charger._get_entity_state_by_translation_key.assert_called_with( + WallboxEntityMap.MaxChargerLimit + ) + + +def test_get_max_current_limit_missing_entity(wallbox_charger): + """Test retrieving the max current limit when entity doesn't exist.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = None + + result = wallbox_charger.get_max_current_limit() + assert result is None + wallbox_charger._get_entity_state_by_translation_key.assert_called_with( + WallboxEntityMap.MaxChargerLimit + ) + + +def test_car_connected_and_can_charge_and_is_charging(wallbox_charger): + """Test status mapping for connected/can_charge/is_charging.""" + # READY means no car connected + wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.Ready + assert wallbox_charger.car_connected() is False + assert wallbox_charger.can_charge() is False + assert wallbox_charger.is_charging() is False + + # Waiting for car demand means car present but not charging + wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.WaitingForCarDemand + assert wallbox_charger.car_connected() is True + assert wallbox_charger.can_charge() is True + assert wallbox_charger.is_charging() is False + + # Charging + wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.Charging + assert wallbox_charger.car_connected() is True + assert wallbox_charger.can_charge() is True + assert wallbox_charger.is_charging() is True + + # Locked, car connected + wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.LockedCarConnected + assert wallbox_charger.car_connected() is True + + +def test_is_charging_false_for_other_states(wallbox_charger): + """is_charging returns False for non-charging states.""" + for status in [ + WallboxStatusMap.Disconnected, + WallboxStatusMap.Error, + WallboxStatusMap.Scheduled, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.Unknown, + None, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.is_charging() is False + + +def test_set_phase_mode_valid_and_invalid(wallbox_charger): + """set_phase_mode accepts valid PhaseMode and rejects invalid values.""" + # Valid (no-op) + try: + wallbox_charger.set_phase_mode(PhaseMode.SINGLE, Phase.L1) + except ValueError: + pytest.fail("set_phase_mode raised ValueError unexpectedly!") + + # Invalid + with pytest.raises(ValueError): + wallbox_charger.set_phase_mode("invalid_mode", Phase.L1) From 9b91a6ac6a610c08b2a3d08b2f43965dd4385c7c Mon Sep 17 00:00:00 2001 From: Dirk Groenen Date: Tue, 27 Jan 2026 20:47:19 +0100 Subject: [PATCH 2/3] fix(wallbox): correct entity mapping, status hierarchy, config registration, and lint - MaxChargerLimit now uses `max_available_power` sensor (hardware max) instead of reusing the `maximum_charging_current` number entity - has_synced_phase_limits returns True (single global current = synced) - car_connected includes all connected statuses (Scheduled, Waiting, WaitingInQueue*, WaitingMid*) - can_charge includes Discharging to fix hierarchy violation - Register CHARGER_DOMAIN_WALLBOX in config_flow device filter - Add # noqa: TID252, fix D213 docstrings, msg pattern for ValueError, remove unnecessary try/except in set_current_limit - Rewrite tests: correct Phase import, dedicated is_charger_device tests, exhaustive status coverage, float parsing, has_synced test --- .../chargers/wallbox_charger.py | 79 +++---- .../evse_load_balancer/config_flow.py | 2 + tests/chargers/test_wallbox_charger.py | 193 +++++++++++++----- 3 files changed, 187 insertions(+), 87 deletions(-) diff --git a/custom_components/evse_load_balancer/chargers/wallbox_charger.py b/custom_components/evse_load_balancer/chargers/wallbox_charger.py index 8f08241..d015521 100644 --- a/custom_components/evse_load_balancer/chargers/wallbox_charger.py +++ b/custom_components/evse_load_balancer/chargers/wallbox_charger.py @@ -6,8 +6,8 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.device_registry import DeviceEntry -from ..const import CHARGER_DOMAIN_WALLBOX, Phase -from ..ha_device import HaDevice +from ..const import CHARGER_DOMAIN_WALLBOX, Phase # noqa: TID252 +from ..ha_device import HaDevice # noqa: TID252 from .charger import Charger, PhaseMode _LOGGER = logging.getLogger(__name__) @@ -17,20 +17,21 @@ class WallboxEntityMap: """ Map Wallbox entities to their respective translation keys. - These translation keys correspond with what the official Wallbox - integration registers (see its `number` and `sensor` platforms). + https://github.com/home-assistant/core/blob/dev/homeassistant/components/wallbox/number.py + https://github.com/home-assistant/core/blob/dev/homeassistant/components/wallbox/sensor.py """ DynamicChargerLimit = "maximum_charging_current" - MaxChargerLimit = "maximum_charging_current" + MaxChargerLimit = "max_available_power" Status = "status_description" class WallboxStatusMap: - """Normalized status values used by the adapter. + """ + Normalized status values used by the adapter. Values mirror the Wallbox integration `ChargerStatus` strings. - See Home Assistant `wallbox.const.ChargerStatus` for full list. + https://github.com/home-assistant/core/blob/dev/homeassistant/components/wallbox/const.py """ Charging = "Charging" @@ -66,44 +67,38 @@ def __init__( @staticmethod def is_charger_device(device: DeviceEntry) -> bool: - """Check if the given device is a Wallbox charger. - - The Wallbox integration registers devices with the domain "wallbox" - as the identifier in the device registry. - """ - return any(id_domain == CHARGER_DOMAIN_WALLBOX for id_domain, _ in device.identifiers) + """Check if the given device is a Wallbox charger.""" + return any( + id_domain == CHARGER_DOMAIN_WALLBOX + for id_domain, _ in device.identifiers + ) async def async_setup(self) -> None: """Set up the charger (no-op).""" def set_phase_mode(self, mode: PhaseMode, _phase: Phase | None = None) -> None: - """Wallbox number entity controls global charging current; phase mode is not managed here.""" + """Set the phase mode (no-op for Wallbox).""" if mode not in PhaseMode: - raise ValueError("Invalid mode. Must be 'single' or 'multi'.") + msg = "Invalid mode. Must be 'single' or 'multi'." + raise ValueError(msg) def has_synced_phase_limits(self) -> bool: """Wallbox number entity exposes a single max current value (global).""" - return False + return True async def set_current_limit(self, limit: dict[Phase, int]) -> None: - """Set the charger current limit. + """ + Set the charger current limit. The official Wallbox integration exposes the charging current as a - `number` entity. We update that entity using the common `number.set_value` - service with the entity_id and a numeric `value`. + `number` entity. We update that entity using the common + `number.set_value` service with the entity_id and a numeric `value`. """ - # Wallbox exposes a single configurable current; be conservative and use lowest phase value amps = int(min(limit.values())) - try: - entity_id = self._get_entity_id_by_translation_key( - WallboxEntityMap.DynamicChargerLimit - ) - except ValueError: - _LOGGER.error( - "Wallbox dynamic limit entity not found for device %s", self.device_entry.id - ) - return + entity_id = self._get_entity_id_by_translation_key( + WallboxEntityMap.DynamicChargerLimit + ) await self.hass.services.async_call( domain="number", @@ -131,7 +126,9 @@ def get_current_limit(self) -> dict[Phase, int] | None: def get_max_current_limit(self) -> dict[Phase, int] | None: """Return the configured maximum charging current.""" - state = self._get_entity_state_by_translation_key(WallboxEntityMap.MaxChargerLimit) + state = self._get_entity_state_by_translation_key( + WallboxEntityMap.MaxChargerLimit + ) if state is None: _LOGGER.warning( "Wallbox max charger limit not available for device %s", @@ -148,31 +145,35 @@ def _get_status(self) -> str | None: return self._get_entity_state_by_translation_key(WallboxEntityMap.Status) def car_connected(self) -> bool: + """Return whether a car is connected to the charger.""" status = self._get_status() - # `Ready` explicitly means no car connected (confirmed by user). - # Consider the car connected in statuses that indicate a vehicle - # is physically present even if it is not actively charging. return status in ( WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.Paused, + WallboxStatusMap.Scheduled, WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Waiting, WallboxStatusMap.LockedCarConnected, - WallboxStatusMap.Paused, - WallboxStatusMap.Discharging, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, ) def can_charge(self) -> bool: + """Return whether the charger can deliver charge.""" status = self._get_status() - # The charger can accept or deliver charge when it is actively - # charging or when a car is connected and waiting for demand. - # We treat `READY` as no car connected and therefore not able to - # charge until a car is present. return status in ( WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, WallboxStatusMap.WaitingForCarDemand, WallboxStatusMap.Paused, ) def is_charging(self) -> bool: + """Return whether the charger is actively charging.""" status = self._get_status() return status == WallboxStatusMap.Charging diff --git a/custom_components/evse_load_balancer/config_flow.py b/custom_components/evse_load_balancer/config_flow.py index 291dfd8..f5a3645 100644 --- a/custom_components/evse_load_balancer/config_flow.py +++ b/custom_components/evse_load_balancer/config_flow.py @@ -27,6 +27,7 @@ CHARGER_DOMAIN_EASEE, CHARGER_DOMAIN_KEBA, CHARGER_DOMAIN_LEKTRICO, + CHARGER_DOMAIN_WALLBOX, CHARGER_DOMAIN_ZAPTEC, CHARGER_MANUFACTURER_AMINA, DOMAIN, @@ -58,6 +59,7 @@ {"integration": CHARGER_DOMAIN_ZAPTEC}, {"integration": CHARGER_DOMAIN_KEBA}, {"integration": CHARGER_DOMAIN_LEKTRICO}, + {"integration": CHARGER_DOMAIN_WALLBOX}, { "integration": HA_INTEGRATION_DOMAIN_MQTT, "manufacturer": CHARGER_MANUFACTURER_AMINA, diff --git a/tests/chargers/test_wallbox_charger.py b/tests/chargers/test_wallbox_charger.py index 0040802..7c40168 100644 --- a/tests/chargers/test_wallbox_charger.py +++ b/tests/chargers/test_wallbox_charger.py @@ -1,18 +1,18 @@ """Tests for the Wallbox charger implementation.""" -from unittest.mock import MagicMock, patch, AsyncMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest from homeassistant.helpers.device_registry import DeviceEntry from pytest_homeassistant_custom_component.common import MockConfigEntry -from custom_components.evse_load_balancer.meters.meter import Phase +from custom_components.evse_load_balancer.chargers.charger import PhaseMode from custom_components.evse_load_balancer.chargers.wallbox_charger import ( WallboxCharger, WallboxEntityMap, WallboxStatusMap, - PhaseMode, ) +from custom_components.evse_load_balancer.const import CHARGER_DOMAIN_WALLBOX, Phase @pytest.fixture @@ -30,7 +30,7 @@ def mock_config_entry(): return MockConfigEntry( domain="evse_load_balancer", title="Wallbox Test Charger", - data={"charger_type": "wallbox"}, + data={}, unique_id="test_wallbox_charger", ) @@ -39,8 +39,8 @@ def mock_config_entry(): def mock_device_entry(): """Create a mock DeviceEntry object for testing.""" device_entry = MagicMock(spec=DeviceEntry) - device_entry.id = "test_device_id" - device_entry.identifiers = {("wallbox", "test_charger")} + device_entry.id = "wallbox_123" + device_entry.identifiers = {(CHARGER_DOMAIN_WALLBOX, "test_charger")} return device_entry @@ -53,17 +53,27 @@ def wallbox_charger(mock_hass, mock_config_entry, mock_device_entry): charger = WallboxCharger( hass=mock_hass, config_entry=mock_config_entry, device_entry=mock_device_entry ) - # Mock entity lookup/state helpers charger._get_entity_state_by_translation_key = MagicMock() charger._get_entity_id_by_translation_key = MagicMock() return charger +def test_is_charger_device_true(mock_device_entry): + """Test is_charger_device returns True for Wallbox devices.""" + assert WallboxCharger.is_charger_device(mock_device_entry) is True + + +def test_is_charger_device_false(): + """Test is_charger_device returns False for non-Wallbox devices.""" + device = MagicMock(spec=DeviceEntry) + device.identifiers = {("other_domain", "test_charger")} + assert WallboxCharger.is_charger_device(device) is False + + async def test_set_current_limit(wallbox_charger, mock_hass): """Test setting current limits on the Wallbox charger.""" test_limits = {Phase.L1: 16, Phase.L2: 14, Phase.L3: 15} entity_id = "number.wallbox_test_maximum_charging_current" - wallbox_charger._get_entity_id_by_translation_key.return_value = entity_id await wallbox_charger.set_current_limit(test_limits) @@ -87,15 +97,20 @@ def test_get_current_limit_success(wallbox_charger): ) -def test_get_current_limit_missing_entity(wallbox_charger): +def test_get_current_limit_float_value(wallbox_charger): + """Test retrieving the current limit when entity returns float string.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "16.5" + + result = wallbox_charger.get_current_limit() + assert result == {Phase.L1: 16, Phase.L2: 16, Phase.L3: 16} + + +def test_get_current_limit_missing(wallbox_charger): """Test retrieving the current limit when entity doesn't exist.""" wallbox_charger._get_entity_state_by_translation_key.return_value = None result = wallbox_charger.get_current_limit() assert result is None - wallbox_charger._get_entity_state_by_translation_key.assert_called_with( - WallboxEntityMap.DynamicChargerLimit - ) def test_get_max_current_limit_success(wallbox_charger): @@ -109,64 +124,146 @@ def test_get_max_current_limit_success(wallbox_charger): ) -def test_get_max_current_limit_missing_entity(wallbox_charger): +def test_get_max_current_limit_float_value(wallbox_charger): + """Test retrieving the max current limit when entity returns float string.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "32.0" + + result = wallbox_charger.get_max_current_limit() + assert result == {Phase.L1: 32, Phase.L2: 32, Phase.L3: 32} + + +def test_get_max_current_limit_missing(wallbox_charger): """Test retrieving the max current limit when entity doesn't exist.""" wallbox_charger._get_entity_state_by_translation_key.return_value = None result = wallbox_charger.get_max_current_limit() assert result is None - wallbox_charger._get_entity_state_by_translation_key.assert_called_with( - WallboxEntityMap.MaxChargerLimit - ) -def test_car_connected_and_can_charge_and_is_charging(wallbox_charger): - """Test status mapping for connected/can_charge/is_charging.""" - # READY means no car connected - wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.Ready - assert wallbox_charger.car_connected() is False - assert wallbox_charger.can_charge() is False - assert wallbox_charger.is_charging() is False +def test_has_synced_phase_limits(wallbox_charger): + """Test that Wallbox charger always has synced phase limits.""" + assert wallbox_charger.has_synced_phase_limits() is True - # Waiting for car demand means car present but not charging - wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.WaitingForCarDemand - assert wallbox_charger.car_connected() is True - assert wallbox_charger.can_charge() is True - assert wallbox_charger.is_charging() is False - # Charging - wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.Charging - assert wallbox_charger.car_connected() is True - assert wallbox_charger.can_charge() is True - assert wallbox_charger.is_charging() is True +def test_car_connected_true(wallbox_charger): + """Test car_connected returns True for connected statuses.""" + for status in [ + WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.Paused, + WallboxStatusMap.Scheduled, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Waiting, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.car_connected() is True, f"Expected car_connected=True for {status}" - # Locked, car connected - wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.LockedCarConnected - assert wallbox_charger.car_connected() is True +def test_car_connected_false(wallbox_charger): + """Test car_connected returns False for disconnected statuses.""" + for status in [ + WallboxStatusMap.Ready, + WallboxStatusMap.Disconnected, + WallboxStatusMap.Error, + WallboxStatusMap.Locked, + WallboxStatusMap.Updating, + WallboxStatusMap.Unknown, + None, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.car_connected() is False, f"Expected car_connected=False for {status}" -def test_is_charging_false_for_other_states(wallbox_charger): - """is_charging returns False for non-charging states.""" + +def test_can_charge_true(wallbox_charger): + """Test can_charge returns True for chargeable statuses.""" for status in [ + WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Paused, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.can_charge() is True, f"Expected can_charge=True for {status}" + + +def test_can_charge_false(wallbox_charger): + """Test can_charge returns False for non-chargeable statuses.""" + for status in [ + WallboxStatusMap.Ready, WallboxStatusMap.Disconnected, WallboxStatusMap.Error, + WallboxStatusMap.Locked, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.Scheduled, + WallboxStatusMap.Waiting, + WallboxStatusMap.Updating, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + WallboxStatusMap.Unknown, + None, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.can_charge() is False, f"Expected can_charge=False for {status}" + + +def test_is_charging_true(wallbox_charger): + """Test is_charging returns True when actively charging.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.Charging + assert wallbox_charger.is_charging() is True + + +def test_is_charging_false(wallbox_charger): + """Test is_charging returns False for non-charging statuses.""" + for status in [ + WallboxStatusMap.Discharging, + WallboxStatusMap.Paused, WallboxStatusMap.Scheduled, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Waiting, + WallboxStatusMap.Disconnected, + WallboxStatusMap.Error, + WallboxStatusMap.Ready, + WallboxStatusMap.Locked, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.Updating, WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + WallboxStatusMap.WaitingInQueueEcoSmart, WallboxStatusMap.Unknown, None, ]: wallbox_charger._get_entity_state_by_translation_key.return_value = status - assert wallbox_charger.is_charging() is False + assert wallbox_charger.is_charging() is False, f"Expected is_charging=False for {status}" -def test_set_phase_mode_valid_and_invalid(wallbox_charger): - """set_phase_mode accepts valid PhaseMode and rejects invalid values.""" - # Valid (no-op) - try: - wallbox_charger.set_phase_mode(PhaseMode.SINGLE, Phase.L1) - except ValueError: - pytest.fail("set_phase_mode raised ValueError unexpectedly!") +def test_set_phase_mode_valid(wallbox_charger): + """Test set_phase_mode accepts valid PhaseMode values.""" + wallbox_charger.set_phase_mode(PhaseMode.SINGLE, Phase.L1) + wallbox_charger.set_phase_mode(PhaseMode.MULTI, Phase.L1) - # Invalid - with pytest.raises(ValueError): + +def test_set_phase_mode_invalid(wallbox_charger): + """Test set_phase_mode rejects invalid values.""" + with pytest.raises(ValueError, match="Invalid mode"): wallbox_charger.set_phase_mode("invalid_mode", Phase.L1) + + +async def test_async_setup(wallbox_charger): + """Test async_setup method (no-op).""" + await wallbox_charger.async_setup() + + +async def test_async_unload(wallbox_charger): + """Test async_unload method (no-op).""" + await wallbox_charger.async_unload() From 8eb3fb5a36cb5f29ffa499296d172a5a668997f0 Mon Sep 17 00:00:00 2001 From: Dirk Groenen Date: Tue, 27 Jan 2026 20:49:12 +0100 Subject: [PATCH 3/3] Formatting --- custom_components/evse_load_balancer/chargers/__init__.py | 3 +-- .../evse_load_balancer/chargers/wallbox_charger.py | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/custom_components/evse_load_balancer/chargers/__init__.py b/custom_components/evse_load_balancer/chargers/__init__.py index 93941e2..b536900 100644 --- a/custom_components/evse_load_balancer/chargers/__init__.py +++ b/custom_components/evse_load_balancer/chargers/__init__.py @@ -11,9 +11,8 @@ from .easee_charger import EaseeCharger from .keba_charger import KebaCharger from .lektrico_charger import LektricoCharger -from .zaptec_charger import ZaptecCharger from .wallbox_charger import WallboxCharger - +from .zaptec_charger import ZaptecCharger if TYPE_CHECKING: from homeassistant.helpers.device_registry import DeviceEntry diff --git a/custom_components/evse_load_balancer/chargers/wallbox_charger.py b/custom_components/evse_load_balancer/chargers/wallbox_charger.py index d015521..0510a3d 100644 --- a/custom_components/evse_load_balancer/chargers/wallbox_charger.py +++ b/custom_components/evse_load_balancer/chargers/wallbox_charger.py @@ -69,8 +69,7 @@ def __init__( def is_charger_device(device: DeviceEntry) -> bool: """Check if the given device is a Wallbox charger.""" return any( - id_domain == CHARGER_DOMAIN_WALLBOX - for id_domain, _ in device.identifiers + id_domain == CHARGER_DOMAIN_WALLBOX for id_domain, _ in device.identifiers ) async def async_setup(self) -> None: