Fetch power off state for Home Connect appliances' power switch (#129289)

2025-04-23 16:57:53 +00:00 · 2024-10-29 13:56:45 +01:00 · 2024-10-29 13:56:45 +01:00 · f194a689cc
commit f194a689cc
parent a36b350954
3 changed files with 217 additions and 48 deletions
--- a/homeassistant/components/home_connect/strings.json
+++ b/homeassistant/components/home_connect/strings.json
@ -63,6 +63,9 @@
    },
    "turn_off_not_supported": {
      "message": "{appliance_name} does not support turning off or entering standby mode."
+    },
+    "unable_to_retrieve_turn_off": {
+      "message": "Unable to turn off {appliance_name} because its support for turning off or entering standby mode could not be determined."
    }
  },
  "issues": {
--- a/homeassistant/components/home_connect/switch.py
+++ b/homeassistant/components/home_connect/switch.py
@ -15,6 +15,8 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from . import get_dict_from_home_connect_error
 from .api import ConfigEntryAuth
 from .const import (
+    ATTR_ALLOWED_VALUES,
+    ATTR_CONSTRAINTS,
    ATTR_VALUE,
    BSH_ACTIVE_PROGRAM,
    BSH_CHILD_LOCK_STATE,
@ -268,19 +270,18 @@ class HomeConnectPowerSwitch(HomeConnectEntity, SwitchEntity):
            device,
            SwitchEntityDescription(key=BSH_POWER_STATE, translation_key="power"),
        )
-        match device.appliance.type:
-            case "Dishwasher" | "Cooktop" | "Hood":
-                self.power_off_state = BSH_POWER_OFF
-            case (
-                "Oven"
-                | "WarmDrawer"
-                | "CoffeeMachine"
-                | "CleaningRobot"
-                | "CookProcessor"
-            ):
-                self.power_off_state = BSH_POWER_STANDBY
-            case _:
-                self.power_off_state = None
+        if (
+            power_state := device.appliance.status.get(BSH_POWER_STATE, {}).get(
+                ATTR_VALUE
+            )
+        ) and power_state in [BSH_POWER_OFF, BSH_POWER_STANDBY]:
+            self.power_off_state = power_state
+
+    async def async_added_to_hass(self) -> None:
+        """Add the entity to the hass instance."""
+        await super().async_added_to_hass()
+        if not hasattr(self, "power_off_state"):
+            await self.async_fetch_power_off_state()

    async def async_turn_on(self, **kwargs: Any) -> None:
        """Switch the device on."""
@ -303,6 +304,15 @@ class HomeConnectPowerSwitch(HomeConnectEntity, SwitchEntity):

    async def async_turn_off(self, **kwargs: Any) -> None:
        """Switch the device off."""
+        if not hasattr(self, "power_off_state"):
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="unable_to_retrieve_turn_off",
+                translation_placeholders={
+                    SVE_TRANSLATION_PLACEHOLDER_APPLIANCE_NAME: self.device.appliance.name
+                },
+            )
+
        if self.power_off_state is None:
            raise ServiceValidationError(
                translation_domain=DOMAIN,
@ -339,7 +349,8 @@ class HomeConnectPowerSwitch(HomeConnectEntity, SwitchEntity):
        ):
            self._attr_is_on = True
        elif (
-            self.device.appliance.status.get(BSH_POWER_STATE, {}).get(ATTR_VALUE)
+            hasattr(self, "power_off_state")
+            and self.device.appliance.status.get(BSH_POWER_STATE, {}).get(ATTR_VALUE)
            == self.power_off_state
        ):
            self._attr_is_on = False
@ -363,3 +374,24 @@ class HomeConnectPowerSwitch(HomeConnectEntity, SwitchEntity):
        else:
            self._attr_is_on = None
        _LOGGER.debug("Updated, new state: %s", self._attr_is_on)
+
+    async def async_fetch_power_off_state(self) -> None:
+        """Fetch the power off state."""
+        try:
+            data = await self.hass.async_add_executor_job(
+                self.device.appliance.get, f"/settings/{self.bsh_key}"
+            )
+        except HomeConnectError as err:
+            _LOGGER.error("An error occurred: %s", err)
+            return
+        if not data or not (
+            allowed_values := data.get(ATTR_CONSTRAINTS, {}).get(ATTR_ALLOWED_VALUES)
+        ):
+            return
+
+        if BSH_POWER_OFF in allowed_values:
+            self.power_off_state = BSH_POWER_OFF
+        elif BSH_POWER_STANDBY in allowed_values:
+            self.power_off_state = BSH_POWER_STANDBY
+        else:
+            self.power_off_state = None
--- a/tests/components/home_connect/test_switch.py
+++ b/tests/components/home_connect/test_switch.py
@ -7,11 +7,14 @@ from homeconnect.api import HomeConnectAppliance, HomeConnectError
 import pytest

 from homeassistant.components.home_connect.const import (
+    ATTR_ALLOWED_VALUES,
+    ATTR_CONSTRAINTS,
    BSH_ACTIVE_PROGRAM,
    BSH_CHILD_LOCK_STATE,
    BSH_OPERATION_STATE,
    BSH_POWER_OFF,
    BSH_POWER_ON,
+    BSH_POWER_STANDBY,
    BSH_POWER_STATE,
    REFRIGERATION_SUPERMODEFREEZER,
 )
@ -81,32 +84,6 @@ async def test_switches(
            STATE_OFF,
            "Dishwasher",
        ),
-        (
-            "switch.dishwasher_power",
-            {BSH_POWER_STATE: {"value": BSH_POWER_ON}},
-            SERVICE_TURN_ON,
-            STATE_ON,
-            "Dishwasher",
-        ),
-        (
-            "switch.dishwasher_power",
-            {BSH_POWER_STATE: {"value": BSH_POWER_OFF}},
-            SERVICE_TURN_OFF,
-            STATE_OFF,
-            "Dishwasher",
-        ),
-        (
-            "switch.dishwasher_power",
-            {
-                BSH_POWER_STATE: {"value": ""},
-                BSH_OPERATION_STATE: {
-                    "value": "BSH.Common.EnumType.OperationState.Inactive"
-                },
-            },
-            SERVICE_TURN_OFF,
-            STATE_OFF,
-            "Dishwasher",
-        ),
        (
            "switch.dishwasher_child_lock",
            {BSH_CHILD_LOCK_STATE: {"value": True}},
@ -179,6 +156,14 @@ async def test_switch_functionality(
            "Dishwasher",
            r"Error.*stop.*program.*",
        ),
+        (
+            "switch.dishwasher_power",
+            {BSH_POWER_STATE: {"value": BSH_POWER_OFF}},
+            SERVICE_TURN_OFF,
+            "set_setting",
+            "Dishwasher",
+            r"Error.*turn.*off.*appliance.*value",
+        ),
        (
            "switch.dishwasher_power",
            {BSH_POWER_STATE: {"value": ""}},
@ -187,14 +172,6 @@ async def test_switch_functionality(
            "Dishwasher",
            r"Error.*turn.*on.*appliance.*",
        ),
-        (
-            "switch.dishwasher_power",
-            {BSH_POWER_STATE: {"value": ""}},
-            SERVICE_TURN_OFF,
-            "set_setting",
-            "Dishwasher",
-            r"Error.*turn.*off.*appliance.*value.*",
-        ),
        (
            "switch.dishwasher_child_lock",
            {BSH_CHILD_LOCK_STATE: {"value": ""}},
@ -372,3 +349,160 @@ async def test_ent_desc_switch_exception_handling(
            SWITCH_DOMAIN, service, {ATTR_ENTITY_ID: entity_id}, blocking=True
        )
    assert getattr(problematic_appliance, mock_attr).call_count == 2
+
+
+@pytest.mark.parametrize(
+    ("entity_id", "status", "allowed_values", "service", "power_state", "appliance"),
+    [
+        (
+            "switch.dishwasher_power",
+            {BSH_POWER_STATE: {"value": BSH_POWER_ON}},
+            [BSH_POWER_ON, BSH_POWER_OFF],
+            SERVICE_TURN_ON,
+            STATE_ON,
+            "Dishwasher",
+        ),
+        (
+            "switch.dishwasher_power",
+            {BSH_POWER_STATE: {"value": BSH_POWER_OFF}},
+            [BSH_POWER_ON, BSH_POWER_OFF],
+            SERVICE_TURN_OFF,
+            STATE_OFF,
+            "Dishwasher",
+        ),
+        (
+            "switch.dishwasher_power",
+            {
+                BSH_POWER_STATE: {"value": ""},
+                BSH_OPERATION_STATE: {
+                    "value": "BSH.Common.EnumType.OperationState.Run"
+                },
+            },
+            [BSH_POWER_ON],
+            SERVICE_TURN_ON,
+            STATE_ON,
+            "Dishwasher",
+        ),
+        (
+            "switch.dishwasher_power",
+            {
+                BSH_POWER_STATE: {"value": ""},
+                BSH_OPERATION_STATE: {
+                    "value": "BSH.Common.EnumType.OperationState.Inactive"
+                },
+            },
+            [BSH_POWER_ON],
+            SERVICE_TURN_ON,
+            STATE_OFF,
+            "Dishwasher",
+        ),
+        (
+            "switch.dishwasher_power",
+            {BSH_POWER_STATE: {"value": BSH_POWER_ON}},
+            [BSH_POWER_ON, BSH_POWER_STANDBY],
+            SERVICE_TURN_ON,
+            STATE_ON,
+            "Dishwasher",
+        ),
+        (
+            "switch.dishwasher_power",
+            {BSH_POWER_STATE: {"value": BSH_POWER_STANDBY}},
+            [BSH_POWER_ON, BSH_POWER_STANDBY],
+            SERVICE_TURN_OFF,
+            STATE_OFF,
+            "Dishwasher",
+        ),
+    ],
+    indirect=["appliance"],
+)
+@pytest.mark.usefixtures("bypass_throttle")
+async def test_power_swtich(
+    entity_id: str,
+    status: dict,
+    allowed_values: list[str],
+    service: str,
+    power_state: str,
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    integration_setup: Callable[[], Awaitable[bool]],
+    setup_credentials: None,
+    appliance: Mock,
+    get_appliances: MagicMock,
+) -> None:
+    """Test power switch functionality."""
+    appliance.get.side_effect = [
+        {
+            ATTR_CONSTRAINTS: {
+                ATTR_ALLOWED_VALUES: allowed_values,
+            },
+        }
+    ]
+    appliance.status.update(SETTINGS_STATUS)
+    appliance.status.update(status)
+    get_appliances.return_value = [appliance]
+
+    assert config_entry.state == ConfigEntryState.NOT_LOADED
+    assert await integration_setup()
+    assert config_entry.state == ConfigEntryState.LOADED
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN, service, {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    assert hass.states.is_state(entity_id, power_state)
+
+
+@pytest.mark.parametrize(
+    ("entity_id", "allowed_values", "service", "appliance", "exception_match"),
+    [
+        (
+            "switch.dishwasher_power",
+            [BSH_POWER_ON],
+            SERVICE_TURN_OFF,
+            "Dishwasher",
+            r".*not support.*turn.*off.*",
+        ),
+        (
+            "switch.dishwasher_power",
+            None,
+            SERVICE_TURN_OFF,
+            "Dishwasher",
+            r".*Unable.*turn.*off.*support.*not.*determined.*",
+        ),
+    ],
+    indirect=["appliance"],
+)
+@pytest.mark.usefixtures("bypass_throttle")
+async def test_power_switch_service_validation_errors(
+    entity_id: str,
+    allowed_values: list[str],
+    service: str,
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    integration_setup: Callable[[], Awaitable[bool]],
+    setup_credentials: None,
+    appliance: Mock,
+    exception_match: str,
+    get_appliances: MagicMock,
+) -> None:
+    """Test power switch functionality validation errors."""
+    if allowed_values:
+        appliance.get.side_effect = [
+            {
+                ATTR_CONSTRAINTS: {
+                    ATTR_ALLOWED_VALUES: allowed_values,
+                },
+            }
+        ]
+    appliance.status.update(SETTINGS_STATUS)
+    get_appliances.return_value = [appliance]
+
+    assert config_entry.state == ConfigEntryState.NOT_LOADED
+    assert await integration_setup()
+    assert config_entry.state == ConfigEntryState.LOADED
+
+    appliance.status.update({BSH_POWER_STATE: {"value": BSH_POWER_ON}})
+
+    with pytest.raises(ServiceValidationError, match=exception_match):
+        await hass.services.async_call(
+            SWITCH_DOMAIN, service, {"entity_id": entity_id}, blocking=True
+        )