From 39b025dfead026ea04de8141b6116a1d5424d96c Mon Sep 17 00:00:00 2001 From: wollew Date: Thu, 8 Jan 2026 00:06:26 +0100 Subject: [PATCH] catch and wrap exceptions when doing pyvlx actions in velux entities (#160430) Co-authored-by: Joostlek --- homeassistant/components/velux/cover.py | 10 +++++- homeassistant/components/velux/entity.py | 33 ++++++++++++++++-- homeassistant/components/velux/light.py | 4 ++- .../components/velux/quality_scale.yaml | 2 +- homeassistant/components/velux/scene.py | 2 ++ homeassistant/components/velux/strings.json | 3 ++ tests/components/velux/test_cover.py | 34 +++++++++++++++++++ 7 files changed, 83 insertions(+), 5 deletions(-) diff --git a/homeassistant/components/velux/cover.py b/homeassistant/components/velux/cover.py index 78fd263e4ec..e287e795390 100644 --- a/homeassistant/components/velux/cover.py +++ b/homeassistant/components/velux/cover.py @@ -25,7 +25,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from . import VeluxConfigEntry -from .entity import VeluxEntity +from .entity import VeluxEntity, wrap_pyvlx_call_exceptions PARALLEL_UPDATES = 1 @@ -110,14 +110,17 @@ class VeluxCover(VeluxEntity, CoverEntity): """Return if the cover is closing or not.""" return self.node.is_closing + @wrap_pyvlx_call_exceptions async def async_close_cover(self, **kwargs: Any) -> None: """Close the cover.""" await self.node.close(wait_for_completion=False) + @wrap_pyvlx_call_exceptions async def async_open_cover(self, **kwargs: Any) -> None: """Open the cover.""" await self.node.open(wait_for_completion=False) + @wrap_pyvlx_call_exceptions async def async_set_cover_position(self, **kwargs: Any) -> None: """Move the cover to a specific position.""" position_percent = 100 - kwargs[ATTR_POSITION] @@ -126,22 +129,27 @@ class VeluxCover(VeluxEntity, CoverEntity): Position(position_percent=position_percent), wait_for_completion=False ) + @wrap_pyvlx_call_exceptions async def async_stop_cover(self, **kwargs: Any) -> None: """Stop the cover.""" await self.node.stop(wait_for_completion=False) + @wrap_pyvlx_call_exceptions async def async_close_cover_tilt(self, **kwargs: Any) -> None: """Close cover tilt.""" await cast(Blind, self.node).close_orientation(wait_for_completion=False) + @wrap_pyvlx_call_exceptions async def async_open_cover_tilt(self, **kwargs: Any) -> None: """Open cover tilt.""" await cast(Blind, self.node).open_orientation(wait_for_completion=False) + @wrap_pyvlx_call_exceptions async def async_stop_cover_tilt(self, **kwargs: Any) -> None: """Stop cover tilt.""" await cast(Blind, self.node).stop_orientation(wait_for_completion=False) + @wrap_pyvlx_call_exceptions async def async_set_cover_tilt_position(self, **kwargs: Any) -> None: """Move cover tilt to a specific position.""" position_percent = 100 - kwargs[ATTR_TILT_POSITION] diff --git a/homeassistant/components/velux/entity.py b/homeassistant/components/velux/entity.py index cdaea35eac1..2a50d923281 100644 --- a/homeassistant/components/velux/entity.py +++ b/homeassistant/components/velux/entity.py @@ -1,10 +1,13 @@ """Support for VELUX KLF 200 devices.""" -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Coroutine +from functools import wraps import logging +from typing import Any, ParamSpec -from pyvlx import Node +from pyvlx import Node, PyVLXException +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import Entity @@ -12,6 +15,32 @@ from .const import DOMAIN _LOGGER = logging.getLogger(__name__) +P = ParamSpec("P") + + +def wrap_pyvlx_call_exceptions( + func: Callable[P, Coroutine[Any, Any, None]], +) -> Callable[P, Coroutine[Any, Any, None]]: + """Decorate pyvlx calls to handle exceptions. + + Catches OSError and PyVLXException and wraps them into HomeAssistantError + with translation support. + """ + + @wraps(func) + async def wrapper(*args: P.args, **kwargs: P.kwargs) -> None: + """Wrap async function to catch exceptions thrown in pyvlx calls.""" + try: + await func(*args, **kwargs) + except (OSError, PyVLXException) as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="device_communication_error", + translation_placeholders={"error": str(err)}, + ) from err + + return wrapper + class VeluxEntity(Entity): """Abstraction for all Velux entities.""" diff --git a/homeassistant/components/velux/light.py b/homeassistant/components/velux/light.py index f6b6eaecce0..3775b8a8797 100644 --- a/homeassistant/components/velux/light.py +++ b/homeassistant/components/velux/light.py @@ -11,7 +11,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from . import VeluxConfigEntry -from .entity import VeluxEntity +from .entity import VeluxEntity, wrap_pyvlx_call_exceptions PARALLEL_UPDATES = 1 @@ -49,6 +49,7 @@ class VeluxLight(VeluxEntity, LightEntity): """Return true if light is on.""" return not self.node.intensity.off and self.node.intensity.known + @wrap_pyvlx_call_exceptions async def async_turn_on(self, **kwargs: Any) -> None: """Instruct the light to turn on.""" if ATTR_BRIGHTNESS in kwargs: @@ -60,6 +61,7 @@ class VeluxLight(VeluxEntity, LightEntity): else: await self.node.turn_on(wait_for_completion=True) + @wrap_pyvlx_call_exceptions async def async_turn_off(self, **kwargs: Any) -> None: """Instruct the light to turn off.""" await self.node.turn_off(wait_for_completion=True) diff --git a/homeassistant/components/velux/quality_scale.yaml b/homeassistant/components/velux/quality_scale.yaml index 0bc3bca2abc..885a80d7670 100644 --- a/homeassistant/components/velux/quality_scale.yaml +++ b/homeassistant/components/velux/quality_scale.yaml @@ -22,7 +22,7 @@ rules: unique-config-entry: done # Silver - action-exceptions: todo + action-exceptions: done config-entry-unloading: done docs-configuration-parameters: todo docs-installation-parameters: todo diff --git a/homeassistant/components/velux/scene.py b/homeassistant/components/velux/scene.py index 93a39752b32..c2c5250517b 100644 --- a/homeassistant/components/velux/scene.py +++ b/homeassistant/components/velux/scene.py @@ -13,6 +13,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from . import VeluxConfigEntry from .const import DOMAIN +from .entity import wrap_pyvlx_call_exceptions PARALLEL_UPDATES = 1 @@ -51,6 +52,7 @@ class VeluxScene(Scene): identifiers={(DOMAIN, f"gateway_{config_entry_id}")}, ) + @wrap_pyvlx_call_exceptions async def async_activate(self, **kwargs: Any) -> None: """Activate the scene.""" await self.scene.run(wait_for_completion=False) diff --git a/homeassistant/components/velux/strings.json b/homeassistant/components/velux/strings.json index 84857dc0ac7..13abb8a0f78 100644 --- a/homeassistant/components/velux/strings.json +++ b/homeassistant/components/velux/strings.json @@ -48,6 +48,9 @@ } }, "exceptions": { + "device_communication_error": { + "message": "Failed to communicate with Velux device: {error}" + }, "no_gateway_loaded": { "message": "No loaded Velux gateway found" }, diff --git a/tests/components/velux/test_cover.py b/tests/components/velux/test_cover.py index b2635407c2a..39df7d25ad0 100644 --- a/tests/components/velux/test_cover.py +++ b/tests/components/velux/test_cover.py @@ -3,6 +3,7 @@ from unittest.mock import AsyncMock import pytest +from pyvlx.exception import PyVLXException from pyvlx.opening_device import Awning, GarageDoor, Gate, RollerShutter, Window from homeassistant.components.cover import ( @@ -20,6 +21,7 @@ from homeassistant.components.cover import ( ) from homeassistant.components.velux import DOMAIN from homeassistant.const import ( + ATTR_ENTITY_ID, STATE_CLOSED, STATE_CLOSING, STATE_OPEN, @@ -27,6 +29,7 @@ from homeassistant.const import ( Platform, ) from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import device_registry as dr, entity_registry as er from . import update_callback_entity @@ -296,3 +299,34 @@ async def test_non_blind_has_no_tilt_position( state = hass.states.get(entity_id) assert state is not None assert "current_tilt_position" not in state.attributes + + +# Exception handling tests + + +@pytest.mark.parametrize( + "exception", + [PyVLXException("PyVLX error"), OSError("OS error")], +) +async def test_cover_command_exception_handling( + hass: HomeAssistant, + mock_window: AsyncMock, + exception: Exception, +) -> None: + """Test that exceptions from node commands are wrapped in HomeAssistantError.""" + + entity_id = "cover.test_window" + + # Make the close method raise an exception + mock_window.close.side_effect = exception + + with pytest.raises( + HomeAssistantError, + match="Failed to communicate with Velux device", + ): + await hass.services.async_call( + COVER_DOMAIN, + SERVICE_CLOSE_COVER, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + )