Add myuplink switch platform (#110810)

* Add switch platform

* Add mypulink switch platform

* Update tests according to review

* Address more review comments

* Adjust types

* More typing

* Fix typo

* Use constants in tests

* Revert constants

* Catch aiohttp.ClientError when API call fails

* Add test case for failed async_set_device_points call

* Test api failures for both toggle directions

* Use parametrize for testing switching

homeassistant/components/myuplink/__init__.py

@@ -23,6 +23,7 @@ from .coordinator import MyUplinkDataCoordinator
 PLATFORMS: list[Platform] = [
     Platform.BINARY_SENSOR,
     Platform.SENSOR,
+    Platform.SWITCH,
     Platform.UPDATE,
 ]
 

homeassistant/components/myuplink/helpers.py

@@ -13,9 +13,7 @@ def find_matching_platform(device_point: DevicePoint) -> Platform:
         and device_point.enum_values[1]["value"] == "1"
     ):
         if device_point.writable:
-            # Change to Platform.SWITCH when platform is implemented
+            return Platform.SWITCH
-            # return Platform.SWITCH
-            return Platform.SENSOR
         return Platform.BINARY_SENSOR
 
     return Platform.SENSOR

homeassistant/components/myuplink/switch.py

@@ -0,0 +1,123 @@
+"""Switch entity for myUplink."""
+
+from typing import Any
+
+import aiohttp
+from myuplink import DevicePoint
+
+from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import MyUplinkDataCoordinator
+from .const import DOMAIN
+from .entity import MyUplinkEntity
+from .helpers import find_matching_platform
+
+CATEGORY_BASED_DESCRIPTIONS: dict[str, dict[str, SwitchEntityDescription]] = {
+    "NIBEF": {
+        "50004": SwitchEntityDescription(
+            key="temporary_lux",
+            icon="mdi:water-alert-outline",
+        ),
+    },
+}
+
+
+def get_description(device_point: DevicePoint) -> SwitchEntityDescription | None:
+    """Get description for a device point.
+
+    Priorities:
+    1. Category specific prefix e.g "NIBEF"
+    2. Default to None
+    """
+    prefix, _, _ = device_point.category.partition(" ")
+    description = CATEGORY_BASED_DESCRIPTIONS.get(prefix, {}).get(
+        device_point.parameter_id
+    )
+
+    return description
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up myUplink switch."""
+    entities: list[SwitchEntity] = []
+    coordinator: MyUplinkDataCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+
+    # Setup device point switches
+    for device_id, point_data in coordinator.data.points.items():
+        for point_id, device_point in point_data.items():
+            if find_matching_platform(device_point) == Platform.SWITCH:
+                description = get_description(device_point)
+
+                entities.append(
+                    MyUplinkDevicePointSwitch(
+                        coordinator=coordinator,
+                        device_id=device_id,
+                        device_point=device_point,
+                        entity_description=description,
+                        unique_id_suffix=point_id,
+                    )
+                )
+
+    async_add_entities(entities)
+
+
+class MyUplinkDevicePointSwitch(MyUplinkEntity, SwitchEntity):
+    """Representation of a myUplink device point switch."""
+
+    def __init__(
+        self,
+        coordinator: MyUplinkDataCoordinator,
+        device_id: str,
+        device_point: DevicePoint,
+        entity_description: SwitchEntityDescription | None,
+        unique_id_suffix: str,
+    ) -> None:
+        """Initialize the switch."""
+        super().__init__(
+            coordinator=coordinator,
+            device_id=device_id,
+            unique_id_suffix=unique_id_suffix,
+        )
+
+        # Internal properties
+        self.point_id = device_point.parameter_id
+        self._attr_name = device_point.parameter_name
+
+        if entity_description is not None:
+            self.entity_description = entity_description
+
+    @property
+    def is_on(self) -> bool:
+        """Switch state value."""
+        device_point = self.coordinator.data.points[self.device_id][self.point_id]
+        return int(device_point.value) != 0
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn on switch."""
+        await self._async_turn_switch(1)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off switch."""
+        await self._async_turn_switch(0)
+
+    async def _async_turn_switch(self, mode: int) -> None:
+        """Set switch mode."""
+        try:
+            await self.coordinator.api.async_set_device_points(
+                self.device_id, data={self.point_id: mode}
+            )
+        except aiohttp.ClientError as err:
+            raise HomeAssistantError(
+                f"Failed to set state for {self.entity_id}"
+            ) from err
+
+        await self.coordinator.async_request_refresh()

tests/components/myuplink/__init__.py

@@ -1,4 +1,5 @@
 """Tests for the myuplink integration."""
+
 from homeassistant.core import HomeAssistant
 
 from tests.common import MockConfigEntry

tests/components/myuplink/conftest.py

@@ -1,5 +1,5 @@
 """Test helpers for myuplink."""
-from collections.abc import Generator
+from collections.abc import AsyncGenerator, Generator
 import time
 from typing import Any
 from unittest.mock import MagicMock, patch
@@ -167,3 +167,23 @@ async def init_integration(
     await hass.async_block_till_done()
 
     return mock_config_entry
+
+
+@pytest.fixture
+def platforms() -> list[str]:
+    """Fixture for platforms."""
+    return []
+
+
+@pytest.fixture
+async def setup_platform(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    platforms,
+) -> AsyncGenerator[None, None]:
+    """Set up one or all platforms."""
+
+    with patch(f"homeassistant.components.{DOMAIN}.PLATFORMS", platforms):
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+        yield

tests/components/myuplink/test_switch.py

@@ -0,0 +1,97 @@
+"""Tests for myuplink switch module."""
+
+from unittest.mock import MagicMock
+
+from aiohttp import ClientError
+import pytest
+
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+    STATE_OFF,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+
+TEST_PLATFORM = Platform.SWITCH
+pytestmark = pytest.mark.parametrize("platforms", [(TEST_PLATFORM,)])
+
+ENTITY_ID = "switch.f730_cu_3x400v_temporary_lux"
+ENTITY_FRIENDLY_NAME = "F730 CU 3x400V Tempo­rary lux"
+ENTITY_UID = "batman-r-1234-20240201-123456-aa-bb-cc-dd-ee-ff-50004"
+
+
+async def test_entity_registry(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+) -> None:
+    """Test that the entities are registered in the entity registry."""
+
+    entry = entity_registry.async_get(ENTITY_ID)
+    assert entry.unique_id == ENTITY_UID
+
+
+async def test_attributes(
+    hass: HomeAssistant,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+) -> None:
+    """Test the switch attributes are correct."""
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_OFF
+    assert state.attributes == {
+        "friendly_name": ENTITY_FRIENDLY_NAME,
+        "icon": "mdi:water-alert-outline",
+    }
+
+
+@pytest.mark.parametrize(
+    ("service"),
+    [
+        (SERVICE_TURN_ON),
+        (SERVICE_TURN_OFF),
+    ],
+)
+async def test_switching(
+    hass: HomeAssistant,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+    service: str,
+) -> None:
+    """Test the switch can be turned on/off."""
+
+    await hass.services.async_call(
+        TEST_PLATFORM, service, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+    )
+    await hass.async_block_till_done()
+    mock_myuplink_client.async_set_device_points.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    ("service"),
+    [
+        (SERVICE_TURN_ON),
+        (SERVICE_TURN_OFF),
+    ],
+)
+async def test_api_failure(
+    hass: HomeAssistant,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+    service: str,
+) -> None:
+    """Test handling of exception from API."""
+
+    with pytest.raises(HomeAssistantError):
+        mock_myuplink_client.async_set_device_points.side_effect = ClientError
+        await hass.services.async_call(
+            TEST_PLATFORM, service, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+        )
+        await hass.async_block_till_done()
+        mock_myuplink_client.async_set_device_points.assert_called_once()