diff --git a/custom_components/zaptec/__init__.py b/custom_components/zaptec/__init__.py index 997be231..bbc83cd1 100644 --- a/custom_components/zaptec/__init__.py +++ b/custom_components/zaptec/__init__.py @@ -26,9 +26,11 @@ from .manager import ZaptecConfigEntry, ZaptecManager from .services import async_setup_services, async_unload_services from .zaptec import ( + RETRYABLE_HTTP_STATUSES, AuthenticationError, Installation, RequestConnectionError, + RequestError, RequestTimeoutError, Zaptec, ZaptecApiError, @@ -36,6 +38,27 @@ _LOGGER = logging.getLogger(__name__) + +def _config_entry_error( + err: ZaptecApiError, +) -> ConfigEntryAuthFailed | ConfigEntryNotReady | ConfigEntryError: + """Map a Zaptec API error from setup login to a HA config-entry error. + + Authentication failures are non-recoverable (trigger re-auth). Connection + and timeout errors, and transient server statuses (429/502/503/504), are + recoverable, so we raise ConfigEntryNotReady to let Home Assistant retry + setup automatically instead of failing permanently (issue #392). All other + API errors remain permanent ConfigEntryError failures. + """ + if isinstance(err, AuthenticationError): + return ConfigEntryAuthFailed(str(err)) + if isinstance(err, (RequestTimeoutError, RequestConnectionError)): + return ConfigEntryNotReady(str(err)) + if isinstance(err, RequestError) and err.error_code in RETRYABLE_HTTP_STATUSES: + return ConfigEntryNotReady(str(err)) + return ConfigEntryError(str(err)) + + PLATFORMS = [ Platform.BINARY_SENSOR, Platform.BUTTON, @@ -80,18 +103,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # Login to the Zaptec account try: await zaptec.login() - except AuthenticationError as err: - _LOGGER.error("Authentication failed: %s", err) - raise ConfigEntryAuthFailed from err - except (RequestTimeoutError, RequestConnectionError) as err: - _LOGGER.error("Connection error: %s", err) - raise ConfigEntryNotReady from err except ZaptecApiError as err: - _LOGGER.error("Zaptec API error: %s", err) - raise ConfigEntryError from err + _LOGGER.error("Zaptec login failed: %s", err) + raise _config_entry_error(err) from err # Get the structure of devices from Zaptec and determine the zaptec objects to track - tracked_devices = await ZaptecManager.first_time_setup( + tracked_devices, all_selected_present = await ZaptecManager.first_time_setup( zaptec=zaptec, configured_chargers=configured_chargers, ) @@ -200,7 +217,50 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # Make a set of the circuit ids from zaptec to check for deprecated Circuit-devices circuit_ids = {cid for c in manager.zaptec.chargers if (cid := c.get("CircuitId"))} - # Clean up unused device entries with no entities + check_untracked = _should_check_untracked(configured_chargers, all_selected_present) + if configured_chargers is not None and not all_selected_present: + _LOGGER.warning( + "One or more selected chargers were not returned by the Zaptec API " + "this session; skipping removal of untracked devices to avoid deleting " + "a still-selected charger due to a transient/partial response" + ) + _cleanup_stale_devices(hass, entry, manager.tracked_devices, circuit_ids, check_untracked) + + return True + + +def _should_check_untracked( + configured_chargers: set[str] | None, all_selected_present: bool +) -> bool: + """Only treat `tracked_devices` as authoritative for device removal in manual-select mode. + + In track-all mode (`configured_chargers` is None) nothing is ever deliberately + deselected, so there's no reliable signal to distinguish a charger actually + removed from the account from one merely missing from a partial API response + this session - untracked-device removal must never run there. In manual-select + mode it's safe only once `first_time_setup` has confirmed every selected + charger was present this session (`all_selected_present`). + """ + return configured_chargers is not None and all_selected_present + + +def _cleanup_stale_devices( + hass: HomeAssistant, + entry: ConfigEntry, + tracked_devices: set[str], + circuit_ids: set[str], + check_untracked: bool, +) -> None: + """Remove device entries that no longer belong to this config entry. + + Three cases are handled: devices left with no entities at all, deprecated + Circuit-devices (pre-dating a hierarchy change), and, only when + `check_untracked` is True, devices whose zaptec id is no longer tracked, + e.g. a charger deselected via reconfigure. `check_untracked` must be False + whenever `tracked_devices` might be incomplete due to a partial API + response rather than a real deselection, or a transient blip could + permanently delete a still-selected charger's device and entities. + """ device_registry = dr.async_get(hass) entity_registry = er.async_get(hass) @@ -222,12 +282,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: "removing device and associated entities", zap_dev_id, ) - for ent in dev_entities: - _LOGGER.debug("Deleting entity %s", ent.entity_id) - entity_registry.async_remove(ent.entity_id) - device_registry.async_remove_device(dev.id) + elif check_untracked and zap_dev_id not in tracked_devices: + _LOGGER.warning( + "Detected stale device %s no longer selected, " + "removing device and associated entities", + zap_dev_id, + ) + else: + continue - return True + for ent in dev_entities: + _LOGGER.debug("Deleting entity %s", ent.entity_id) + entity_registry.async_remove(ent.entity_id) + device_registry.async_remove_device(dev.id) def remove_deprecated_entities(hass: HomeAssistant, entry: ZaptecConfigEntry) -> None: diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 79b9fc02..26d6ff9d 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -229,8 +229,18 @@ async def stream_callback(self, event: dict) -> None: coordinator.async_update_listeners() @staticmethod - async def first_time_setup(zaptec: Zaptec, configured_chargers: set[str] | None) -> set[str]: - """Run the first time setup for the account.""" + async def first_time_setup( + zaptec: Zaptec, configured_chargers: set[str] | None + ) -> tuple[set[str], bool]: + """Run the first time setup for the account. + + Returns the set of tracked device ids, plus a bool that is False if + one or more user-selected chargers were not returned by the Zaptec + API this session (a transient/partial response, not a deselection). + Callers must not treat a device as user-deselected off `tracked_devices` + alone when this is False, or a temporary API blip could be mistaken + for the user having removed the charger. + """ _LOGGER.debug("Running first time setup") # Build the Zaptec hierarchy @@ -238,6 +248,7 @@ async def first_time_setup(zaptec: Zaptec, configured_chargers: set[str] | None) all_objects = set(zaptec) tracked_devices = all_objects + all_selected_present = True # Selected chargers to add if configured_chargers is not None: @@ -247,6 +258,7 @@ async def first_time_setup(zaptec: Zaptec, configured_chargers: set[str] | None) # Log if there are any objects listed not found in Zaptec if not_present := want - all_objects: _LOGGER.error("Charger objects %s not found", not_present) + all_selected_present = False # Calculate the objects to keep. From the list of chargers we # want to keep, we also want to keep the installation objects. @@ -263,4 +275,4 @@ async def first_time_setup(zaptec: Zaptec, configured_chargers: set[str] | None) # These objects will be updated by the coordinator tracked_devices = keep - return tracked_devices + return tracked_devices, all_selected_present diff --git a/custom_components/zaptec/services.yaml b/custom_components/zaptec/services.yaml index 1eee58b4..0d4024ac 100644 --- a/custom_components/zaptec/services.yaml +++ b/custom_components/zaptec/services.yaml @@ -73,7 +73,7 @@ restart_charger: description: Charger identifier example: 00000000-1111-2222-3333-444444444444 -update_firmware: +upgrade_firmware: name: Update firmware description: >- Send update firmware request to the charger. Select charger diff --git a/custom_components/zaptec/zaptec/__init__.py b/custom_components/zaptec/zaptec/__init__.py index 45d06971..a71ebc9a 100644 --- a/custom_components/zaptec/zaptec/__init__.py +++ b/custom_components/zaptec/zaptec/__init__.py @@ -3,11 +3,12 @@ from __future__ import annotations from .api import Charger, Installation, Zaptec, ZaptecBase -from .const import MISSING, Missing +from .const import MISSING, RETRYABLE_HTTP_STATUSES, Missing from .exceptions import ( AuthenticationError, RequestConnectionError, RequestDataError, + RequestError, RequestRetryError, RequestTimeoutError, ZaptecApiError, @@ -18,6 +19,7 @@ __all__ = [ "MISSING", + "RETRYABLE_HTTP_STATUSES", "ZCONST", "AuthenticationError", "Charger", @@ -26,6 +28,7 @@ "Redactor", "RequestConnectionError", "RequestDataError", + "RequestError", "RequestRetryError", "RequestTimeoutError", "Zaptec", diff --git a/custom_components/zaptec/zaptec/api.py b/custom_components/zaptec/zaptec/api.py index ed9593ea..87f35ce8 100644 --- a/custom_components/zaptec/zaptec/api.py +++ b/custom_components/zaptec/zaptec/api.py @@ -35,6 +35,7 @@ DEFAULT_MAX_CURRENT, MAX_DEBUG_TEXT_LEN_ON_500, MISSING, + RETRYABLE_HTTP_STATUSES, TOKEN_URL, TRUTHY, ) @@ -958,6 +959,19 @@ async def _response_log(resp: aiohttp.ClientResponse) -> AsyncGenerator[str]: except Exception: _LOGGER.exception("Failed to log response (ignored exception)") + @staticmethod + def _parse_retry_after(value: str | None) -> float | None: + """Parse a Retry-After header value into seconds. + + Only the integer delta-seconds form is supported; the HTTP-date form + returns None so the caller falls back to the exponential backoff.""" + if not value: + return None + try: + return max(0.0, float(value)) + except (TypeError, ValueError): + return None + async def _request_worker( self, url: str, method: str = "get", retries: int = API_RETRIES, **kwargs: Any ) -> AsyncGenerator[tuple[aiohttp.ClientResponse, TLogExc]]: @@ -971,6 +985,7 @@ async def _request_worker( error: Exception | None = None delay: float = API_RETRY_INIT_DELAY sleep_delay: float = 0.0 + retry_after: float | None = None start_time: float = time.perf_counter() iteration = 0 for iteration in range(1, retries + 1): @@ -1016,6 +1031,22 @@ def log_exc( _LOGGER.error(str(exc), exc_info=exc) return exc + # Retry transient, infrastructure-level server errors + # (429/502/503/504) regardless of method. These indicate + # the request likely never reached the application, so a + # retry is safe even for POST/PUT -- unlike 500, which is + # handled per-method by the caller. On the final iteration + # we fall through to yield so the caller raises the error. + if response.status in RETRYABLE_HTTP_STATUSES and iteration < retries: + if DEBUG_API_CALLS: + _LOGGER.debug( + "@@@ RETRYABLE STATUS %s, retrying (attempt %s)", + response.status, + iteration, + ) + retry_after = self._parse_retry_after(response.headers.get("Retry-After")) + continue # Retry after backoff (or the Retry-After delay) + # Let the caller handle the response. If the caller # calls __next__ on the generator the request will be # retried. @@ -1043,6 +1074,12 @@ def log_exc( # longer than the calculated delay, so we don't need to sleep. sleep_delay = delay - time.perf_counter() + start_time + # A Retry-After header from a transient response overrides the + # computed backoff for the next attempt. + if retry_after is not None: + sleep_delay = min(retry_after, self._max_time) + retry_after = None + if isinstance(error, TimeoutError): raise RequestTimeoutError( f"Request to {url} timed out after {iteration} retries" diff --git a/custom_components/zaptec/zaptec/const.py b/custom_components/zaptec/zaptec/const.py index 251e6979..0cf95c7d 100644 --- a/custom_components/zaptec/zaptec/const.py +++ b/custom_components/zaptec/zaptec/const.py @@ -18,6 +18,15 @@ class Missing: API_RETRIES = 8 # Corresponds to median ~100 seconds of retries before giving up """Number of retries for API requests.""" +RETRYABLE_HTTP_STATUSES = frozenset({429, 502, 503, 504}) +"""Transient HTTP statuses that are retried with backoff regardless of method. + +Too Many Requests, Bad Gateway, Service Unavailable and Gateway Timeout are +infrastructure-level "try again shortly" errors where the request typically +never reached the application, so retrying is safe even for POST/PUT. This is +distinct from 500 (Internal Server Error), which Zaptec returns in various +application-level cases and is only retried for GET (see ``Zaptec.request``).""" + API_RETRY_INIT_DELAY = 0.3 """Initial delay for the first API retry.""" diff --git a/tests/conftest.py b/tests/conftest.py index ba07a5c5..0f3ca51c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,12 +2,64 @@ import asyncio import os +from typing import Any +from unittest.mock import MagicMock +import aiohttp import pytest from custom_components.zaptec.zaptec.api import Zaptec +class FakeConfigEntry: + """Minimal stand-in for HA's ConfigEntry. + + Exposes only what coordinator.py and entity.py actually touch + (`pref_disable_polling`, `async_on_unload`, + `async_create_background_task`). A real ConfigEntry pulls in HA's full + test-harness machinery, which cannot run on native Windows in this dev + environment - see CLAUDE.md's environment notes. + """ + + pref_disable_polling = False + title = "Mock Title" + + def async_on_unload(self, func: Any) -> None: + """No-op stand-in for HA's unload-callback registration. Never invoked by these tests.""" + + def async_create_background_task( + self, hass: Any, target: Any, name: str, eager_start: bool = True + ) -> asyncio.Task: + """Schedule target as a real asyncio Task. + + This ensures trigger_poll()'s cancel-and-replace logic is genuinely + exercised by tests. + """ + return asyncio.ensure_future(target) + + +@pytest.fixture +def config_entry() -> FakeConfigEntry: + """A fake config entry for coordinator/entity tests.""" + return FakeConfigEntry() + + +@pytest.fixture +async def hass() -> MagicMock: + """A minimal fake HomeAssistant object exposing a real running event loop. + + DataUpdateCoordinator only reads `hass.loop` (to schedule refreshes via + `loop.call_at()`/`loop.time()`); coordinator.py and entity.py never touch + any other HomeAssistant functionality (config, states, services, etc.). + `is_stopping` is pinned False to match a real (non-shutting-down) + HomeAssistant instance, since a bare MagicMock would otherwise be truthy. + """ + fake_hass = MagicMock() + fake_hass.loop = asyncio.get_running_loop() + fake_hass.is_stopping = False + return fake_hass + + @pytest.fixture(scope="session") def skip_if_in_github_actions() -> None: """Check if we are running in Github actions and skip any dependant tests if true.""" @@ -59,7 +111,12 @@ def zaptec_constants() -> dict: """Get latest constants from Zaptec API.""" async def get_zaptec_constants() -> dict: - async with Zaptec("N/A", "N/A") as zaptec: + # aiohttp defaults to aiodns/pycares (AsyncResolver) whenever they're + # importable, which bypasses the OS resolver stack and fails to + # contact DNS in this dev environment. Force the OS-backed resolver. + connector = aiohttp.TCPConnector(resolver=aiohttp.resolver.ThreadedResolver()) + client = aiohttp.ClientSession(connector=connector) + async with Zaptec("N/A", "N/A", client=client) as zaptec: # the constants API endpoint does not require login const: dict = await zaptec.request("constants") return const diff --git a/tests/test_binary_sensor.py b/tests/test_binary_sensor.py new file mode 100644 index 00000000..a3390579 --- /dev/null +++ b/tests/test_binary_sensor.py @@ -0,0 +1,68 @@ +"""Tests for binary_sensor.py.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from homeassistant.helpers.entity import DeviceInfo +import pytest + +from custom_components.zaptec.binary_sensor import ( + ZapBinarySensorEntityDescription, + ZaptecBinarySensor, + ZaptecBinarySensorWithAttrs, +) +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.zaptec import Charger + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +def make_charger(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Charger) whose .get() reads from data.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + charger.qual_id = "Charger[charger1]" + charger.get.side_effect = data.get + return charger + + +def test_binary_sensor_update_from_zaptec_sets_is_on( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecBinarySensor._update_from_zaptec reads the raw boolean value for its key.""" + charger = make_charger({"is_online": True}) + description = ZapBinarySensorEntityDescription(key="is_online", cls=ZaptecBinarySensor) + entity = ZaptecBinarySensor(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_is_on is True # noqa: SLF001 + assert entity._attr_available is True # noqa: SLF001 + + +def test_binary_sensor_with_attrs_post_init_sets_attrs_and_unique_id( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecBinarySensorWithAttrs._post_init copies all raw attrs and overrides unique_id.""" + charger = make_charger({}) + charger.asdict.return_value = {"Id": "charger1", "Active": True} + description = ZapBinarySensorEntityDescription(key="active", cls=ZaptecBinarySensorWithAttrs) + entity = ZaptecBinarySensorWithAttrs(coordinator, charger, description, DeviceInfo()) + + assert entity._attr_extra_state_attributes == {"Id": "charger1", "Active": True} # noqa: SLF001 + assert entity._attr_unique_id == "charger1" # noqa: SLF001 diff --git a/tests/test_button.py b/tests/test_button.py new file mode 100644 index 00000000..5a89b51e --- /dev/null +++ b/tests/test_button.py @@ -0,0 +1,91 @@ +"""Tests for button.py.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity import DeviceInfo +import pytest + +from custom_components.zaptec.button import ZapButtonEntityDescription, ZaptecButton +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.zaptec import Charger + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +def make_charger(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Charger) whose .get() reads from data.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + charger.qual_id = "Charger[charger1]" + charger.get.side_effect = data.get + return charger + + +def test_button_available_delegates_to_is_command_valid( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecButton.available checks is_command_valid using its own key as the command.""" + charger = make_charger({}) + charger.is_command_valid.return_value = True + description = ZapButtonEntityDescription(key="restart_charger", cls=ZaptecButton) + entity = ZaptecButton(coordinator, charger, description, DeviceInfo()) + + assert entity.available is True + charger.is_command_valid.assert_called_once_with("restart_charger") + + +def test_button_unavailable_when_command_invalid(coordinator: ZaptecUpdateCoordinator) -> None: + """ZaptecButton.available is False when is_command_valid returns False.""" + charger = make_charger({}) + charger.is_command_valid.return_value = False + description = ZapButtonEntityDescription(key="resume_charging", cls=ZaptecButton) + entity = ZaptecButton(coordinator, charger, description, DeviceInfo()) + + assert entity.available is False + + +async def test_button_press_sends_command_and_polls( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_press sends the command named by the button's key and triggers a poll.""" + charger = make_charger({}) + charger.command = AsyncMock() + description = ZapButtonEntityDescription(key="restart_charger", cls=ZaptecButton) + entity = ZaptecButton(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_press() + + charger.command.assert_awaited_once_with("restart_charger") + entity.trigger_poll.assert_awaited_once() + + +async def test_button_press_wraps_command_failure(coordinator: ZaptecUpdateCoordinator) -> None: + """async_press wraps a command failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.command = AsyncMock(side_effect=Exception("boom")) + description = ZapButtonEntityDescription(key="restart_charger", cls=ZaptecButton) + entity = ZaptecButton(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_press() + + entity.trigger_poll.assert_not_called() diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py new file mode 100644 index 00000000..84e403f5 --- /dev/null +++ b/tests/test_coordinator.py @@ -0,0 +1,298 @@ +"""Tests for coordinator.py.""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from homeassistant.helpers.update_coordinator import UpdateFailed +import pytest + +from custom_components.zaptec.const import ( + DOMAIN, + ZAPTEC_POLL_CHARGER_TRIGGER_DELAYS, + ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS, +) +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.zaptec import Charger, Installation, Zaptec, ZaptecApiError + + +@pytest.fixture +def manager() -> MagicMock: + """A fake ZaptecManager exposing only what the coordinator touches.""" + mgr = MagicMock() + mgr.zaptec = MagicMock(spec=Zaptec) + mgr.device_coordinators = {} + mgr.tracked_devices = set() + return mgr + + +def make_options(**overrides: Any) -> ZaptecUpdateOptions: + """Build ZaptecUpdateOptions with sane defaults, overridable per test.""" + defaults: dict[str, Any] = { + "name": "test", + "update_interval": 600, + "charging_update_interval": None, + "tracked_devices": {"dev1"}, + "poll_args": {}, + "zaptec_object": None, + } + defaults.update(overrides) + return ZaptecUpdateOptions(**defaults) + + +async def test_init_sets_name_and_default_interval( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that coordinator init sets name and update interval correctly.""" + options = make_options(name="MyInstall", update_interval=300) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + assert coordinator.name == f"{DOMAIN}-myinstall" + assert coordinator.update_interval == timedelta(seconds=300) + assert coordinator.zaptec is manager.zaptec + + +async def test_init_raises_if_charging_interval_without_charger( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that charging interval requires a Charger object.""" + options = make_options( + charging_update_interval=60, + zaptec_object=MagicMock(spec=Installation), + ) + + with pytest.raises(ValueError, match="Charging update interval requires a Charger object"): + ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +async def test_init_accepts_charging_interval_with_charger( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that charging interval is accepted when a Charger object is provided.""" + charger = MagicMock(spec=Charger) + charger.is_charging.return_value = False + options = make_options(charging_update_interval=60, zaptec_object=charger) + + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + assert coordinator._charging_update_interval == timedelta(seconds=60) # noqa: SLF001 + + +async def test_set_update_interval_switches_between_charging_and_default( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that set_update_interval switches between charging and default intervals.""" + charger = MagicMock(spec=Charger) + charger.is_charging.return_value = False + charger.qual_id = "Charger[abc123]" + options = make_options( + update_interval=600, + charging_update_interval=60, + zaptec_object=charger, + ) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + assert coordinator.update_interval == timedelta(seconds=600) + + charger.is_charging.return_value = True + coordinator.set_update_interval() + assert coordinator.update_interval == timedelta(seconds=60) + + charger.is_charging.return_value = False + coordinator.set_update_interval() + assert coordinator.update_interval == timedelta(seconds=600) + + +async def test_set_update_interval_is_noop_when_unchanged( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that set_update_interval doesn't reschedule when interval is unchanged.""" + charger = MagicMock(spec=Charger) + charger.is_charging.return_value = False + options = make_options( + update_interval=600, + charging_update_interval=60, + zaptec_object=charger, + ) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + with patch.object(coordinator, "_schedule_refresh") as mock_schedule: + coordinator.set_update_interval() + mock_schedule.assert_not_called() + + +async def test_async_update_data_polls_zaptec_with_options( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that _async_update_data calls zaptec.poll with correct parameters.""" + manager.zaptec.poll = AsyncMock() + options = make_options( + tracked_devices={"dev1", "dev2"}, + poll_args={"poll_state": True}, + ) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + await coordinator._async_update_data() # noqa: SLF001 + + manager.zaptec.poll.assert_awaited_once_with({"dev1", "dev2"}, poll_state=True) + + +async def test_async_update_data_raises_update_failed_on_api_error( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that _async_update_data raises UpdateFailed on ZaptecApiError.""" + api_error = ZaptecApiError("boom") + manager.zaptec.poll = AsyncMock(side_effect=api_error) + options = make_options() + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + with pytest.raises(UpdateFailed) as exc_info: + await coordinator._async_update_data() # noqa: SLF001 + assert exc_info.value.__cause__ is api_error + + +async def test_trigger_poll_charger_uses_charger_delays( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that _trigger_poll sleeps/refreshes once per charger delay.""" + charger = MagicMock(spec=Charger) + charger.qual_id = "Charger[abc123]" + options = make_options(zaptec_object=charger) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + with ( + patch("custom_components.zaptec.coordinator.asyncio.sleep", AsyncMock()) as mock_sleep, + patch.object(coordinator, "async_refresh", AsyncMock()) as mock_refresh, + ): + await coordinator._trigger_poll(charger) # noqa: SLF001 + + assert mock_sleep.await_count == len(ZAPTEC_POLL_CHARGER_TRIGGER_DELAYS) + assert mock_refresh.await_count == len(ZAPTEC_POLL_CHARGER_TRIGGER_DELAYS) + + +async def test_trigger_poll_installation_also_triggers_tracked_children( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that _trigger_poll on an Installation also polls tracked child chargers.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + installation = MagicMock(spec=Installation) + installation.qual_id = "Installation[abc123]" + installation.chargers = [charger] + manager.tracked_devices = {"charger1"} + + child_coordinator = MagicMock() + child_coordinator.trigger_poll = AsyncMock() + manager.device_coordinators = {"charger1": child_coordinator} + + options = make_options(zaptec_object=installation) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + with ( + patch("custom_components.zaptec.coordinator.asyncio.sleep", AsyncMock()) as mock_sleep, + patch.object(coordinator, "async_refresh", AsyncMock()) as mock_refresh, + ): + await coordinator._trigger_poll(installation) # noqa: SLF001 + + assert mock_sleep.await_count == len(ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS) + assert mock_refresh.await_count == len(ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS) + child_coordinator.trigger_poll.assert_awaited_once() + + +async def test_trigger_poll_installation_skips_untracked_children( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that _trigger_poll skips children that aren't in tracked_devices.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + installation = MagicMock(spec=Installation) + installation.qual_id = "Installation[abc123]" + installation.chargers = [charger] + manager.tracked_devices = set() # charger1 is not tracked + + options = make_options(zaptec_object=installation) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + with ( + patch("custom_components.zaptec.coordinator.asyncio.sleep", AsyncMock()), + patch.object(coordinator, "async_refresh", AsyncMock()), + ): + # Would raise KeyError from manager.device_coordinators[charger.id] if + # the untracked charger were not filtered out first. + await coordinator._trigger_poll(installation) # noqa: SLF001 + + +async def test_trigger_poll_noop_without_zaptec_object( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that trigger_poll is a no-op when there is no zaptec_object.""" + options = make_options(zaptec_object=None) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + await coordinator.trigger_poll() + + assert coordinator._trigger_task is None # noqa: SLF001 + + +async def test_trigger_poll_cancels_inflight_task_before_starting_new_one( + hass: MagicMock, config_entry: Any, manager: MagicMock +) -> None: + """Test that a second trigger_poll cancels the in-flight task and starts a new one.""" + charger = MagicMock(spec=Charger) + charger.qual_id = "Charger[abc123]" + options = make_options(zaptec_object=charger) + coordinator = ZaptecUpdateCoordinator( + hass, entry=config_entry, manager=manager, options=options + ) + + call_count = 0 + first_started = asyncio.Event() + + async def fake_trigger_poll(_zaptec_obj: Any) -> None: + nonlocal call_count + call_count += 1 + if call_count == 1: + first_started.set() + await asyncio.Event().wait() # blocks forever, until cancelled + + with patch.object(coordinator, "_trigger_poll", fake_trigger_poll): + await coordinator.trigger_poll() + await first_started.wait() + first_task = coordinator._trigger_task # noqa: SLF001 + assert first_task is not None + assert not first_task.done() + + await coordinator.trigger_poll() + + assert first_task.cancelled() + # Two loop iterations are required here: the first lets the second + # task run to completion; the second lets its done-callback (which + # clears coordinator._trigger_task) actually fire, since + # Task.add_done_callback schedules callbacks via call_soon rather + # than invoking them synchronously on completion. + await asyncio.sleep(0) + await asyncio.sleep(0) # let the second task's done-callback run + assert coordinator._trigger_task is None # noqa: SLF001 + assert call_count == 2 # noqa: PLR2004 diff --git a/tests/test_entity.py b/tests/test_entity.py new file mode 100644 index 00000000..15542069 --- /dev/null +++ b/tests/test_entity.py @@ -0,0 +1,310 @@ +"""Tests for entity.py.""" + +from __future__ import annotations + +import logging +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from homeassistant.helpers.entity import DeviceInfo, EntityDescription +import pytest + +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.entity import KeyUnavailableError, ZaptecBaseEntity +from custom_components.zaptec.zaptec import MISSING + + +class FakeZaptecObj: + """Minimal stand-in for a ZaptecBase object, exposing only what ZaptecBaseEntity uses.""" + + def __init__(self, obj_id: str, data: dict[str, Any]) -> None: + """Initialize the FakeZaptecObj.""" + self.id = obj_id + self._data = data + + @property + def qual_id(self) -> str: + """Return the qualified id.""" + return f"Fake[{self.id}]" + + def get(self, key: str, default: Any = MISSING) -> Any: + """Get a value from the data dict.""" + return self._data.get(key, default) + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +@pytest.fixture +def zaptec_obj() -> FakeZaptecObj: + """Create a FakeZaptecObj for testing.""" + return FakeZaptecObj( + "dev1", + {"operating_mode": "Connected", "nested": {"inner": "value"}}, + ) + + +@pytest.fixture +def entity(coordinator: ZaptecUpdateCoordinator, zaptec_obj: FakeZaptecObj) -> ZaptecBaseEntity: + """Create a ZaptecBaseEntity for testing.""" + description = EntityDescription(key="operating_mode") + return ZaptecBaseEntity(coordinator, zaptec_obj, description, DeviceInfo()) + + +def test_init_sets_unique_id_device_info_and_log_key( + entity: ZaptecBaseEntity, zaptec_obj: FakeZaptecObj +) -> None: + """Test that init sets _attr_unique_id, _attr_device_info, and _log_zaptec_key.""" + assert entity._attr_unique_id == "dev1_operating_mode" # noqa: SLF001 + assert entity._attr_device_info == DeviceInfo() # noqa: SLF001 + assert entity._log_zaptec_key == "operating_mode" # noqa: SLF001 + + +def test_key_property_returns_description_key(entity: ZaptecBaseEntity) -> None: + """Test that key property returns the entity description key.""" + assert entity.key == "operating_mode" + + +def test_get_zaptec_value_returns_value(entity: ZaptecBaseEntity) -> None: + """Test that _get_zaptec_value returns the value from zaptec_obj.""" + assert entity._get_zaptec_value() == "Connected" # noqa: SLF001 + + +def test_get_zaptec_value_lower_cases_string(entity: ZaptecBaseEntity) -> None: + """Test that _get_zaptec_value lower cases strings when requested.""" + assert entity._get_zaptec_value(lower_case_str=True) == "connected" # noqa: SLF001 + + +def test_get_zaptec_value_follows_dotted_key(entity: ZaptecBaseEntity) -> None: + """Test that _get_zaptec_value follows dotted keys.""" + assert entity._get_zaptec_value(key="nested.inner") == "value" # noqa: SLF001 + + +def test_get_zaptec_value_returns_default_without_raising(entity: ZaptecBaseEntity) -> None: + """Test that _get_zaptec_value returns default when key is missing.""" + assert entity._get_zaptec_value(key="missing_key", default="fallback") == "fallback" # noqa: SLF001 + + +def test_get_zaptec_value_raises_when_key_missing(entity: ZaptecBaseEntity) -> None: + """Test that _get_zaptec_value raises KeyUnavailableError for missing keys.""" + with pytest.raises(KeyUnavailableError) as exc_info: + entity._get_zaptec_value(key="missing_key") # noqa: SLF001 + assert exc_info.value.key == "missing_key" + + +def test_get_zaptec_value_raises_when_object_is_not_a_mapping(entity: ZaptecBaseEntity) -> None: + """Test that _get_zaptec_value raises KeyUnavailableError when zaptec_obj is not a mapping.""" + + class NotMapping: + @property + def qual_id(self) -> str: + """Return a fake qualified id.""" + return "NotMapping[test]" + + entity.zaptec_obj = NotMapping() + + with pytest.raises(KeyUnavailableError): + entity._get_zaptec_value(key="operating_mode") # noqa: SLF001 + + +def test_handle_coordinator_update_success_updates_value_and_writes_state( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that a successful update logs the new value and writes HA state.""" + entity.entity_id = "sensor.test" + entity.async_write_ha_state = MagicMock() + entity._log_attribute = "some_attr" # noqa: SLF001 + entity.some_attr = "new_value" + entity._update_from_zaptec = lambda: None # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._handle_coordinator_update() # noqa: SLF001 + + entity.async_write_ha_state.assert_called_once() + assert "new_value" in caplog.text + + +def test_handle_coordinator_update_key_unavailable_sets_attr_available_false( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that a KeyUnavailableError during update marks the entity unavailable.""" + entity.entity_id = "sensor.test" + entity.async_write_ha_state = MagicMock() + + def raise_unavailable() -> None: + raise KeyUnavailableError("operating_mode", "boom") + + entity._update_from_zaptec = raise_unavailable # noqa: SLF001 + + with caplog.at_level(logging.INFO): + entity._handle_coordinator_update() # noqa: SLF001 + + # NOTE: this sets _attr_available, but ZaptecBaseEntity does not override + # the `available` property inherited from HA's CoordinatorEntity (which + # returns coordinator.last_update_success instead), so this flag currently + # has no effect on the entity's actual reported availability. This test + # documents today's real behavior, not the intended one - see the "Known + # finding" note at the top of this plan. + assert entity._attr_available is False # noqa: SLF001 + assert "sensor.test is unavailable" in caplog.text + entity.async_write_ha_state.assert_called_once() + + +def test_log_zaptec_attribute_formats_string_key(entity: ZaptecBaseEntity) -> None: + """Test that _log_zaptec_attribute formats a string key with a leading dot.""" + entity._log_zaptec_key = "operating_mode" # noqa: SLF001 + assert entity._log_zaptec_attribute == ".operating_mode" # noqa: SLF001 + + +def test_log_zaptec_attribute_formats_none_key(entity: ZaptecBaseEntity) -> None: + """Test that _log_zaptec_attribute returns an empty string when the key is None.""" + entity._log_zaptec_key = None # noqa: SLF001 + assert entity._log_zaptec_attribute == "" # noqa: SLF001 + + +def test_log_zaptec_attribute_formats_iterable_key(entity: ZaptecBaseEntity) -> None: + """Test that _log_zaptec_attribute joins iterable keys with 'and'.""" + entity._log_zaptec_key = ["mode", "state"] # noqa: SLF001 + assert entity._log_zaptec_attribute == ".mode and .state" # noqa: SLF001 + + +def test_log_value_logs_on_change( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_value logs when the value has changed.""" + entity.entity_id = "sensor.test" + entity.some_attr = "value1" + + with caplog.at_level(logging.DEBUG): + entity._log_value("some_attr") # noqa: SLF001 + + assert "value1" in caplog.text + assert entity._prev_value == "value1" # noqa: SLF001 + + +def test_log_value_skips_logging_when_unchanged( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_value skips logging when the value is unchanged.""" + entity.entity_id = "sensor.test" + entity.some_attr = "value1" + entity._prev_value = "value1" # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._log_value("some_attr") # noqa: SLF001 + + assert caplog.text == "" + + +def test_log_value_force_logs_even_when_unchanged( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_value logs even when unchanged if force is True.""" + entity.entity_id = "sensor.test" + entity.some_attr = "value1" + entity._prev_value = "value1" # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._log_value("some_attr", force=True) # noqa: SLF001 + + assert "value1" in caplog.text + + +def test_log_value_noop_for_none_attribute( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_value is a no-op when the attribute is None.""" + with caplog.at_level(logging.DEBUG): + entity._log_value(None) # noqa: SLF001 + + assert caplog.text == "" + + +def test_log_unavailable_logs_on_transition_to_unavailable( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_unavailable logs when the entity transitions to unavailable.""" + entity.entity_id = "sensor.test" + entity._attr_available = False # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._log_unavailable() # noqa: SLF001 + + assert "Entity sensor.test is unavailable" in caplog.text + + +def test_log_unavailable_logs_error_for_unexpected_exception( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_unavailable logs an error for an unexpected exception.""" + entity.entity_id = "sensor.test" + entity._attr_available = False # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._log_unavailable(exception=ValueError("boom")) # noqa: SLF001 + + assert "Getting value failed" in caplog.text + + +def test_log_unavailable_skips_error_for_key_unavailable_error( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_unavailable skips the error log for KeyUnavailableError.""" + entity.entity_id = "sensor.test" + entity._attr_available = False # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._log_unavailable(exception=KeyUnavailableError("some_key", "boom")) # noqa: SLF001 + + assert "Getting value failed" not in caplog.text + + +def test_log_unavailable_skips_error_for_keys_in_skip_set( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_unavailable skips the error log for keys in the skip set.""" + entity.entity_id = "sensor.test" + entity.entity_description = EntityDescription(key="three_to_one_phase_switch_current") + entity._attr_available = False # noqa: SLF001 + + with caplog.at_level(logging.DEBUG): + entity._log_unavailable(exception=ValueError("boom")) # noqa: SLF001 + + assert "Getting value failed" not in caplog.text + + +def test_log_unavailable_logs_on_recovery( + entity: ZaptecBaseEntity, caplog: pytest.LogCaptureFixture +) -> None: + """Test that _log_unavailable logs when the entity recovers to available.""" + entity.entity_id = "sensor.test" + entity._prev_available = False # noqa: SLF001 + entity._attr_available = True # noqa: SLF001 + + with caplog.at_level(logging.INFO): + entity._log_unavailable() # noqa: SLF001 + + assert "Entity sensor.test is available" in caplog.text + + +async def test_trigger_poll_delegates_to_coordinator( + entity: ZaptecBaseEntity, coordinator: ZaptecUpdateCoordinator +) -> None: + """Test that trigger_poll delegates to coordinator.trigger_poll.""" + coordinator.trigger_poll = AsyncMock() + + await entity.trigger_poll() + + coordinator.trigger_poll.assert_awaited_once() diff --git a/tests/test_init.py b/tests/test_init.py new file mode 100644 index 00000000..a4ac16ec --- /dev/null +++ b/tests/test_init.py @@ -0,0 +1,267 @@ +"""Tests for custom_components.zaptec.__init__.""" + +from collections.abc import Generator +from http import HTTPStatus +from unittest.mock import MagicMock, patch + +from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryError, ConfigEntryNotReady +import pytest + +from custom_components.zaptec import ( + _cleanup_stale_devices, + _config_entry_error, + _should_check_untracked, +) +from custom_components.zaptec.zaptec.exceptions import ( + AuthenticationError, + RequestConnectionError, + RequestError, + RequestTimeoutError, +) + + +@pytest.mark.parametrize( + ("err", "expected"), + [ + # Bad credentials are non-recoverable -> re-auth flow. + (AuthenticationError("bad"), ConfigEntryAuthFailed), + # Connection/timeout are recoverable -> HA retries setup. + (RequestTimeoutError("slow"), ConfigEntryNotReady), + (RequestConnectionError("down"), ConfigEntryNotReady), + # Transient server statuses are recoverable -> HA retries setup (issue #392). + (RequestError("unavailable", HTTPStatus.SERVICE_UNAVAILABLE), ConfigEntryNotReady), + (RequestError("too many", HTTPStatus.TOO_MANY_REQUESTS), ConfigEntryNotReady), + # Other HTTP errors stay permanent. + (RequestError("forbidden", HTTPStatus.FORBIDDEN), ConfigEntryError), + (RequestError("not found", HTTPStatus.NOT_FOUND), ConfigEntryError), + ], +) +def test_config_entry_error_mapping(err: Exception, expected: type[Exception]) -> None: + """Setup login errors map to the right Home Assistant config-entry error.""" + assert isinstance(_config_entry_error(err), expected) + + +def _mock_registries( + device_entries: list, entities_by_device: dict +) -> tuple[MagicMock, MagicMock]: + """Patch dr.async_get/er.async_get and the two registry lookup helpers. + + Returns (device_registry_mock, entity_registry_mock) so callers can + assert on async_remove_device/async_remove calls. + """ + device_registry = MagicMock() + entity_registry = MagicMock() + + def entries_for_device( + _entity_registry: MagicMock, device_id: str, include_disabled_entities: bool = True + ) -> list: + return entities_by_device.get(device_id, []) + + patch("custom_components.zaptec.dr.async_get", return_value=device_registry).start() + patch("custom_components.zaptec.er.async_get", return_value=entity_registry).start() + patch( + "custom_components.zaptec.dr.async_entries_for_config_entry", + return_value=device_entries, + ).start() + patch( + "custom_components.zaptec.er.async_entries_for_device", + side_effect=entries_for_device, + ).start() + + return device_registry, entity_registry + + +@pytest.fixture(autouse=True) +def _stop_patches() -> Generator[None]: + """Undo any patch.start() calls made via _mock_registries after each test.""" + yield + patch.stopall() + + +def _device(device_id: str, zaptec_id: str) -> MagicMock: + dev = MagicMock() + dev.id = device_id + dev.identifiers = {("zaptec", zaptec_id)} + return dev + + +def _entity(entity_id: str) -> MagicMock: + ent = MagicMock() + ent.entity_id = entity_id + return ent + + +def test_cleanup_removes_device_with_no_entities() -> None: + """A device with zero registered entities is removed outright.""" + empty_device = _device("dev-empty", "charger-empty") + device_registry, entity_registry = _mock_registries( + device_entries=[empty_device], + entities_by_device={"dev-empty": []}, + ) + + _cleanup_stale_devices( + MagicMock(), + MagicMock(entry_id="entry1"), + tracked_devices={"charger-empty"}, # tracked, but still has no entities + circuit_ids=set(), + check_untracked=True, + ) + + device_registry.async_remove_device.assert_called_once_with("dev-empty") + entity_registry.async_remove.assert_not_called() + + +def test_cleanup_removes_deprecated_circuit_device() -> None: + """A device matching a known Circuit id is removed along with its entities.""" + circuit_device = _device("dev-circuit", "circuit-123") + circuit_entity = _entity("sensor.circuit_123_power") + device_registry, entity_registry = _mock_registries( + device_entries=[circuit_device], + entities_by_device={"dev-circuit": [circuit_entity]}, + ) + + _cleanup_stale_devices( + MagicMock(), + MagicMock(entry_id="entry1"), + tracked_devices=set(), + circuit_ids={"circuit-123"}, + check_untracked=True, + ) + + entity_registry.async_remove.assert_called_once_with("sensor.circuit_123_power") + device_registry.async_remove_device.assert_called_once_with("dev-circuit") + + +def test_cleanup_keeps_tracked_device_with_entities() -> None: + """A device that is tracked and has entities is left alone.""" + kept_device = _device("dev-kept", "charger-kept") + kept_entity = _entity("sensor.kept_charger_power") + device_registry, entity_registry = _mock_registries( + device_entries=[kept_device], + entities_by_device={"dev-kept": [kept_entity]}, + ) + + _cleanup_stale_devices( + MagicMock(), + MagicMock(entry_id="entry1"), + tracked_devices={"charger-kept"}, + circuit_ids=set(), + check_untracked=True, + ) + + device_registry.async_remove_device.assert_not_called() + entity_registry.async_remove.assert_not_called() + + +def test_cleanup_removes_deselected_charger_device() -> None: + """Issue #272: a charger deselected via reconfigure loses its device and entities. + + Before this fix, `tracked_devices` correctly excludes the deselected + charger (so no new entities are created for it, per the #274/#275 fix), + but its old entity-registry entries from the prior session were never + explicitly removed, so `dev_entities` stayed non-empty and the device + was never cleaned up. + """ + stale_device = _device("dev-stale", "charger-stale") + kept_device = _device("dev-kept", "charger-kept") + stale_entity = _entity("sensor.stale_charger_power") + kept_entity = _entity("sensor.kept_charger_power") + + device_registry, entity_registry = _mock_registries( + device_entries=[stale_device, kept_device], + entities_by_device={ + "dev-stale": [stale_entity], + "dev-kept": [kept_entity], + }, + ) + + _cleanup_stale_devices( + MagicMock(), + MagicMock(entry_id="entry1"), + tracked_devices={"charger-kept"}, # charger-stale was deselected + circuit_ids=set(), + check_untracked=True, + ) + + entity_registry.async_remove.assert_called_once_with("sensor.stale_charger_power") + device_registry.async_remove_device.assert_called_once_with("dev-stale") + + +def test_cleanup_skips_untracked_removal_when_selection_incomplete() -> None: + """A device not in tracked_devices is left alone when check_untracked is False. + + This guards against a real risk found in review: if the Zaptec API returns + a partial account this session (outage, transient blip), a charger the + user genuinely still has selected can silently drop out of + tracked_devices. Without this guard, that transient blip would be + mistaken for the user deselecting the charger and its device+entities + would be permanently deleted. + """ + maybe_stale_device = _device("dev-maybe-stale", "charger-maybe-stale") + maybe_stale_entity = _entity("sensor.maybe_stale_charger_power") + + device_registry, entity_registry = _mock_registries( + device_entries=[maybe_stale_device], + entities_by_device={"dev-maybe-stale": [maybe_stale_entity]}, + ) + + _cleanup_stale_devices( + MagicMock(), + MagicMock(entry_id="entry1"), + tracked_devices=set(), # would look untracked... + circuit_ids=set(), + check_untracked=False, # ...but the API response was incomplete this session + ) + + device_registry.async_remove_device.assert_not_called() + entity_registry.async_remove.assert_not_called() + + +def test_cleanup_keeps_tracked_installation_device() -> None: + """An installation device (not a charger) with a tracked id is left alone. + + tracked_devices holds both charger ids and their installation ids + (ZaptecManager.first_time_setup keeps an installation whenever any of its + chargers is selected) - this documents that installations go through the + same tracked-device check as chargers. + """ + installation_device = _device("dev-installation", "installation-1") + installation_entity = _entity("sensor.installation_1_power") + + device_registry, entity_registry = _mock_registries( + device_entries=[installation_device], + entities_by_device={"dev-installation": [installation_entity]}, + ) + + _cleanup_stale_devices( + MagicMock(), + MagicMock(entry_id="entry1"), + tracked_devices={"installation-1"}, + circuit_ids=set(), + check_untracked=True, + ) + + device_registry.async_remove_device.assert_not_called() + entity_registry.async_remove.assert_not_called() + + +@pytest.mark.parametrize( + ("configured_chargers", "all_selected_present", "expected"), + [ + # Track-all mode: never check untracked, regardless of all_selected_present + # (which is always True in this mode anyway - included for defensiveness). + (None, True, False), + (None, False, False), + # Manual-select mode, every selected charger confirmed present -> safe to check. + ({"charger-a"}, True, True), + # Manual-select mode, API response was partial this session -> must not check. + ({"charger-a"}, False, False), + ], +) +def test_should_check_untracked( + configured_chargers: set[str] | None, + all_selected_present: bool, + expected: bool, +) -> None: + """Untracked-device removal is only ever safe in confirmed manual-select mode.""" + assert _should_check_untracked(configured_chargers, all_selected_present) is expected diff --git a/tests/test_number.py b/tests/test_number.py new file mode 100644 index 00000000..21b14de1 --- /dev/null +++ b/tests/test_number.py @@ -0,0 +1,284 @@ +"""Tests for number.py.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity import DeviceInfo +import pytest + +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.number import ( + ZapNumberEntityDescription, + ZaptecAvailableCurrentNumber, + ZaptecHmiBrightness, + ZaptecNumber, + ZaptecSettingNumber, + ZaptecThreeToOnePhaseSwitchCurrent, +) +from custom_components.zaptec.zaptec import Charger, Installation + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +def make_charger(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Charger) whose .get() reads from data.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + charger.qual_id = "Charger[charger1]" + charger.get.side_effect = data.get + return charger + + +def make_installation(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Installation) whose .get() reads from data.""" + installation = MagicMock(spec=Installation) + installation.id = "install1" + installation.qual_id = "Installation[install1]" + installation.get.side_effect = data.get + return installation + + +def test_number_update_from_zaptec_sets_value(coordinator: ZaptecUpdateCoordinator) -> None: + """ZaptecNumber._update_from_zaptec reads the raw value for its key.""" + installation = make_installation({"available_current": 16.0}) + description = ZapNumberEntityDescription(key="available_current", cls=ZaptecNumber) + entity = ZaptecNumber(coordinator, installation, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == 16.0 # noqa: SLF001, PLR2004 + assert entity._attr_available is True # noqa: SLF001 + + +def test_available_current_post_init_uses_reported_max_current( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecAvailableCurrentNumber._post_init sets native_max_value from MaxCurrent.""" + installation = make_installation({"MaxCurrent": 20}) + description = ZapNumberEntityDescription( + key="available_current", native_max_value=0, cls=ZaptecAvailableCurrentNumber + ) + entity = ZaptecAvailableCurrentNumber(coordinator, installation, description, DeviceInfo()) + + assert entity.entity_description.native_max_value == 20 # noqa: PLR2004 + + +def test_available_current_post_init_defaults_to_32( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecAvailableCurrentNumber._post_init defaults to 32A when MaxCurrent is absent.""" + installation = make_installation({}) + description = ZapNumberEntityDescription( + key="available_current", native_max_value=0, cls=ZaptecAvailableCurrentNumber + ) + entity = ZaptecAvailableCurrentNumber(coordinator, installation, description, DeviceInfo()) + + assert entity.entity_description.native_max_value == 32 # noqa: PLR2004 + + +async def test_available_current_set_native_value_success( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value sets the current limit and triggers a poll on success.""" + installation = make_installation({}) + installation.set_limit_current = AsyncMock() + description = ZapNumberEntityDescription( + key="available_current", cls=ZaptecAvailableCurrentNumber + ) + entity = ZaptecAvailableCurrentNumber(coordinator, installation, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_set_native_value(10.0) + + installation.set_limit_current.assert_awaited_once_with(availableCurrent=10.0) + entity.trigger_poll.assert_awaited_once() + + +async def test_available_current_set_native_value_wraps_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value wraps a failure in HomeAssistantError and skips the poll.""" + installation = make_installation({}) + installation.set_limit_current = AsyncMock(side_effect=Exception("boom")) + description = ZapNumberEntityDescription( + key="available_current", cls=ZaptecAvailableCurrentNumber + ) + entity = ZaptecAvailableCurrentNumber(coordinator, installation, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_set_native_value(10.0) + + entity.trigger_poll.assert_not_called() + + +async def test_three_to_one_phase_set_native_value_success( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value sets the switch current and triggers a poll on success.""" + installation = make_installation({}) + installation.set_three_to_one_phase_switch_current = AsyncMock() + description = ZapNumberEntityDescription( + key="three_to_one_phase_switch_current", cls=ZaptecThreeToOnePhaseSwitchCurrent + ) + entity = ZaptecThreeToOnePhaseSwitchCurrent( + coordinator, installation, description, DeviceInfo() + ) + entity.trigger_poll = AsyncMock() + + await entity.async_set_native_value(8.0) + + installation.set_three_to_one_phase_switch_current.assert_awaited_once_with(8.0) + entity.trigger_poll.assert_awaited_once() + + +async def test_three_to_one_phase_set_native_value_wraps_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value wraps a failure in HomeAssistantError and skips the poll.""" + installation = make_installation({}) + installation.set_three_to_one_phase_switch_current = AsyncMock(side_effect=Exception("boom")) + description = ZapNumberEntityDescription( + key="three_to_one_phase_switch_current", cls=ZaptecThreeToOnePhaseSwitchCurrent + ) + entity = ZaptecThreeToOnePhaseSwitchCurrent( + coordinator, installation, description, DeviceInfo() + ) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_set_native_value(8.0) + + entity.trigger_poll.assert_not_called() + + +def test_setting_number_post_init_uses_reported_max_limit( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecSettingNumber._post_init sets native_max_value from ChargeCurrentInstallationMaxLimit.""" + charger = make_charger({"ChargeCurrentInstallationMaxLimit": 25}) + description = ZapNumberEntityDescription( + key="charger_max_current", + native_max_value=0, + setting="maxChargeCurrent", + cls=ZaptecSettingNumber, + ) + entity = ZaptecSettingNumber(coordinator, charger, description, DeviceInfo()) + + assert entity.entity_description.native_max_value == 25 # noqa: PLR2004 + + +async def test_setting_number_missing_setting_raises_without_calling_api( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value raises HomeAssistantError when no setting is configured.""" + charger = make_charger({}) + charger.set_settings = AsyncMock() + description = ZapNumberEntityDescription( + key="charger_max_current", setting=None, cls=ZaptecSettingNumber + ) + entity = ZaptecSettingNumber(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_set_native_value(16.0) + + charger.set_settings.assert_not_called() + entity.trigger_poll.assert_not_called() + + +async def test_setting_number_set_native_value_success( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value writes the configured setting and triggers a poll on success.""" + charger = make_charger({}) + charger.set_settings = AsyncMock() + description = ZapNumberEntityDescription( + key="charger_max_current", setting="maxChargeCurrent", cls=ZaptecSettingNumber + ) + entity = ZaptecSettingNumber(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_set_native_value(16.0) + + charger.set_settings.assert_awaited_once_with({"maxChargeCurrent": 16.0}) + entity.trigger_poll.assert_awaited_once() + + +async def test_setting_number_set_native_value_wraps_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value wraps a failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.set_settings = AsyncMock(side_effect=Exception("boom")) + description = ZapNumberEntityDescription( + key="charger_max_current", setting="maxChargeCurrent", cls=ZaptecSettingNumber + ) + entity = ZaptecSettingNumber(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_set_native_value(16.0) + + entity.trigger_poll.assert_not_called() + + +def test_hmi_brightness_update_from_zaptec_scales_up( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecHmiBrightness._update_from_zaptec scales the 0-1 API value to a 0-100 percentage.""" + charger = make_charger({"hmi_brightness": 0.55}) + description = ZapNumberEntityDescription(key="hmi_brightness", cls=ZaptecHmiBrightness) + entity = ZaptecHmiBrightness(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == pytest.approx(55.0) # noqa: SLF001 + + +async def test_hmi_brightness_set_native_value_scales_down( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value scales the 0-100 percentage back to 0-1 and triggers a poll.""" + charger = make_charger({}) + charger.set_hmi_brightness = AsyncMock() + description = ZapNumberEntityDescription(key="hmi_brightness", cls=ZaptecHmiBrightness) + entity = ZaptecHmiBrightness(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_set_native_value(50.0) + + charger.set_hmi_brightness.assert_awaited_once_with(0.5) + entity.trigger_poll.assert_awaited_once() + + +async def test_hmi_brightness_set_native_value_wraps_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_set_native_value wraps a failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.set_hmi_brightness = AsyncMock(side_effect=Exception("boom")) + description = ZapNumberEntityDescription(key="hmi_brightness", cls=ZaptecHmiBrightness) + entity = ZaptecHmiBrightness(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_set_native_value(50.0) + + entity.trigger_poll.assert_not_called() diff --git a/tests/test_sensor.py b/tests/test_sensor.py new file mode 100644 index 00000000..6126de08 --- /dev/null +++ b/tests/test_sensor.py @@ -0,0 +1,159 @@ +"""Tests for sensor.py.""" + +from __future__ import annotations + +import logging +from typing import Any +from unittest.mock import MagicMock + +from homeassistant.helpers.entity import DeviceInfo +import pytest + +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.sensor import ( + ZapSensorEntityDescription, + ZaptecChargeSensor, + ZaptecEnengySensor, + ZaptecSensor, + ZaptecSensorTranslate, +) +from custom_components.zaptec.zaptec import Charger + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +def make_charger(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Charger) whose .get() reads from data.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + charger.qual_id = "Charger[charger1]" + charger.get.side_effect = data.get + return charger + + +def test_sensor_update_from_zaptec_sets_value(coordinator: ZaptecUpdateCoordinator) -> None: + """ZaptecSensor._update_from_zaptec reads the raw value for its key.""" + charger = make_charger({"total_charge_power": 1500.0}) + description = ZapSensorEntityDescription(key="total_charge_power", cls=ZaptecSensor) + entity = ZaptecSensor(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == 1500.0 # noqa: SLF001, PLR2004 + assert entity._attr_available is True # noqa: SLF001 + + +def test_sensor_translate_post_init_lower_cases_options( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecSensorTranslate._post_init lower-cases entity_description.options.""" + charger = make_charger({"device_type": "PRO"}) + description = ZapSensorEntityDescription( + key="device_type", options=["Pro", "GO"], cls=ZaptecSensorTranslate + ) + entity = ZaptecSensorTranslate(coordinator, charger, description, DeviceInfo()) + + assert entity.entity_description.options == ["pro", "go"] + + +def test_sensor_translate_update_from_zaptec_lower_cases_value( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecSensorTranslate._update_from_zaptec lower-cases the retrieved value.""" + charger = make_charger({"device_type": "PRO"}) + description = ZapSensorEntityDescription( + key="device_type", options=["pro"], cls=ZaptecSensorTranslate + ) + entity = ZaptecSensorTranslate(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == "pro" # noqa: SLF001 + assert entity._attr_available is True # noqa: SLF001 + + +def test_charge_sensor_maps_known_mode_to_icon(coordinator: ZaptecUpdateCoordinator) -> None: + """ZaptecChargeSensor picks the icon matching a known charger_operation_mode.""" + charger = make_charger({"charger_operation_mode": "Connected_Charging"}) + description = ZapSensorEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSensor) + entity = ZaptecChargeSensor(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == "connected_charging" # noqa: SLF001 + assert entity._attr_icon == "mdi:lightning-bolt" # noqa: SLF001 + + +def test_charge_sensor_falls_back_to_unknown_icon(coordinator: ZaptecUpdateCoordinator) -> None: + """ZaptecChargeSensor falls back to the 'unknown' icon for an unmapped mode.""" + charger = make_charger({"charger_operation_mode": "Something_Weird"}) + description = ZapSensorEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSensor) + entity = ZaptecChargeSensor(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_icon == "mdi:help-rhombus-outline" # noqa: SLF001 + + +def test_energy_sensor_uses_meter_value_when_no_session( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecEnengySensor falls back to the meter reading when no session is present.""" + charger = make_charger({"signed_meter_value": {"RD": [{"RV": 12.5}]}}) + description = ZapSensorEntityDescription(key="signed_meter_value_kwh", cls=ZaptecEnengySensor) + entity = ZaptecEnengySensor(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == 12.5 # noqa: SLF001, PLR2004 + + +def test_energy_sensor_uses_session_value_when_larger( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecEnengySensor uses the session reading when it exceeds the meter reading.""" + charger = make_charger( + { + "signed_meter_value": {"RD": [{"RV": 10.0}]}, + "completed_session": {"SignedSession": {"RD": [{"RV": 20.0}]}}, + } + ) + description = ZapSensorEntityDescription(key="signed_meter_value_kwh", cls=ZaptecEnengySensor) + entity = ZaptecEnengySensor(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == 20.0 # noqa: SLF001, PLR2004 + + +def test_energy_sensor_ignores_non_dict_session( + coordinator: ZaptecUpdateCoordinator, caplog: pytest.LogCaptureFixture +) -> None: + """ZaptecEnengySensor logs and defaults the session reading to 0.0 when it isn't a dict.""" + charger = make_charger( + { + "signed_meter_value": {"RD": [{"RV": 10.0}]}, + "completed_session": "not-a-dict", + } + ) + description = ZapSensorEntityDescription(key="signed_meter_value_kwh", cls=ZaptecEnengySensor) + entity = ZaptecEnengySensor(coordinator, charger, description, DeviceInfo()) + + with caplog.at_level(logging.DEBUG): + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_native_value == 10.0 # noqa: SLF001, PLR2004 + assert "Incorrect typing for completed_session" in caplog.text diff --git a/tests/test_services.py b/tests/test_services.py new file mode 100644 index 00000000..7e6d7f5a --- /dev/null +++ b/tests/test_services.py @@ -0,0 +1,685 @@ +"""Tests for services.py.""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from homeassistant.core import ServiceCall +from homeassistant.exceptions import HomeAssistantError +import pytest +import voluptuous as vol +import yaml + +from custom_components.zaptec.const import DOMAIN +import custom_components.zaptec.services as services_module +from custom_components.zaptec.services import ( + CHARGER_ID_SCHEMA, + LIMIT_CURRENT_SCHEMA, + SEND_COMMAND_SCHEMA, + async_setup_services, + async_unload_services, +) +from custom_components.zaptec.zaptec import Charger, Installation + +SERVICES_YAML_PATH = Path(services_module.__file__).with_name("services.yaml") + + +def make_call(hass: MagicMock, data: dict[str, Any]) -> ServiceCall: + """Build a ServiceCall carrying the given data, bypassing schema validation.""" + return ServiceCall(hass, DOMAIN, "test_service", data) + + +def make_charger(uid: str = "charger1") -> MagicMock: + """Create a MagicMock(spec=Charger) with async command methods.""" + charger = MagicMock(spec=Charger) + charger.id = uid + charger.command = AsyncMock() + charger.authorize_charge = AsyncMock() + return charger + + +def make_installation(uid: str = "install1") -> MagicMock: + """Create a MagicMock(spec=Installation) with an async set_limit_current.""" + installation = MagicMock(spec=Installation) + installation.id = uid + installation.set_limit_current = AsyncMock() + return installation + + +@pytest.fixture +def manager() -> MagicMock: + """A manager stub exposing plain dicts for `.zaptec` and `.device_coordinators`.""" + mgr = MagicMock() + mgr.zaptec = {} + mgr.device_coordinators = {} + return mgr + + +@pytest.fixture +def fake_registries() -> SimpleNamespace: + """Patch er.async_get/dr.async_get with dict-backed fakes and expose the dicts.""" + entities: dict[str, Any] = {} + devices: dict[str, Any] = {} + + ent_reg = MagicMock() + ent_reg.async_get.side_effect = entities.get + + dev_reg = MagicMock() + dev_reg.async_get.side_effect = devices.get + + with ( + patch("custom_components.zaptec.services.er.async_get", return_value=ent_reg), + patch("custom_components.zaptec.services.dr.async_get", return_value=dev_reg), + ): + yield SimpleNamespace(entities=entities, devices=devices) + + +@pytest.fixture +def add_charger(manager: MagicMock) -> Any: + """Register a charger + coordinator pair under a given uid in the manager stubs.""" + + def _add(uid: str = "charger1") -> tuple[MagicMock, MagicMock]: + charger = make_charger(uid) + coordinator = MagicMock() + coordinator.trigger_poll = AsyncMock() + manager.zaptec[uid] = charger + manager.device_coordinators[uid] = coordinator + return charger, coordinator + + return _add + + +@pytest.fixture +def add_installation(manager: MagicMock) -> Any: + """Register an installation + coordinator pair under a given uid in the manager stubs.""" + + def _add(uid: str = "install1") -> tuple[MagicMock, MagicMock]: + installation = make_installation(uid) + coordinator = MagicMock() + coordinator.trigger_poll = AsyncMock() + manager.zaptec[uid] = installation + manager.device_coordinators[uid] = coordinator + return installation, coordinator + + return _add + + +@pytest.fixture +async def handlers(hass: MagicMock, manager: MagicMock) -> dict[str, Any]: + """Register zaptec services and return {name: handler} for direct invocation.""" + hass.services.has_service = MagicMock(return_value=False) + await async_setup_services(hass, manager) + return {call.args[1]: call.args[2] for call in hass.services.async_register.call_args_list} + + +# --------------------------------------------------------------------------- +# async_setup_services / async_unload_services +# --------------------------------------------------------------------------- + + +async def test_async_setup_services_registers_all_services(hass: MagicMock) -> None: + """All eight zaptec services get registered under the zaptec domain.""" + manager = MagicMock() + hass.services.has_service = MagicMock(return_value=False) + + await async_setup_services(hass, manager) + + registered = {call.args[1] for call in hass.services.async_register.call_args_list} + assert registered == { + "stop_charging", + "resume_charging", + "authorize_charging", + "deauthorize_charging", + "restart_charger", + "upgrade_firmware", + "limit_current", + "send_command", + } + assert all(call.args[0] == DOMAIN for call in hass.services.async_register.call_args_list) + + +async def test_async_setup_services_skips_already_registered(hass: MagicMock) -> None: + """A service that has_service reports as already present is not re-registered.""" + manager = MagicMock() + hass.services.has_service = MagicMock( + side_effect=lambda _domain, name: name == "stop_charging" + ) + + await async_setup_services(hass, manager) + + registered = {call.args[1] for call in hass.services.async_register.call_args_list} + assert "stop_charging" not in registered + assert "resume_charging" in registered + + +async def test_async_unload_services_removes_all_domain_services(hass: MagicMock) -> None: + """All services under the zaptec domain get removed.""" + hass.services.async_services.return_value = { + DOMAIN: {"stop_charging": None, "limit_current": None}, + "other_domain": {"foo": None}, + } + + await async_unload_services(hass) + + assert hass.services.async_remove.call_count == 2 # noqa: PLR2004 + removed = {call.args[1] for call in hass.services.async_remove.call_args_list} + assert removed == {"stop_charging", "limit_current"} + assert all(call.args[0] == DOMAIN for call in hass.services.async_remove.call_args_list) + + +# --------------------------------------------------------------------------- +# iter_objects resolution / error paths (exercised through stop_charging) +# --------------------------------------------------------------------------- + + +async def test_resolves_via_legacy_charger_id( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A bare charger_id resolves directly to the zaptec object.""" + charger, coordinator = add_charger("charger1") + + await handlers["stop_charging"](make_call(hass, {"charger_id": "charger1"})) + + charger.command.assert_awaited_once_with("stop_charging_final") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_resolves_via_device_id( + hass: MagicMock, + manager: MagicMock, + add_charger: Any, + handlers: dict[str, Any], + fake_registries: SimpleNamespace, +) -> None: + """A device_id resolves through the device registry's zaptec identifier.""" + charger, coordinator = add_charger("charger1") + fake_registries.devices["device1"] = SimpleNamespace( + identifiers={(DOMAIN, "charger1")}, name="Device 1" + ) + + await handlers["stop_charging"](make_call(hass, {"device_id": "device1"})) + + charger.command.assert_awaited_once_with("stop_charging_final") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_resolves_via_entity_id( + hass: MagicMock, + manager: MagicMock, + add_charger: Any, + handlers: dict[str, Any], + fake_registries: SimpleNamespace, +) -> None: + """An entity_id resolves through the entity registry's device, then the device registry.""" + charger, coordinator = add_charger("charger1") + fake_registries.entities["sensor.foo"] = SimpleNamespace(device_id="device1") + fake_registries.devices["device1"] = SimpleNamespace( + identifiers={(DOMAIN, "charger1")}, name="Device 1" + ) + + await handlers["stop_charging"](make_call(hass, {"entity_id": "sensor.foo"})) + + charger.command.assert_awaited_once_with("stop_charging_final") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_entity_id_not_found_raises( + hass: MagicMock, handlers: dict[str, Any], fake_registries: SimpleNamespace +) -> None: + """An unknown entity_id raises a HomeAssistantError.""" + with pytest.raises(HomeAssistantError, match="Unable to find entity"): + await handlers["stop_charging"](make_call(hass, {"entity_id": "sensor.missing"})) + + +async def test_entity_without_device_raises( + hass: MagicMock, handlers: dict[str, Any], fake_registries: SimpleNamespace +) -> None: + """An entity with no device_id raises a HomeAssistantError.""" + fake_registries.entities["sensor.foo"] = SimpleNamespace(device_id=None) + + with pytest.raises(HomeAssistantError, match="doesn't have a device"): + await handlers["stop_charging"](make_call(hass, {"entity_id": "sensor.foo"})) + + +async def test_device_id_not_found_raises( + hass: MagicMock, handlers: dict[str, Any], fake_registries: SimpleNamespace +) -> None: + """An unknown device_id raises a HomeAssistantError.""" + with pytest.raises(HomeAssistantError, match="Unable to find device"): + await handlers["stop_charging"](make_call(hass, {"device_id": "device_missing"})) + + +async def test_device_without_identifiers_raises( + hass: MagicMock, handlers: dict[str, Any], fake_registries: SimpleNamespace +) -> None: + """A device with no identifiers raises a HomeAssistantError.""" + fake_registries.devices["device1"] = SimpleNamespace(identifiers=set(), name="Device 1") + + with pytest.raises(HomeAssistantError, match="Unable to find identifiers"): + await handlers["stop_charging"](make_call(hass, {"device_id": "device1"})) + + +async def test_device_with_non_zaptec_identifier_raises( + hass: MagicMock, handlers: dict[str, Any], fake_registries: SimpleNamespace +) -> None: + """A device tied to a non-zaptec identifier domain raises a HomeAssistantError.""" + fake_registries.devices["device1"] = SimpleNamespace( + identifiers={("other_domain", "foo")}, name="Device 1" + ) + + with pytest.raises(HomeAssistantError, match="Non-zaptec device specified"): + await handlers["stop_charging"](make_call(hass, {"device_id": "device1"})) + + +async def test_no_ids_specified_raises_with_missing_field( + hass: MagicMock, handlers: dict[str, Any] +) -> None: + """Calling a handler with none of charger_id/device_id/entity_id set names the missing field.""" + with pytest.raises(HomeAssistantError, match="Missing field 'charger_id'"): + await handlers["stop_charging"](make_call(hass, {})) + + +async def test_unknown_zaptec_object_raises(hass: MagicMock, handlers: dict[str, Any]) -> None: + """A uid with no matching zaptec object raises a HomeAssistantError.""" + with pytest.raises(HomeAssistantError, match="Unable to find zaptec object"): + await handlers["stop_charging"](make_call(hass, {"charger_id": "charger_missing"})) + + +async def test_wrong_object_type_raises( + hass: MagicMock, manager: MagicMock, add_installation: Any, handlers: dict[str, Any] +) -> None: + """A uid resolving to the wrong zaptec object type raises a HomeAssistantError.""" + add_installation("install1") + + with pytest.raises(HomeAssistantError, match="is not a Charger"): + await handlers["stop_charging"](make_call(hass, {"charger_id": "install1"})) + + +async def test_object_without_coordinator_raises( + hass: MagicMock, manager: MagicMock, handlers: dict[str, Any] +) -> None: + """A resolved zaptec object with no matching coordinator raises a HomeAssistantError.""" + manager.zaptec["charger1"] = make_charger("charger1") + + with pytest.raises(HomeAssistantError, match="is not available"): + await handlers["stop_charging"](make_call(hass, {"charger_id": "charger1"})) + + +async def test_multiple_chargers_in_one_call_are_all_processed( + hass: MagicMock, + manager: MagicMock, + add_charger: Any, + handlers: dict[str, Any], + fake_registries: SimpleNamespace, +) -> None: + """A single call mixing a legacy charger_id and a device_id targets both chargers.""" + charger1, coordinator1 = add_charger("charger1") + charger2, coordinator2 = add_charger("charger2") + fake_registries.devices["device2"] = SimpleNamespace( + identifiers={(DOMAIN, "charger2")}, name="Device 2" + ) + + await handlers["stop_charging"]( + make_call(hass, {"charger_id": "charger1", "device_id": ["device2"]}) + ) + + charger1.command.assert_awaited_once_with("stop_charging_final") + coordinator1.trigger_poll.assert_awaited_once() + charger2.command.assert_awaited_once_with("stop_charging_final") + coordinator2.trigger_poll.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# Individual service handlers +# --------------------------------------------------------------------------- + + +async def test_stop_charging_wraps_command_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A command failure is wrapped in HomeAssistantError and no poll is triggered.""" + charger, coordinator = add_charger("charger1") + charger.command.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="stop_charging_final"): + await handlers["stop_charging"](make_call(hass, {"charger_id": "charger1"})) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_resume_charging_sends_command( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """resume_charging sends the resume_charging command and polls.""" + charger, coordinator = add_charger("charger1") + + await handlers["resume_charging"](make_call(hass, {"charger_id": "charger1"})) + + charger.command.assert_awaited_once_with("resume_charging") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_resume_charging_wraps_command_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A command failure is wrapped in HomeAssistantError and no poll is triggered.""" + charger, coordinator = add_charger("charger1") + charger.command.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="resume_charging"): + await handlers["resume_charging"](make_call(hass, {"charger_id": "charger1"})) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_authorize_charging_calls_authorize_charge( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """authorize_charging calls authorize_charge and polls.""" + charger, coordinator = add_charger("charger1") + + await handlers["authorize_charging"](make_call(hass, {"charger_id": "charger1"})) + + charger.authorize_charge.assert_awaited_once() + coordinator.trigger_poll.assert_awaited_once() + + +async def test_authorize_charging_wraps_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A authorize_charge failure is wrapped in HomeAssistantError.""" + charger, coordinator = add_charger("charger1") + charger.authorize_charge.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="authorize_charge"): + await handlers["authorize_charging"](make_call(hass, {"charger_id": "charger1"})) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_deauthorize_charging_sends_command( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """deauthorize_charging sends the deauthorize_and_stop command and polls.""" + charger, coordinator = add_charger("charger1") + + await handlers["deauthorize_charging"](make_call(hass, {"charger_id": "charger1"})) + + charger.command.assert_awaited_once_with("deauthorize_and_stop") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_deauthorize_charging_wraps_command_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A command failure is wrapped in HomeAssistantError and no poll is triggered.""" + charger, coordinator = add_charger("charger1") + charger.command.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="deauthorize_and_stop"): + await handlers["deauthorize_charging"](make_call(hass, {"charger_id": "charger1"})) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_restart_charger_sends_command( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """restart_charger sends the restart_charger command and polls.""" + charger, coordinator = add_charger("charger1") + + await handlers["restart_charger"](make_call(hass, {"charger_id": "charger1"})) + + charger.command.assert_awaited_once_with("restart_charger") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_restart_charger_wraps_command_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A command failure is wrapped in HomeAssistantError and no poll is triggered.""" + charger, coordinator = add_charger("charger1") + charger.command.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="restart_charger"): + await handlers["restart_charger"](make_call(hass, {"charger_id": "charger1"})) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_upgrade_firmware_sends_command( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """upgrade_firmware sends the upgrade_firmware command and polls.""" + charger, coordinator = add_charger("charger1") + + await handlers["upgrade_firmware"](make_call(hass, {"charger_id": "charger1"})) + + charger.command.assert_awaited_once_with("upgrade_firmware") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_upgrade_firmware_wraps_command_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A command failure is wrapped in HomeAssistantError and no poll is triggered.""" + charger, coordinator = add_charger("charger1") + charger.command.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="upgrade_firmware"): + await handlers["upgrade_firmware"](make_call(hass, {"charger_id": "charger1"})) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_limit_current_with_available_current_only( + hass: MagicMock, manager: MagicMock, add_installation: Any, handlers: dict[str, Any] +) -> None: + """Only availableCurrent is passed through when available_current is set.""" + installation, coordinator = add_installation("install1") + + await handlers["limit_current"]( + make_call(hass, {"installation_id": "install1", "available_current": 16}) + ) + + installation.set_limit_current.assert_awaited_once_with(availableCurrent=16) + coordinator.trigger_poll.assert_awaited_once() + + +async def test_limit_current_with_all_phases( + hass: MagicMock, manager: MagicMock, add_installation: Any, handlers: dict[str, Any] +) -> None: + """All three phase kwargs are passed through when the phase fields are set.""" + installation, coordinator = add_installation("install1") + + await handlers["limit_current"]( + make_call( + hass, + { + "installation_id": "install1", + "available_current_phase1": 10, + "available_current_phase2": 11, + "available_current_phase3": 12, + }, + ) + ) + + installation.set_limit_current.assert_awaited_once_with( + availableCurrentPhase1=10, availableCurrentPhase2=11, availableCurrentPhase3=12 + ) + coordinator.trigger_poll.assert_awaited_once() + + +async def test_limit_current_wraps_failure( + hass: MagicMock, manager: MagicMock, add_installation: Any, handlers: dict[str, Any] +) -> None: + """A set_limit_current failure is wrapped in HomeAssistantError and skips the poll.""" + installation, coordinator = add_installation("install1") + installation.set_limit_current.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="Limit current failed"): + await handlers["limit_current"]( + make_call(hass, {"installation_id": "install1", "available_current": 16}) + ) + + coordinator.trigger_poll.assert_not_awaited() + + +async def test_send_command_with_string_command( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """send_command forwards a string command and polls.""" + charger, coordinator = add_charger("charger1") + + await handlers["send_command"]( + make_call(hass, {"charger_id": "charger1", "command": "StopChargingFinal"}) + ) + + charger.command.assert_awaited_once_with("StopChargingFinal") + coordinator.trigger_poll.assert_awaited_once() + + +async def test_send_command_with_integer_command( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """send_command forwards an integer command.""" + charger, _coordinator = add_charger("charger1") + + await handlers["send_command"](make_call(hass, {"charger_id": "charger1", "command": 507})) + + charger.command.assert_awaited_once_with(507) + + +async def test_send_command_missing_command_raises( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """send_command without a command value raises before calling the charger.""" + charger, _coordinator = add_charger("charger1") + + with pytest.raises(HomeAssistantError, match="No Command received"): + await handlers["send_command"](make_call(hass, {"charger_id": "charger1"})) + + charger.command.assert_not_awaited() + + +async def test_send_command_wraps_failure( + hass: MagicMock, manager: MagicMock, add_charger: Any, handlers: dict[str, Any] +) -> None: + """A command failure is wrapped in HomeAssistantError and skips the poll.""" + charger, coordinator = add_charger("charger1") + charger.command.side_effect = Exception("boom") + + with pytest.raises(HomeAssistantError, match="'StopChargingFinal' failed"): + await handlers["send_command"]( + make_call(hass, {"charger_id": "charger1", "command": "StopChargingFinal"}) + ) + + coordinator.trigger_poll.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# Schema validation +# --------------------------------------------------------------------------- + + +def test_charger_id_schema_requires_one_of_the_id_fields() -> None: + """CHARGER_ID_SCHEMA rejects data with none of charger_id/device_id/entity_id.""" + with pytest.raises(vol.Invalid, match="At leas one of"): + CHARGER_ID_SCHEMA({}) + + +def test_charger_id_schema_accepts_entity_id() -> None: + """CHARGER_ID_SCHEMA accepts a bare entity_id and normalizes it to a list.""" + result = CHARGER_ID_SCHEMA({"entity_id": "sensor.foo"}) + assert result["entity_id"] == ["sensor.foo"] + + +def test_limit_current_schema_requires_current_value() -> None: + """LIMIT_CURRENT_SCHEMA rejects data with neither available_current nor all three phases.""" + with pytest.raises(vol.Invalid, match="Either 'available_current'"): + LIMIT_CURRENT_SCHEMA({"installation_id": "x"}) + + +def test_limit_current_schema_accepts_available_current() -> None: + """LIMIT_CURRENT_SCHEMA accepts a bare available_current.""" + result = LIMIT_CURRENT_SCHEMA({"installation_id": "x", "available_current": 16}) + assert result["available_current"] == 16 # noqa: PLR2004 + + +def test_limit_current_schema_accepts_all_three_phases() -> None: + """LIMIT_CURRENT_SCHEMA accepts all three phase fields together.""" + result = LIMIT_CURRENT_SCHEMA( + { + "installation_id": "x", + "available_current_phase1": 1, + "available_current_phase2": 2, + "available_current_phase3": 3, + } + ) + assert result["available_current_phase3"] == 3 # noqa: PLR2004 + + +def test_limit_current_schema_rejects_partial_phases() -> None: + """LIMIT_CURRENT_SCHEMA rejects only two of the three phase fields.""" + with pytest.raises(vol.Invalid, match="Either 'available_current'"): + LIMIT_CURRENT_SCHEMA( + { + "installation_id": "x", + "available_current_phase1": 1, + "available_current_phase2": 2, + } + ) + + +def test_limit_current_schema_rejects_current_and_phases_together() -> None: + """LIMIT_CURRENT_SCHEMA rejects mixing available_current with the phase fields.""" + with pytest.raises(vol.Invalid, match="Either 'available_current'"): + LIMIT_CURRENT_SCHEMA( + { + "installation_id": "x", + "available_current": 16, + "available_current_phase1": 1, + "available_current_phase2": 2, + "available_current_phase3": 3, + } + ) + + +def test_send_command_schema_accepts_string_and_int_commands() -> None: + """SEND_COMMAND_SCHEMA accepts both string and integer commands.""" + assert ( + SEND_COMMAND_SCHEMA({"charger_id": "x", "command": "StopChargingFinal"})["command"] + == "StopChargingFinal" + ) + assert SEND_COMMAND_SCHEMA({"charger_id": "x", "command": 5})["command"] == 5 # noqa: PLR2004 + + +def test_send_command_schema_requires_command() -> None: + """SEND_COMMAND_SCHEMA rejects data missing the command field.""" + with pytest.raises(vol.Invalid, match="required key not provided"): + SEND_COMMAND_SCHEMA({"charger_id": "x"}) + + +# --------------------------------------------------------------------------- +# services.yaml consistency +# --------------------------------------------------------------------------- + + +async def test_services_yaml_keys_match_registered_service_names(hass: MagicMock) -> None: + """services.yaml documents exactly the services async_setup_services registers. + + A key mismatch here (e.g. a typo) means HA's UI silently falls back to an + undocumented, field-less form for the real service, while the yaml entry + documents a service that doesn't exist. + """ + manager = MagicMock() + hass.services.has_service = MagicMock(return_value=False) + await async_setup_services(hass, manager) + registered = {call.args[1] for call in hass.services.async_register.call_args_list} + + documented = set(yaml.safe_load(SERVICES_YAML_PATH.read_text())) + + assert documented == registered diff --git a/tests/test_switch.py b/tests/test_switch.py new file mode 100644 index 00000000..bb788b5d --- /dev/null +++ b/tests/test_switch.py @@ -0,0 +1,232 @@ +"""Tests for switch.py.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity import DeviceInfo +import pytest + +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.switch import ( + ZapSwitchEntityDescription, + ZaptecCableLockSwitch, + ZaptecChargeSwitch, + ZaptecSwitch, +) +from custom_components.zaptec.zaptec import Charger + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +def make_charger(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Charger) whose .get() reads from data.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + charger.qual_id = "Charger[charger1]" + charger.get.side_effect = data.get + return charger + + +def test_switch_update_from_zaptec_sets_is_on(coordinator: ZaptecUpdateCoordinator) -> None: + """ZaptecSwitch._update_from_zaptec reads the raw boolean value for its key.""" + charger = make_charger({"permanent_cable_lock": True}) + description = ZapSwitchEntityDescription(key="permanent_cable_lock", cls=ZaptecSwitch) + entity = ZaptecSwitch(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_is_on is True # noqa: SLF001 + assert entity._attr_available is True # noqa: SLF001 + + +def test_charge_switch_update_from_zaptec_true_only_when_charging( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecChargeSwitch is only "on" when the mode is exactly Connected_Charging.""" + charger = make_charger({"charger_operation_mode": "Connected_Charging"}) + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_is_on is True # noqa: SLF001 + + +def test_charge_switch_available_checks_stop_command_when_on( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """When on, ZaptecChargeSwitch.available checks the stop_charging_final command.""" + charger = make_charger({}) + charger.is_command_valid.return_value = True + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + entity._attr_is_on = True # noqa: SLF001 + + assert entity.available is True + charger.is_command_valid.assert_called_once_with("stop_charging_final") + + +def test_charge_switch_available_checks_resume_command_when_off( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """When off, ZaptecChargeSwitch.available checks the resume_charging command.""" + charger = make_charger({}) + charger.is_command_valid.return_value = False + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + entity._attr_is_on = False # noqa: SLF001 + + assert entity.available is False + charger.is_command_valid.assert_called_once_with("resume_charging") + + +async def test_charge_switch_turn_on_resumes_charging_and_polls( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_on sends resume_charging and triggers a poll on success.""" + charger = make_charger({}) + charger.command = AsyncMock() + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_turn_on() + + charger.command.assert_awaited_once_with("resume_charging") + entity.trigger_poll.assert_awaited_once() + + +async def test_charge_switch_turn_on_wraps_command_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_on wraps a command failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.command = AsyncMock(side_effect=Exception("boom")) + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_turn_on() + + entity.trigger_poll.assert_not_called() + + +async def test_charge_switch_turn_off_stops_charging_and_polls( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_off sends stop_charging_final and triggers a poll on success.""" + charger = make_charger({}) + charger.command = AsyncMock() + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_turn_off() + + charger.command.assert_awaited_once_with("stop_charging_final") + entity.trigger_poll.assert_awaited_once() + + +async def test_charge_switch_turn_off_wraps_command_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_off wraps a command failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.command = AsyncMock(side_effect=Exception("boom")) + description = ZapSwitchEntityDescription(key="charger_operation_mode", cls=ZaptecChargeSwitch) + entity = ZaptecChargeSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_turn_off() + + entity.trigger_poll.assert_not_called() + + +async def test_cable_lock_switch_turn_on_locks_and_polls( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_on locks the cable and triggers a poll on success.""" + charger = make_charger({}) + charger.set_permanent_cable_lock = AsyncMock() + description = ZapSwitchEntityDescription( + key="permanent_cable_lock", cls=ZaptecCableLockSwitch + ) + entity = ZaptecCableLockSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_turn_on() + + charger.set_permanent_cable_lock.assert_awaited_once_with(True) # noqa: FBT003 + entity.trigger_poll.assert_awaited_once() + + +async def test_cable_lock_switch_turn_on_wraps_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_on wraps a failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.set_permanent_cable_lock = AsyncMock(side_effect=Exception("boom")) + description = ZapSwitchEntityDescription( + key="permanent_cable_lock", cls=ZaptecCableLockSwitch + ) + entity = ZaptecCableLockSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_turn_on() + + entity.trigger_poll.assert_not_called() + + +async def test_cable_lock_switch_turn_off_unlocks_and_polls( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_off unlocks the cable and triggers a poll on success.""" + charger = make_charger({}) + charger.set_permanent_cable_lock = AsyncMock() + description = ZapSwitchEntityDescription( + key="permanent_cable_lock", cls=ZaptecCableLockSwitch + ) + entity = ZaptecCableLockSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_turn_off() + + charger.set_permanent_cable_lock.assert_awaited_once_with(False) # noqa: FBT003 + entity.trigger_poll.assert_awaited_once() + + +async def test_cable_lock_switch_turn_off_wraps_failure( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_turn_off wraps a failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.set_permanent_cable_lock = AsyncMock(side_effect=Exception("boom")) + description = ZapSwitchEntityDescription( + key="permanent_cable_lock", cls=ZaptecCableLockSwitch + ) + entity = ZaptecCableLockSwitch(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_turn_off() + + entity.trigger_poll.assert_not_called() diff --git a/tests/test_update.py b/tests/test_update.py new file mode 100644 index 00000000..feaa27fd --- /dev/null +++ b/tests/test_update.py @@ -0,0 +1,88 @@ +"""Tests for update.py.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity import DeviceInfo +import pytest + +from custom_components.zaptec.coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions +from custom_components.zaptec.update import ZaptecUpdate, ZapUpdateEntityDescription +from custom_components.zaptec.zaptec import Charger + + +@pytest.fixture +def coordinator(hass: MagicMock, config_entry: Any) -> ZaptecUpdateCoordinator: + """Create a ZaptecUpdateCoordinator for testing.""" + manager = MagicMock() + options = ZaptecUpdateOptions( + name="test", + update_interval=600, + charging_update_interval=None, + tracked_devices=set(), + poll_args={}, + zaptec_object=None, + ) + return ZaptecUpdateCoordinator(hass, entry=config_entry, manager=manager, options=options) + + +def make_charger(data: dict[str, Any]) -> MagicMock: + """Create a MagicMock(spec=Charger) whose .get() reads from data.""" + charger = MagicMock(spec=Charger) + charger.id = "charger1" + charger.qual_id = "Charger[charger1]" + charger.get.side_effect = data.get + return charger + + +def test_update_from_zaptec_sets_installed_and_latest_version( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """ZaptecUpdate._update_from_zaptec reads both firmware version keys.""" + charger = make_charger( + { + "firmware_current_version": "1.0.0", + "firmware_available_version": "1.1.0", + } + ) + description = ZapUpdateEntityDescription(key="firmware_update", cls=ZaptecUpdate) + entity = ZaptecUpdate(coordinator, charger, description, DeviceInfo()) + + entity._update_from_zaptec() # noqa: SLF001 + + assert entity._attr_installed_version == "1.0.0" # noqa: SLF001 + assert entity._attr_latest_version == "1.1.0" # noqa: SLF001 + assert entity._attr_available is True # noqa: SLF001 + + +async def test_async_install_sends_upgrade_firmware_and_polls( + coordinator: ZaptecUpdateCoordinator, +) -> None: + """async_install sends the upgrade_firmware command and triggers a poll on success.""" + charger = make_charger({}) + charger.command = AsyncMock() + description = ZapUpdateEntityDescription(key="firmware_update", cls=ZaptecUpdate) + entity = ZaptecUpdate(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + await entity.async_install(version=None, backup=False) + + charger.command.assert_awaited_once_with("upgrade_firmware") + entity.trigger_poll.assert_awaited_once() + + +async def test_async_install_wraps_command_failure(coordinator: ZaptecUpdateCoordinator) -> None: + """async_install wraps a command failure in HomeAssistantError and skips the poll.""" + charger = make_charger({}) + charger.command = AsyncMock(side_effect=Exception("boom")) + description = ZapUpdateEntityDescription(key="firmware_update", cls=ZaptecUpdate) + entity = ZaptecUpdate(coordinator, charger, description, DeviceInfo()) + entity.trigger_poll = AsyncMock() + + with pytest.raises(HomeAssistantError): + await entity.async_install(version=None, backup=False) + + entity.trigger_poll.assert_not_called() diff --git a/tests/zaptec/test_api.py b/tests/zaptec/test_api.py index d469fdd2..825b05d9 100644 --- a/tests/zaptec/test_api.py +++ b/tests/zaptec/test_api.py @@ -1,13 +1,34 @@ """Tests for zaptec/api.py.""" +from http import HTTPStatus +import json import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock +import aiohttp import pytest -from custom_components.zaptec.zaptec.api import Zaptec +from custom_components.zaptec.zaptec.api import Charger, Installation, Zaptec, ZaptecBase +from custom_components.zaptec.zaptec.const import API_RETRIES +from custom_components.zaptec.zaptec.exceptions import ( + AuthenticationError, + RequestConnectionError, + RequestDataError, + RequestError, + RequestRetryError, + RequestTimeoutError, +) +from custom_components.zaptec.zaptec.redact import Redactor +from custom_components.zaptec.zaptec.zconst import ZCONST _LOGGER = logging.getLogger(__name__) +# One retry means the request was attempted twice (one failure, one success). +CALLS_AFTER_ONE_RETRY = 2 +# The Retry-After header value (seconds) used in the transient-status tests. +RETRY_AFTER_SECONDS = 2.0 + @pytest.mark.asyncio async def test_api(zaptec_username: str, zaptec_password: str) -> None: @@ -33,3 +54,922 @@ async def test_api(zaptec_username: str, zaptec_password: str) -> None: # Print all the attributes. for obj in zaptec.objects(): _LOGGER.info(obj.asdict()) + + +# =========================================================================== +# Offline unit tests (no network / no live login required) +# =========================================================================== +# +# These exercise the pure logic and the request/retry machinery using a fake +# aiohttp ClientSession, so they run without credentials or DNS access. + + +class FakeResponse: + """Minimal stand-in for aiohttp.ClientResponse.""" + + def __init__( + self, + status: int, + *, + json_data: object = None, + read_data: bytes = b"", + text_data: str = "", + headers: dict | None = None, + ) -> None: + """Store the canned response data.""" + self.status = status + self._json_data = json_data + self._read_data = read_data + self._text_data = text_data + self.headers = headers or {} + + async def json(self, content_type: str | None = None) -> object: + """Return the canned JSON body, or raise if none was configured.""" + if self._json_data is None: + raise json.JSONDecodeError("no json body", "", 0) + return self._json_data + + async def read(self) -> bytes: + """Return the canned raw body.""" + return self._read_data + + async def text(self) -> str: + """Return the canned text body.""" + return self._text_data + + +class _FakeRequestCM: + """Async context manager returned by FakeSession.request().""" + + def __init__( + self, *, response: FakeResponse | None = None, exc: BaseException | None = None + ) -> None: + """Store the response to yield, or the exception to raise on enter.""" + self._response = response + self._exc = exc + + async def __aenter__(self) -> FakeResponse: + """Raise the configured exception, or return the response.""" + if self._exc is not None: + raise self._exc + assert self._response is not None + return self._response + + async def __aexit__(self, *exc_info: object) -> bool: + """Never suppress exceptions.""" + return False + + +class FakeSession: + """Minimal aiohttp.ClientSession stand-in driven by a list of outcomes. + + Each call to request() consumes the next outcome (a FakeResponse to yield + or an exception to raise); the last outcome repeats for further calls. + """ + + def __init__(self, outcomes: list[FakeResponse | BaseException]) -> None: + """Store the sequence of per-call outcomes.""" + self._outcomes = outcomes + self.calls: list[tuple[str, str, dict]] = [] + + def request(self, *, method: str, url: str, **kwargs: object) -> _FakeRequestCM: + """Record the call and return the matching outcome.""" + idx = len(self.calls) + self.calls.append((method, url, kwargs)) + outcome = self._outcomes[min(idx, len(self._outcomes) - 1)] + if isinstance(outcome, BaseException): + return _FakeRequestCM(exc=outcome) + return _FakeRequestCM(response=outcome) + + async def close(self) -> None: + """No-op close.""" + + +def _make_zaptec( + outcomes: list[FakeResponse | BaseException], **kwargs: object +) -> tuple[Zaptec, FakeSession]: + """Build a Zaptec client backed by a FakeSession, returning both.""" + session = FakeSession(outcomes) + zap = Zaptec("user", "pass", client=session, redact_logs=False, **kwargs) + return zap, session + + +def _fake_owner() -> SimpleNamespace: + """Return a stand-in for the `zaptec` owner used by set_attributes.""" + return SimpleNamespace(redact=Redactor(do_redact=False), show_all_updates=False) + + +# --------------------------------------------------------------------------- +# Zaptec.request status handling +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_request_ok_returns_json() -> None: + """A 200 response returns the decoded JSON payload.""" + payload = {"value": "answer"} + zap, session = _make_zaptec([FakeResponse(HTTPStatus.OK, json_data=payload)]) + result = await zap.request("unregistered/url") + assert result == payload + assert len(session.calls) == 1 + + +@pytest.mark.asyncio +async def test_request_no_content_returns_bytes() -> None: + """A 204/201 response returns the raw body bytes.""" + zap, _ = _make_zaptec([FakeResponse(HTTPStatus.NO_CONTENT, read_data=b"done")]) + result = await zap.request("unregistered/url", method="post") + assert result == b"done" + + +@pytest.mark.asyncio +async def test_request_invalid_json_raises_data_error() -> None: + """A 200 with an undecodable body raises RequestDataError.""" + zap, _ = _make_zaptec([FakeResponse(HTTPStatus.OK, json_data=None)]) + with pytest.raises(RequestDataError): + await zap.request("unregistered/url") + + +@pytest.mark.asyncio +async def test_request_error_status_raises_with_code() -> None: + """A non-retryable error status raises RequestError carrying the code.""" + zap, session = _make_zaptec([FakeResponse(HTTPStatus.NOT_FOUND)]) + with pytest.raises(RequestError) as excinfo: + await zap.request("unregistered/url") + assert excinfo.value.error_code == HTTPStatus.NOT_FOUND + assert len(session.calls) == 1 + + +@pytest.mark.asyncio +async def test_request_500_on_post_raises_immediately() -> None: + """A 500 on a POST is not retried and raises RequestError immediately.""" + zap, session = _make_zaptec( + [FakeResponse(HTTPStatus.INTERNAL_SERVER_ERROR, text_data="server error")] + ) + with pytest.raises(RequestError) as excinfo: + await zap.request("unregistered/url", method="post") + assert excinfo.value.error_code == HTTPStatus.INTERNAL_SERVER_ERROR + assert len(session.calls) == 1 + + +@pytest.mark.asyncio +async def test_request_500_on_get_retries_then_raises_retry_error() -> None: + """A persistent 500 on a GET is retried until exhaustion (RequestRetryError).""" + zap, session = _make_zaptec([FakeResponse(HTTPStatus.INTERNAL_SERVER_ERROR)], max_time=0.001) + with pytest.raises(RequestRetryError): + await zap.request("unregistered/url") + assert len(session.calls) == API_RETRIES + + +@pytest.mark.asyncio +async def test_request_401_refreshes_token_then_succeeds( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A 401 triggers a token refresh and the request is retried.""" + payload = {"ok": "yes"} + zap, _ = _make_zaptec( + [FakeResponse(HTTPStatus.UNAUTHORIZED), FakeResponse(HTTPStatus.OK, json_data=payload)] + ) + refresh = AsyncMock() + monkeypatch.setattr(zap, "_refresh_token", refresh) + + result = await zap.request("unregistered/url") + # Reaching the 200 payload after a 401 proves the request was retried. + assert result == payload + refresh.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_request_connection_error_retried_then_raises() -> None: + """Connection errors are retried and finally raise RequestConnectionError.""" + zap, session = _make_zaptec([aiohttp.ClientConnectionError("boom")], max_time=0.001) + with pytest.raises(RequestConnectionError): + await zap.request("unregistered/url") + assert len(session.calls) == API_RETRIES + + +@pytest.mark.asyncio +async def test_request_timeout_retried_then_raises() -> None: + """Timeouts are retried and finally raise RequestTimeoutError.""" + zap, session = _make_zaptec([TimeoutError()], max_time=0.001) + with pytest.raises(RequestTimeoutError): + await zap.request("unregistered/url") + assert len(session.calls) == API_RETRIES + + +# --------------------------------------------------------------------------- +# Transient HTTP status retry (429/502/503/504) — issue #392 +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "status", + [ + HTTPStatus.TOO_MANY_REQUESTS, + HTTPStatus.BAD_GATEWAY, + HTTPStatus.SERVICE_UNAVAILABLE, + HTTPStatus.GATEWAY_TIMEOUT, + ], +) +async def test_request_transient_status_retries_then_succeeds(status: HTTPStatus) -> None: + """A transient server status is retried, and a later 200 is returned.""" + payload = {"value": "ok"} + zap, session = _make_zaptec( + [FakeResponse(status), FakeResponse(HTTPStatus.OK, json_data=payload)], + max_time=0.001, + ) + result = await zap.request("unregistered/url") + assert result == payload + assert len(session.calls) == CALLS_AFTER_ONE_RETRY + + +@pytest.mark.asyncio +async def test_request_transient_status_on_post_retries() -> None: + """Unlike 500, a transient 503 is retried even for POST (infra-level error).""" + zap, session = _make_zaptec( + [FakeResponse(HTTPStatus.SERVICE_UNAVAILABLE), FakeResponse(HTTPStatus.NO_CONTENT)], + max_time=0.001, + ) + result = await zap.request("unregistered/url", method="post") + assert result == b"" + assert len(session.calls) == CALLS_AFTER_ONE_RETRY + + +@pytest.mark.asyncio +async def test_request_persistent_503_raises_request_error_with_code() -> None: + """A persistent 503 is retried to exhaustion, then raises RequestError(503).""" + zap, session = _make_zaptec([FakeResponse(HTTPStatus.SERVICE_UNAVAILABLE)], max_time=0.001) + with pytest.raises(RequestError) as excinfo: + await zap.request("unregistered/url") + assert excinfo.value.error_code == HTTPStatus.SERVICE_UNAVAILABLE + assert len(session.calls) == API_RETRIES + + +@pytest.mark.asyncio +async def test_refresh_token_retries_transient_then_succeeds() -> None: + """A transient 503 on the token endpoint is retried instead of failing setup.""" + zap, session = _make_zaptec( + [ + FakeResponse(HTTPStatus.SERVICE_UNAVAILABLE), + FakeResponse(HTTPStatus.OK, json_data={"access_token": "abc"}), + ], + max_time=0.001, + ) + await zap.login() + assert len(session.calls) == CALLS_AFTER_ONE_RETRY + await zap.request("some/url") + _, _, kwargs = session.calls[-1] + assert kwargs["headers"]["Authorization"] == "Bearer abc" + + +@pytest.mark.asyncio +async def test_refresh_token_persistent_503_raises_request_error() -> None: + """A persistent 503 on the token endpoint eventually raises RequestError(503).""" + zap, session = _make_zaptec([FakeResponse(HTTPStatus.SERVICE_UNAVAILABLE)], max_time=0.001) + with pytest.raises(RequestError) as excinfo: + await zap.login() + assert excinfo.value.error_code == HTTPStatus.SERVICE_UNAVAILABLE + assert len(session.calls) == API_RETRIES + + +@pytest.mark.asyncio +async def test_retry_after_header_is_honored(monkeypatch: pytest.MonkeyPatch) -> None: + """A Retry-After header sets the next backoff delay for a transient status.""" + sleep = AsyncMock() + monkeypatch.setattr("custom_components.zaptec.zaptec.api.asyncio.sleep", sleep) + zap, _ = _make_zaptec( + [ + FakeResponse(HTTPStatus.SERVICE_UNAVAILABLE, headers={"Retry-After": "2"}), + FakeResponse(HTTPStatus.OK, json_data={"ok": True}), + ], + ) + await zap.request("unregistered/url") + slept = [call.args[0] for call in sleep.await_args_list] + assert RETRY_AFTER_SECONDS in slept + + +# --------------------------------------------------------------------------- +# ZaptecBase.state_to_attrs +# --------------------------------------------------------------------------- + + +def test_state_to_attrs_maps_and_prefers_value() -> None: + """Values are mapped via keydict; `Value` wins over `ValueAsString`.""" + keydict = {"1": "current", "2": "voltage"} + data = [ + {"StateId": "1", "ValueAsString": "10"}, + {"StateId": "2", "Value": "230", "ValueAsString": "ignored"}, + ] + out = ZaptecBase.state_to_attrs(data, "StateId", keydict) + assert out == {"current": "10", "voltage": "230"} + + +def test_state_to_attrs_unknown_key_uses_fallback_name() -> None: + """A StateId missing from keydict falls back to ' '.""" + out = ZaptecBase.state_to_attrs([{"StateId": "99", "Value": "x"}], "StateId", {}) + assert out == {"StateId 99": "x"} + + +def test_state_to_attrs_skips_missing_key_and_missing_value() -> None: + """Entries without the key, or without any value, are skipped.""" + data = [ + {"NoStateId": "1", "Value": "x"}, # missing key -> skipped + {"StateId": "1"}, # no Value/ValueAsString -> skipped + ] + out = ZaptecBase.state_to_attrs(data, "StateId", {"1": "current"}) + assert out == {} + + +def test_state_to_attrs_excludes() -> None: + """Excluded ids are dropped.""" + data = [ + {"StateId": "1", "Value": "a"}, + {"StateId": "2", "Value": "b"}, + ] + out = ZaptecBase.state_to_attrs(data, "StateId", {"1": "one", "2": "two"}, excludes={"2"}) + assert out == {"one": "a"} + + +def test_state_to_attrs_duplicate_last_wins() -> None: + """When two entries map to the same attribute, the last one wins.""" + data = [ + {"StateId": "1", "Value": "first"}, + {"StateId": "1", "Value": "second"}, + ] + out = ZaptecBase.state_to_attrs(data, "StateId", {"1": "current"}) + assert out == {"current": "second"} + + +# --------------------------------------------------------------------------- +# ZaptecBase.set_attributes type conversion +# --------------------------------------------------------------------------- + + +def test_set_attributes_applies_type_conversion() -> None: + """Known attributes are converted per ATTR_TYPES; keys become snake_case.""" + chg = Charger( + {"ChargerMaxCurrent": "16", "IsOnline": "true", "Name": "Garage"}, _fake_owner() + ) + assert chg["ChargerMaxCurrent"] == float("16") + assert chg["IsOnline"] is True + assert chg["Name"] == "Garage" + + +def test_set_attributes_unknown_key_passthrough() -> None: + """Unknown attributes are stored unchanged under a snake_case key.""" + chg = Charger({"SomeUnknownKey": "value"}, _fake_owner()) + assert chg["SomeUnknownKey"] == "value" + assert "some_unknown_key" in chg.asdict() + + +def test_set_attributes_conversion_failure_falls_back_to_raw() -> None: + """A failing type conversion keeps the raw value instead of raising.""" + chg = Charger({"ChargerMaxCurrent": "not-a-number"}, _fake_owner()) + assert chg["ChargerMaxCurrent"] == "not-a-number" + + +def test_set_attributes_updates_existing_value() -> None: + """Re-setting an attribute overwrites the previous value.""" + chg = Charger({"ChargerMaxCurrent": "16"}, _fake_owner()) + chg.set_attributes({"ChargerMaxCurrent": "32"}) + assert chg["ChargerMaxCurrent"] == float("32") + + +# --------------------------------------------------------------------------- +# Charger.is_command_valid +# --------------------------------------------------------------------------- + + +def _charger_with_state( + *, operation_mode: str | None = None, final_stop_active: str | None = None +) -> Charger: + """Build a Charger carrying the state attributes is_command_valid reads.""" + data: dict[str, str] = {"Id": "chg-1"} + if operation_mode is not None: + data["ChargerOperationMode"] = operation_mode + if final_stop_active is not None: + data["FinalStopActive"] = final_stop_active + return Charger(data, _fake_owner()) + + +def test_is_command_valid_unrelated_command_is_always_valid() -> None: + """Commands other than resume/stop are always valid (no state needed).""" + chg = _charger_with_state() + assert chg.is_command_valid("restart_charger", raise_value_error_if_invalid=True) is True + + +def test_is_command_valid_resume_when_paused_is_valid() -> None: + """Resume is allowed only when the charger is paused.""" + chg = _charger_with_state(operation_mode="Connected_Finished", final_stop_active="1") + assert chg.is_command_valid("resume_charging") is True + + +def test_is_command_valid_resume_when_not_paused_is_invalid() -> None: + """Resume is rejected when not paused, and raises when requested.""" + chg = _charger_with_state(operation_mode="Connected_Charging", final_stop_active="0") + assert chg.is_command_valid("resume_charging") is False + with pytest.raises(ValueError, match="not paused"): + chg.is_command_valid("resume_charging", raise_value_error_if_invalid=True) + + +def test_is_command_valid_stop_when_paused_is_invalid() -> None: + """Stop/pause is rejected when already paused.""" + chg = _charger_with_state(operation_mode="Connected_Finished", final_stop_active="1") + assert chg.is_command_valid("stop_charging_final") is False + + +def test_is_command_valid_stop_when_disconnected_is_invalid() -> None: + """Stop/pause is rejected when disconnected.""" + chg = _charger_with_state(operation_mode="Disconnected", final_stop_active="0") + assert chg.is_command_valid("stop_charging_final") is False + + +def test_is_command_valid_stop_when_charging_is_valid() -> None: + """Stop/pause is allowed while actively charging.""" + chg = _charger_with_state(operation_mode="Connected_Charging", final_stop_active="0") + assert chg.is_command_valid("stop_charging_final") is True + + +def test_is_command_valid_missing_final_stop_raises_type_error() -> None: + """KNOWN ISSUE (Phase 3): a missing FinalStopActive makes int(None) raise. + + Characterizes current behavior; the correctness cleanup should guard this. + """ + chg = _charger_with_state(operation_mode="Connected_Finished") + with pytest.raises(TypeError): + chg.is_command_valid("resume_charging") + + +# --------------------------------------------------------------------------- +# Installation.stream_update routing +# --------------------------------------------------------------------------- + + +def _installation_with_charger() -> tuple[Installation, Charger]: + """Build an installation owning a single charger spy.""" + owner = _fake_owner() + inst = Installation({"Id": "inst-1"}, owner) + charger = Charger({"Id": "chg-1"}, owner) + charger.set_attributes = Mock() # spy on the routed update + inst.chargers = [charger] + return inst, charger + + +def test_stream_update_routes_to_matching_charger(monkeypatch: pytest.MonkeyPatch) -> None: + """A message with a known ChargerId updates that charger.""" + # observations is populated by build(); inject it for this offline test. + monkeypatch.setattr(ZCONST, "observations", {"1": "current"}, raising=False) + inst, charger = _installation_with_charger() + inst.stream_update({"ChargerId": "chg-1", "StateId": "1", "ValueAsString": "5"}) + charger.set_attributes.assert_called_once() + + +def test_stream_update_unknown_charger_is_ignored() -> None: + """A message for an unknown charger does not update anything.""" + inst, charger = _installation_with_charger() + inst.stream_update({"ChargerId": "other", "StateId": "1", "ValueAsString": "5"}) + charger.set_attributes.assert_not_called() + + +def test_stream_update_missing_charger_id_is_ignored() -> None: + """A message without a ChargerId is ignored.""" + inst, charger = _installation_with_charger() + inst.stream_update({"StateId": "1", "ValueAsString": "5"}) + charger.set_attributes.assert_not_called() + + +def test_stream_update_zero_guid_is_ignored() -> None: + """The all-zero charger id is explicitly ignored.""" + inst, charger = _installation_with_charger() + inst.stream_update({"ChargerId": "00000000-0000-0000-0000-000000000000"}) + charger.set_attributes.assert_not_called() + + +# --------------------------------------------------------------------------- +# Zaptec mapping / registry + poll dispatch +# --------------------------------------------------------------------------- + + +def test_zaptec_register_and_contains() -> None: + """register/unregister and __contains__ handle both ids and objects.""" + zap, _ = _make_zaptec([]) + charger = Charger({"Id": "c1"}, zap) + + zap.register("c1", charger) + assert "c1" in zap # by id (str) + assert charger in zap # by object (ZaptecBase branch) + assert "nope" not in zap + assert object() not in zap # arbitrary object -> not present + + zap.unregister("c1") + assert "c1" not in zap + + +def test_zaptec_register_duplicate_raises() -> None: + """Registering the same id twice raises.""" + zap, _ = _make_zaptec([]) + charger = Charger({"Id": "c1"}, zap) + zap.register("c1", charger) + with pytest.raises(ValueError, match="already registered"): + zap.register("c1", charger) + + +def test_zaptec_qual_id_unknown_returns_id() -> None: + """qual_id returns the raw id for an unknown object.""" + zap, _ = _make_zaptec([]) + assert zap.qual_id("missing-id") == "missing-id" + + +@pytest.mark.asyncio +async def test_poll_dispatches_info_and_state() -> None: + """poll() calls poll_info/poll_state on the selected objects.""" + zap, _ = _make_zaptec([]) + obj = Mock() + obj.poll_info = AsyncMock() + obj.poll_state = AsyncMock() + zap.register("x", obj) + + await zap.poll(["x"], info=True, state=True) + obj.poll_info.assert_awaited_once() + obj.poll_state.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_poll_unknown_object_raises() -> None: + """poll() raises for an unregistered object id.""" + zap, _ = _make_zaptec([]) + with pytest.raises(ValueError, match="not found"): + await zap.poll(["missing"]) + + +# --------------------------------------------------------------------------- +# Charger command / settings wrappers +# --------------------------------------------------------------------------- + + +def _charger_with_session( + outcomes: list[FakeResponse | BaseException], +) -> tuple[Charger, FakeSession]: + """Build a charger whose owner performs requests against a FakeSession.""" + zap, session = _make_zaptec(outcomes) + return Charger({"Id": "c1"}, zap), session + + +def _installation_with_session( + outcomes: list[FakeResponse | BaseException], +) -> tuple[Installation, FakeSession]: + """Build an installation whose owner performs requests against a FakeSession.""" + zap, session = _make_zaptec(outcomes) + return Installation({"Id": "i1"}, zap), session + + +@pytest.mark.asyncio +async def test_command_posts_to_send_command_url(monkeypatch: pytest.MonkeyPatch) -> None: + """A named command resolves to its id and POSTs to SendCommand.""" + monkeypatch.setattr(ZCONST, "commands", {"restart_charger": 102}, raising=False) + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.command("restart_charger") + + method, url, _ = session.calls[-1] + assert method == "post" + assert url.endswith("chargers/c1/SendCommand/102") + + +@pytest.mark.asyncio +async def test_command_authorize_charge_alias() -> None: + """The authorize_charge alias POSTs to the authorizecharge endpoint.""" + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.command("authorize_charge") + + method, url, _ = session.calls[-1] + assert method == "post" + assert url.endswith("chargers/c1/authorizecharge") + + +@pytest.mark.asyncio +async def test_command_unknown_raises(monkeypatch: pytest.MonkeyPatch) -> None: + """An unknown command raises without issuing a request.""" + monkeypatch.setattr(ZCONST, "commands", {}, raising=False) + charger, session = _charger_with_session([]) + + with pytest.raises(ValueError, match="Unknown command"): + await charger.command("does_not_exist") + assert session.calls == [] + + +@pytest.mark.asyncio +async def test_set_settings_valid() -> None: + """Valid settings are POSTed to the charger update endpoint.""" + settings = {"maxChargeCurrent": 16} + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.set_settings(settings) + + method, url, kwargs = session.calls[-1] + assert method == "post" + assert url.endswith("chargers/c1/update") + assert kwargs["json"] == settings + + +@pytest.mark.asyncio +async def test_set_settings_unknown_key_raises() -> None: + """An unknown setting key raises without issuing a request.""" + charger, session = _charger_with_session([]) + with pytest.raises(ValueError, match="Unknown setting"): + await charger.set_settings({"bogusKey": 1}) + assert session.calls == [] + + +@pytest.mark.asyncio +async def test_authorize_charge_posts() -> None: + """authorize_charge POSTs to the authorizecharge endpoint.""" + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.authorize_charge() + + method, url, _ = session.calls[-1] + assert method == "post" + assert url.endswith("chargers/c1/authorizecharge") + + +@pytest.mark.asyncio +async def test_set_permanent_cable_lock_payload() -> None: + """The permanent cable lock is sent under Cable.PermanentLock.""" + expected = {"Cable": {"PermanentLock": True}} + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.set_permanent_cable_lock(lock=True) + + method, url, kwargs = session.calls[-1] + assert method == "post" + assert url.endswith("chargers/c1/localSettings") + assert kwargs["json"] == expected + + +@pytest.mark.asyncio +async def test_set_hmi_brightness_payload() -> None: + """The HMI brightness is sent under Device.HmiBrightness.""" + brightness = 0.5 + expected = {"Device": {"HmiBrightness": brightness}} + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.set_hmi_brightness(brightness) + + method, url, kwargs = session.calls[-1] + assert method == "post" + assert url.endswith("chargers/c1/localSettings") + assert kwargs["json"] == expected + + +# --------------------------------------------------------------------------- +# Installation current-limit setters +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_set_limit_current_available_current() -> None: + """A single availableCurrent limit is POSTed to the installation update.""" + expected = {"availableCurrent": 16} + inst, session = _installation_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await inst.set_limit_current(**expected) + + method, url, kwargs = session.calls[-1] + assert method == "post" + assert url.endswith("installation/i1/update") + assert kwargs["json"] == expected + + +@pytest.mark.asyncio +async def test_set_limit_current_requires_current_argument() -> None: + """Calling without any current argument raises.""" + inst, session = _installation_with_session([]) + with pytest.raises(ValueError, match="availableCurrent"): + await inst.set_limit_current() + assert session.calls == [] + + +@pytest.mark.asyncio +async def test_set_limit_current_partial_phases_raise() -> None: + """Providing availableCurrent with only some per-phase currents raises.""" + inst, session = _installation_with_session([]) + with pytest.raises(ValueError, match="all of them must be set"): + await inst.set_limit_current(availableCurrent=10, availableCurrentPhase1=10) + assert session.calls == [] + + +@pytest.mark.asyncio +async def test_set_limit_current_out_of_range_raises() -> None: + """A current above the installation maximum raises.""" + inst, session = _installation_with_session([]) + with pytest.raises(ValueError, match="between 0 and"): + await inst.set_limit_current(availableCurrent=1000) + assert session.calls == [] + + +@pytest.mark.asyncio +async def test_set_three_to_one_phase_switch_current() -> None: + """The 3-to-1 phase switch current is POSTed to the installation update.""" + current = 16 + expected = {"threeToOnePhaseSwitchCurrent": current} + inst, session = _installation_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await inst.set_three_to_one_phase_switch_current(current) + + method, url, kwargs = session.calls[-1] + assert method == "post" + assert url.endswith("installation/i1/update") + assert kwargs["json"] == expected + + +@pytest.mark.asyncio +async def test_set_three_to_one_phase_switch_current_out_of_range_raises() -> None: + """An out-of-range 3-to-1 phase switch current raises.""" + inst, session = _installation_with_session([]) + with pytest.raises(ValueError, match="between 0 and"): + await inst.set_three_to_one_phase_switch_current(1000) + assert session.calls == [] + + +# --------------------------------------------------------------------------- +# Charger.poll_info / poll_state +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_poll_info_happy_path(monkeypatch: pytest.MonkeyPatch) -> None: + """poll_info fetches the charger and applies the attributes.""" + # Payload validation has its own tests; bypass it here. + monkeypatch.setattr("custom_components.zaptec.zaptec.api.validate", Mock()) + charger, _ = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={"Id": "c1"})]) + charger.set_attributes = Mock() + + await charger.poll_info() + + charger.set_attributes.assert_called_once() + + +@pytest.mark.asyncio +async def test_poll_info_falls_back_to_charger_list_on_forbidden( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A 403 on the charger endpoint falls back to the chargers list.""" + monkeypatch.setattr("custom_components.zaptec.zaptec.api.validate", Mock()) + charger, _ = _charger_with_session( + [ + FakeResponse(HTTPStatus.FORBIDDEN), + FakeResponse(HTTPStatus.OK, json_data={"Data": [{"Id": "c1", "Name": "x"}]}), + ] + ) + charger.set_attributes = Mock() + + await charger.poll_info() + + # Reached only via the fallback branch, since the first request raised. + charger.set_attributes.assert_called_once() + + +@pytest.mark.asyncio +async def test_poll_info_non_forbidden_error_propagates() -> None: + """A non-403 error is re-raised rather than falling back.""" + charger, _ = _charger_with_session([FakeResponse(HTTPStatus.NOT_FOUND)]) + with pytest.raises(RequestError): + await charger.poll_info() + + +@pytest.mark.asyncio +async def test_poll_state_happy_path(monkeypatch: pytest.MonkeyPatch) -> None: + """poll_state fetches the state list and applies the mapped attributes.""" + monkeypatch.setattr(ZCONST, "observations", {"1": "current"}, raising=False) + monkeypatch.setattr("custom_components.zaptec.zaptec.api.validate", Mock()) + charger, session = _charger_with_session( + [FakeResponse(HTTPStatus.OK, json_data=[{"StateId": "1", "ValueAsString": "5"}])] + ) + charger.set_attributes = Mock() + + await charger.poll_state() + + _, url, _ = session.calls[-1] + assert url.endswith("chargers/c1/state") + charger.set_attributes.assert_called_once() + + +@pytest.mark.asyncio +async def test_poll_state_forbidden_is_ignored() -> None: + """A 403 on the state endpoint is swallowed (no attribute update).""" + charger, _ = _charger_with_session([FakeResponse(HTTPStatus.FORBIDDEN)]) + charger.set_attributes = Mock() + + await charger.poll_state() + + charger.set_attributes.assert_not_called() + + +# --------------------------------------------------------------------------- +# Zaptec._refresh_token / login +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_login_stores_and_uses_access_token() -> None: + """A successful login obtains a token and sends it on later requests.""" + zap, session = _make_zaptec( + [ + FakeResponse(HTTPStatus.OK, json_data={"access_token": "abc"}), + FakeResponse(HTTPStatus.OK, json_data={}), + ] + ) + + await zap.login() + await zap.request("some/url") + + _, _, kwargs = session.calls[-1] + assert kwargs["headers"]["Authorization"] == "Bearer abc" + + +@pytest.mark.asyncio +async def test_login_bad_credentials_raises_authentication_error() -> None: + """A 400 from the token endpoint raises AuthenticationError.""" + zap, _ = _make_zaptec( + [FakeResponse(HTTPStatus.BAD_REQUEST, json_data={"error_description": "nope"})] + ) + with pytest.raises(AuthenticationError): + await zap.login() + + +# --------------------------------------------------------------------------- +# Assorted small accessors / lifecycle +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_command_by_numeric_id(monkeypatch: pytest.MonkeyPatch) -> None: + """A numeric command id is sent directly to SendCommand.""" + monkeypatch.setattr(ZCONST, "commands", {102: "restart_charger"}, raising=False) + charger, session = _charger_with_session([FakeResponse(HTTPStatus.OK, json_data={})]) + + await charger.command(102) + + _, url, _ = session.calls[-1] + assert url.endswith("chargers/c1/SendCommand/102") + + +@pytest.mark.asyncio +async def test_installation_poll_info_strips_logo(monkeypatch: pytest.MonkeyPatch) -> None: + """poll_info removes the bulky SupportGroup logo before storing attributes.""" + monkeypatch.setattr("custom_components.zaptec.zaptec.api.validate", Mock()) + inst, _ = _installation_with_session( + [FakeResponse(HTTPStatus.OK, json_data={"SupportGroup": {"LogoBase64": "AAAA"}})] + ) + inst.set_attributes = Mock() + + await inst.poll_info() + + inst.set_attributes.assert_called_once() + stored = inst.set_attributes.call_args[0][0] + assert stored["SupportGroup"]["LogoBase64"].startswith(" None: + """is_charging reflects the operation mode.""" + assert Charger({"ChargerOperationMode": "Connected_Charging"}, _fake_owner()).is_charging() + assert not Charger({"ChargerOperationMode": "Disconnected"}, _fake_owner()).is_charging() + + +def test_charger_model_from_device_id() -> None: + """The model is derived from the DeviceId prefix.""" + chg = Charger({"DeviceId": "ZAP123456"}, _fake_owner()) + assert chg.model_prefix == "ZAP" + assert chg.model == "Zaptec Go" + + +def test_zaptec_collections_and_accessors() -> None: + """objects/installations/chargers and iteration reflect the registry.""" + zap, _ = _make_zaptec([]) + inst = Installation({"Id": "i1"}, zap) + chg = Charger({"Id": "c1"}, zap) + zap.register("i1", inst) + zap.register("c1", chg) + + ids = {"i1", "c1"} + assert set(zap) == ids # __iter__ + assert len(zap) == len(ids) # __len__ + # ZaptecBase subclasses Mapping (unhashable), so compare as lists. + objs = list(zap.objects()) + assert inst in objs + assert chg in objs + assert list(zap.installations) == [inst] + assert list(zap.chargers) == [chg] + + zap.unregister("c1") + assert set(zap) == {"i1"} + + +@pytest.mark.asyncio +async def test_zaptec_async_context_manager_closes_internal_client() -> None: + """Entering/exiting the context manager works with an internally-created client.""" + async with Zaptec("user", "pass") as zap: + assert isinstance(zap, Zaptec)