Add button error handling for Peblar Rocksolid EV Chargers (#133802)

homeassistant/components/peblar/button.py

@@ -19,6 +19,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .coordinator import PeblarConfigEntry, PeblarUserConfigurationDataUpdateCoordinator
 from .entity import PeblarEntity
+from .helpers import peblar_exception_handler
 
 PARALLEL_UPDATES = 1
 
@@ -72,6 +73,7 @@ class PeblarButtonEntity(
 
     entity_description: PeblarButtonEntityDescription
 
+    @peblar_exception_handler
     async def async_press(self) -> None:
         """Trigger button press on the Peblar device."""
         await self.entity_description.press_fn(self.coordinator.peblar)

homeassistant/components/peblar/helpers.py

@@ -0,0 +1,55 @@
+"""Helpers for Peblar."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Coroutine
+from typing import Any, Concatenate
+
+from peblar import PeblarAuthenticationError, PeblarConnectionError, PeblarError
+
+from homeassistant.exceptions import HomeAssistantError
+
+from .const import DOMAIN
+from .entity import PeblarEntity
+
+
+def peblar_exception_handler[_PeblarEntityT: PeblarEntity, **_P](
+    func: Callable[Concatenate[_PeblarEntityT, _P], Coroutine[Any, Any, Any]],
+) -> Callable[Concatenate[_PeblarEntityT, _P], Coroutine[Any, Any, None]]:
+    """Decorate Peblar calls to handle exceptions.
+
+    A decorator that wraps the passed in function, catches Peblar errors.
+    """
+
+    async def handler(
+        self: _PeblarEntityT, *args: _P.args, **kwargs: _P.kwargs
+    ) -> None:
+        try:
+            await func(self, *args, **kwargs)
+            self.coordinator.async_update_listeners()
+
+        except PeblarAuthenticationError as error:
+            # Reload the config entry to trigger reauth flow
+            self.hass.config_entries.async_schedule_reload(
+                self.coordinator.config_entry.entry_id
+            )
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="authentication_error",
+            ) from error
+
+        except PeblarConnectionError as error:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="communication_error",
+                translation_placeholders={"error": str(error)},
+            ) from error
+
+        except PeblarError as error:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="unknown_error",
+                translation_placeholders={"error": str(error)},
+            ) from error
+
+    return handler

homeassistant/components/peblar/strings.json

@@ -161,5 +161,16 @@
         "name": "Customization"
       }
     }
+  },
+  "exceptions": {
+    "authentication_error": {
+      "message": "An authentication failure occurred while communicating with the Peblar device."
+    },
+    "communication_error": {
+      "message": "An error occurred while communicating with the Peblar device: {error}"
+    },
+    "unknown_error": {
+      "message": "An unknown error occurred while communicating with the Peblar device: {error}"
+    }
   }
 }

tests/components/peblar/test_button.py

@@ -1,19 +1,29 @@
 """Tests for the Peblar button platform."""
 
+from unittest.mock import MagicMock
+
+from peblar import PeblarAuthenticationError, PeblarConnectionError, PeblarError
 import pytest
 from syrupy.assertion import SnapshotAssertion
 
+from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
 from homeassistant.components.peblar.const import DOMAIN
-from homeassistant.const import Platform
+from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
+from homeassistant.const import ATTR_ENTITY_ID, Platform
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 
 from tests.common import MockConfigEntry, snapshot_platform
 
+pytestmark = [
+    pytest.mark.freeze_time("2024-12-21 21:45:00"),
+    pytest.mark.parametrize("init_integration", [Platform.BUTTON], indirect=True),
+    pytest.mark.usefixtures("init_integration"),
+]
 
-@pytest.mark.freeze_time("2024-12-21 21:45:00")
-@pytest.mark.parametrize("init_integration", [Platform.BUTTON], indirect=True)
-@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
 async def test_entities(
     hass: HomeAssistant,
     snapshot: SnapshotAssertion,
@@ -34,3 +44,110 @@ async def test_entities(
     )
     for entity_entry in entity_entries:
         assert entity_entry.device_id == device_entry.id
+
+
+@pytest.mark.parametrize(
+    ("entity_id", "method"),
+    [
+        ("button.peblar_ev_charger_identify", "identify"),
+        ("button.peblar_ev_charger_restart", "reboot"),
+    ],
+)
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_buttons(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    mock_config_entry: MockConfigEntry,
+    entity_id: str,
+    method: str,
+) -> None:
+    """Test the Peblar EV charger buttons."""
+    mocked_method = getattr(mock_peblar, method)
+
+    # Test normal happy path button press
+    await hass.services.async_call(
+        BUTTON_DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+
+    assert len(mocked_method.mock_calls) == 1
+    mocked_method.assert_called_with()
+
+    # Test connection error handling
+    mocked_method.side_effect = PeblarConnectionError("Could not connect")
+    with pytest.raises(
+        HomeAssistantError,
+        match=(
+            r"An error occurred while communicating "
+            r"with the Peblar device: Could not connect"
+        ),
+    ) as excinfo:
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+
+    assert excinfo.value.translation_domain == DOMAIN
+    assert excinfo.value.translation_key == "communication_error"
+    assert excinfo.value.translation_placeholders == {"error": "Could not connect"}
+
+    # Test unknown error handling
+    mocked_method.side_effect = PeblarError("Unknown error")
+    with pytest.raises(
+        HomeAssistantError,
+        match=(
+            r"An unknown error occurred while communicating "
+            r"with the Peblar device: Unknown error"
+        ),
+    ) as excinfo:
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+
+    assert excinfo.value.translation_domain == DOMAIN
+    assert excinfo.value.translation_key == "unknown_error"
+    assert excinfo.value.translation_placeholders == {"error": "Unknown error"}
+
+    # Test authentication error handling
+    mocked_method.side_effect = PeblarAuthenticationError("Authentication error")
+    mock_peblar.login.side_effect = PeblarAuthenticationError("Authentication error")
+    with pytest.raises(
+        HomeAssistantError,
+        match=(
+            r"An authentication failure occurred while communicating "
+            r"with the Peblar device"
+        ),
+    ) as excinfo:
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+
+    assert excinfo.value.translation_domain == DOMAIN
+    assert excinfo.value.translation_key == "authentication_error"
+    assert not excinfo.value.translation_placeholders
+
+    # Ensure the device is reloaded on authentication error and triggers
+    # a reauthentication flow.
+    await hass.async_block_till_done()
+    assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+
+    flow = flows[0]
+    assert flow["step_id"] == "reauth_confirm"
+    assert flow["handler"] == DOMAIN
+
+    assert "context" in flow
+    assert flow["context"].get("source") == SOURCE_REAUTH
+    assert flow["context"].get("entry_id") == mock_config_entry.entry_id