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/tests/conftest.py b/tests/conftest.py index ba07a5c5..fbe31216 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,12 +2,63 @@ import asyncio import os +from typing import Any +from unittest.mock import MagicMock 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.""" 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_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()