jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 05:07:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add exception-translations for switchbot integration (#143444)

Browse Source 
* add exception handler

* add unit test

* test exception per platform

* optimize unit tests

* update quality scale
This commit is contained in:
Retha Runolfsson 2025-05-09 19:09:18 +08:00 committed by GitHub
parent d6e5fdceb7
commit 4271d3f32f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 478 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
homeassistant/components/switchbot/cover.py
View File

@ -21,7 +21,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
from .entity import SwitchbotEntity, exception_handler
# Initialize the logger
_LOGGER = logging.getLogger(__name__)
@ -76,6 +76,7 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
if self._attr_current_cover_position is not None:
self._attr_is_closed = self._attr_current_cover_position <= 20
@exception_handler
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the curtain."""
@ -85,6 +86,7 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@exception_handler
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the curtain."""
@ -94,6 +96,7 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@exception_handler
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the moving of this device."""
@ -103,6 +106,7 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@exception_handler
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover shutter to a specific position."""
position = kwargs.get(ATTR_POSITION)
@ -161,6 +165,7 @@ class SwitchBotBlindTiltEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
_tilt > self.CLOSED_UP_THRESHOLD
)
@exception_handler
async def async_open_cover_tilt(self, **kwargs: Any) -> None:
"""Open the tilt."""
@ -168,6 +173,7 @@ class SwitchBotBlindTiltEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._last_run_success = bool(await self._device.open())
self.async_write_ha_state()
@exception_handler
async def async_close_cover_tilt(self, **kwargs: Any) -> None:
"""Close the tilt."""
@ -175,6 +181,7 @@ class SwitchBotBlindTiltEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._last_run_success = bool(await self._device.close())
self.async_write_ha_state()
@exception_handler
async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
"""Stop the moving of this device."""
@ -182,6 +189,7 @@ class SwitchBotBlindTiltEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._last_run_success = bool(await self._device.stop())
self.async_write_ha_state()
@exception_handler
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
"""Move the cover tilt to a specific position."""
position = kwargs.get(ATTR_TILT_POSITION)
@ -237,6 +245,7 @@ class SwitchBotRollerShadeEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
if self._attr_current_cover_position is not None:
self._attr_is_closed = self._attr_current_cover_position <= 20
@exception_handler
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the roller shade."""
@ -246,6 +255,7 @@ class SwitchBotRollerShadeEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@exception_handler
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the roller shade."""
@ -255,6 +265,7 @@ class SwitchBotRollerShadeEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@exception_handler
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the moving of roller shade."""
@ -264,6 +275,7 @@ class SwitchBotRollerShadeEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@exception_handler
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""

31
homeassistant/components/switchbot/entity.py
View File

@ -2,22 +2,24 @@
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Callable, Coroutine, Mapping
import logging
from typing import Any
from typing import Any, Concatenate
from switchbot import Switchbot, SwitchbotDevice
from switchbot.devices.device import SwitchbotOperationError
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothCoordinatorEntity,
)
from homeassistant.const import ATTR_CONNECTIONS
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import ToggleEntity
from .const import MANUFACTURER
from .const import DOMAIN, MANUFACTURER
from .coordinator import SwitchbotDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -88,11 +90,33 @@ class SwitchbotEntity(
await self._device.update()
def exception_handler[_EntityT: SwitchbotEntity, **_P](
func: Callable[Concatenate[_EntityT, _P], Coroutine[Any, Any, Any]],
) -> Callable[Concatenate[_EntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate Switchbot calls to handle exceptions..
A decorator that wraps the passed in function, catches Switchbot errors.
"""
async def handler(self: _EntityT, *args: _P.args, **kwargs: _P.kwargs) -> None:
try:
await func(self, *args, **kwargs)
except SwitchbotOperationError as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="operation_error",
translation_placeholders={"error": str(error)},
) from error
return handler
class SwitchbotSwitchedEntity(SwitchbotEntity, ToggleEntity):
"""Base class for Switchbot entities that can be turned on and off."""
_device: Switchbot
@exception_handler
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn device on."""
_LOGGER.debug("Turn Switchbot device on %s", self._address)
@ -102,6 +126,7 @@ class SwitchbotSwitchedEntity(SwitchbotEntity, ToggleEntity):
self._attr_is_on = True
self.async_write_ha_state()
@exception_handler
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn device off."""
_LOGGER.debug("Turn Switchbot device off %s", self._address)

4
homeassistant/components/switchbot/humidifier.py
View File

@ -15,7 +15,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SwitchbotConfigEntry
from .entity import SwitchbotSwitchedEntity
from .entity import SwitchbotSwitchedEntity, exception_handler
PARALLEL_UPDATES = 0
@ -55,11 +55,13 @@ class SwitchBotHumidifier(SwitchbotSwitchedEntity, HumidifierEntity):
"""Return the humidity we try to reach."""
return self._device.get_target_humidity()
@exception_handler
async def async_set_humidity(self, humidity: int) -> None:
"""Set new target humidity."""
self._last_run_success = bool(await self._device.set_level(humidity))
self.async_write_ha_state()
@exception_handler
async def async_set_mode(self, mode: str) -> None:
"""Set new target humidity."""
if mode == MODE_AUTO:

9
homeassistant/components/switchbot/light.py
View File

@ -4,7 +4,8 @@ from __future__ import annotations
from typing import Any, cast
from switchbot import ColorMode as SwitchBotColorMode, SwitchbotBaseLight
import switchbot
from switchbot import ColorMode as SwitchBotColorMode
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -17,7 +18,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
from .entity import SwitchbotEntity, exception_handler
SWITCHBOT_COLOR_MODE_TO_HASS = {
SwitchBotColorMode.RGB: ColorMode.RGB,
@ -39,7 +40,7 @@ async def async_setup_entry(
class SwitchbotLightEntity(SwitchbotEntity, LightEntity):
"""Representation of switchbot light bulb."""
_device: SwitchbotBaseLight
_device: switchbot.SwitchbotBaseLight
_attr_name = None
def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
@ -66,6 +67,7 @@ class SwitchbotLightEntity(SwitchbotEntity, LightEntity):
self._attr_rgb_color = device.rgb
self._attr_color_mode = ColorMode.RGB
@exception_handler
async def async_turn_on(self, **kwargs: Any) -> None:
"""Instruct the light to turn on."""
brightness = round(
@ -89,6 +91,7 @@ class SwitchbotLightEntity(SwitchbotEntity, LightEntity):
return
await self._device.turn_on()
@exception_handler
async def async_turn_off(self, **kwargs: Any) -> None:
"""Instruct the light to turn off."""
await self._device.turn_off()

5
homeassistant/components/switchbot/lock.py
View File

@ -11,7 +11,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_LOCK_NIGHTLATCH, DEFAULT_LOCK_NIGHTLATCH
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
from .entity import SwitchbotEntity, exception_handler
PARALLEL_UPDATES = 0
@ -54,11 +54,13 @@ class SwitchBotLock(SwitchbotEntity, LockEntity):
LockStatus.UNLOCKING_STOP,
}
@exception_handler
async def async_lock(self, **kwargs: Any) -> None:
"""Lock the lock."""
self._last_run_success = await self._device.lock()
self.async_write_ha_state()
@exception_handler
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock the lock."""
if self._attr_supported_features & (LockEntityFeature.OPEN):
@ -67,6 +69,7 @@ class SwitchBotLock(SwitchbotEntity, LockEntity):
self._last_run_success = await self._device.unlock()
self.async_write_ha_state()
@exception_handler
async def async_open(self, **kwargs: Any) -> None:
"""Open the lock."""
self._last_run_success = await self._device.unlock()

2
homeassistant/components/switchbot/quality_scale.yaml
View File

@ -70,7 +70,7 @@ rules:
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: todo
exception-translations: done
icon-translations:
status: exempt
comment: |

5
homeassistant/components/switchbot/strings.json
View File

@ -181,5 +181,10 @@
}
}
}
},
"exceptions": {
"operation_error": {
"message": "An error occurred while performing the action: {error}"
}
}
}

158
tests/components/switchbot/test_cover.py
View File

@ -3,6 +3,10 @@
from collections.abc import Callable
from unittest.mock import AsyncMock, patch
import pytest
from switchbot.devices.device import SwitchbotOperationError
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
ATTR_CURRENT_TILT_POSITION,
@ -23,6 +27,7 @@ from homeassistant.const import (
SERVICE_STOP_COVER_TILT,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from . import (
ROLLER_SHADE_SERVICE_INFO,
@ -490,3 +495,156 @@ async def test_roller_shade_controlling(
state = hass.states.get(entity_id)
assert state.state == CoverState.OPEN
assert state.attributes[ATTR_CURRENT_POSITION] == 50
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
SwitchbotOperationError("Operation failed"),
"An error occurred while performing the action: Operation failed",
),
],
)
@pytest.mark.parametrize(
(
"sensor_type",
"service_info",
"class_name",
"service",
"service_data",
"mock_method",
),
[
(
"curtain",
WOCURTAIN3_SERVICE_INFO,
"SwitchbotCurtain",
SERVICE_CLOSE_COVER,
{},
"close",
),
(
"curtain",
WOCURTAIN3_SERVICE_INFO,
"SwitchbotCurtain",
SERVICE_OPEN_COVER,
{},
"open",
),
(
"curtain",
WOCURTAIN3_SERVICE_INFO,
"SwitchbotCurtain",
SERVICE_STOP_COVER,
{},
"stop",
),
(
"curtain",
WOCURTAIN3_SERVICE_INFO,
"SwitchbotCurtain",
SERVICE_SET_COVER_POSITION,
{ATTR_POSITION: 50},
"set_position",
),
(
"roller_shade",
ROLLER_SHADE_SERVICE_INFO,
"SwitchbotRollerShade",
SERVICE_SET_COVER_POSITION,
{ATTR_POSITION: 50},
"set_position",
),
(
"roller_shade",
ROLLER_SHADE_SERVICE_INFO,
"SwitchbotRollerShade",
SERVICE_OPEN_COVER,
{},
"open",
),
(
"roller_shade",
ROLLER_SHADE_SERVICE_INFO,
"SwitchbotRollerShade",
SERVICE_CLOSE_COVER,
{},
"close",
),
(
"roller_shade",
ROLLER_SHADE_SERVICE_INFO,
"SwitchbotRollerShade",
SERVICE_STOP_COVER,
{},
"stop",
),
(
"blind_tilt",
WOBLINDTILT_SERVICE_INFO,
"SwitchbotBlindTilt",
SERVICE_SET_COVER_TILT_POSITION,
{ATTR_TILT_POSITION: 50},
"set_position",
),
(
"blind_tilt",
WOBLINDTILT_SERVICE_INFO,
"SwitchbotBlindTilt",
SERVICE_OPEN_COVER_TILT,
{},
"open",
),
(
"blind_tilt",
WOBLINDTILT_SERVICE_INFO,
"SwitchbotBlindTilt",
SERVICE_CLOSE_COVER_TILT,
{},
"close",
),
(
"blind_tilt",
WOBLINDTILT_SERVICE_INFO,
"SwitchbotBlindTilt",
SERVICE_STOP_COVER_TILT,
{},
"stop",
),
],
)
async def test_exception_handling_cover_service(
hass: HomeAssistant,
mock_entry_factory: Callable[[str], MockConfigEntry],
sensor_type: str,
service_info: BluetoothServiceInfoBleak,
class_name: str,
service: str,
service_data: dict,
mock_method: str,
exception: Exception,
error_message: str,
) -> None:
"""Test exception handling for cover service with exception."""
inject_bluetooth_service_info(hass, service_info)
entry = mock_entry_factory(sensor_type=sensor_type)
entry.add_to_hass(hass)
entity_id = "cover.test_name"
with patch.multiple(
f"homeassistant.components.switchbot.cover.switchbot.{class_name}",
update=AsyncMock(return_value=None),
**{mock_method: AsyncMock(side_effect=exception)},
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
COVER_DOMAIN,
service,
{**service_data, ATTR_ENTITY_ID: entity_id},
blocking=True,
)

52
tests/components/switchbot/test_humidifier.py
View File

@ -4,6 +4,7 @@ from collections.abc import Callable
from unittest.mock import AsyncMock, patch
import pytest
from switchbot.devices.device import SwitchbotOperationError
from homeassistant.components.humidifier import (
ATTR_HUMIDITY,
@ -18,6 +19,7 @@ from homeassistant.components.humidifier import (
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from . import HUMIDIFIER_SERVICE_INFO
@ -121,3 +123,53 @@ async def test_humidifier_services(
}
mock_instance = mock_map[mock_method]
mock_instance.assert_awaited_once_with(*expected_args)
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
SwitchbotOperationError("Operation failed"),
"An error occurred while performing the action: Operation failed",
),
],
)
@pytest.mark.parametrize(
("service", "service_data", "mock_method"),
[
(SERVICE_TURN_ON, {}, "turn_on"),
(SERVICE_TURN_OFF, {}, "turn_off"),
(SERVICE_SET_HUMIDITY, {ATTR_HUMIDITY: 60}, "set_level"),
(SERVICE_SET_MODE, {ATTR_MODE: MODE_AUTO}, "async_set_auto"),
(SERVICE_SET_MODE, {ATTR_MODE: MODE_NORMAL}, "async_set_manual"),
],
)
async def test_exception_handling_humidifier_service(
hass: HomeAssistant,
mock_entry_factory: Callable[[str], MockConfigEntry],
service: str,
service_data: dict,
mock_method: str,
exception: Exception,
error_message: str,
) -> None:
"""Test exception handling for humidifier service with exception."""
inject_bluetooth_service_info(hass, HUMIDIFIER_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="humidifier")
entry.add_to_hass(hass)
entity_id = "humidifier.test_name"
patch_target = f"homeassistant.components.switchbot.humidifier.switchbot.SwitchbotHumidifier.{mock_method}"
with patch(patch_target, new=AsyncMock(side_effect=exception)):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
HUMIDIFIER_DOMAIN,
service,
{**service_data, ATTR_ENTITY_ID: entity_id},
blocking=True,
)

130
tests/components/switchbot/test_light.py
View File

@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, patch
import pytest
from switchbot import ColorMode as switchbotColorMode
from switchbot.devices.device import SwitchbotOperationError
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -17,6 +18,7 @@ from homeassistant.components.light import (
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from . import WOSTRIP_SERVICE_INFO
@ -93,30 +95,14 @@ async def test_light_strip_services(
entry.add_to_hass(hass)
entity_id = "light.test_name"
with (
patch("switchbot.SwitchbotLightStrip.color_modes", new=color_modes),
patch("switchbot.SwitchbotLightStrip.color_mode", new=color_mode),
patch(
"switchbot.SwitchbotLightStrip.turn_on",
new=AsyncMock(return_value=True),
) as mock_turn_on,
patch(
"switchbot.SwitchbotLightStrip.turn_off",
new=AsyncMock(return_value=True),
) as mock_turn_off,
patch(
"switchbot.SwitchbotLightStrip.set_brightness",
new=AsyncMock(return_value=True),
) as mock_set_brightness,
patch(
"switchbot.SwitchbotLightStrip.set_rgb",
new=AsyncMock(return_value=True),
) as mock_set_rgb,
patch(
"switchbot.SwitchbotLightStrip.set_color_temp",
new=AsyncMock(return_value=True),
) as mock_set_color_temp,
patch("switchbot.SwitchbotLightStrip.update", new=AsyncMock(return_value=None)),
mocked_instance = AsyncMock(return_value=True)
with patch.multiple(
"homeassistant.components.switchbot.light.switchbot.SwitchbotLightStrip",
color_modes=color_modes,
color_mode=color_mode,
update=AsyncMock(return_value=None),
**{mock_method: mocked_instance},
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
@ -128,12 +114,90 @@ async def test_light_strip_services(
blocking=True,
)
mock_map = {
"turn_off": mock_turn_off,
"turn_on": mock_turn_on,
"set_brightness": mock_set_brightness,
"set_rgb": mock_set_rgb,
"set_color_temp": mock_set_color_temp,
}
mock_instance = mock_map[mock_method]
mock_instance.assert_awaited_once_with(*expected_args)
mocked_instance.assert_awaited_once_with(*expected_args)
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
SwitchbotOperationError("Operation failed"),
"An error occurred while performing the action: Operation failed",
),
],
)
@pytest.mark.parametrize(
("service", "service_data", "mock_method", "color_modes", "color_mode"),
[
(
SERVICE_TURN_ON,
{},
"turn_on",
{switchbotColorMode.RGB},
switchbotColorMode.RGB,
),
(
SERVICE_TURN_OFF,
{},
"turn_off",
{switchbotColorMode.RGB},
switchbotColorMode.RGB,
),
(
SERVICE_TURN_ON,
{ATTR_BRIGHTNESS: 128},
"set_brightness",
{switchbotColorMode.RGB},
switchbotColorMode.RGB,
),
(
SERVICE_TURN_ON,
{ATTR_RGB_COLOR: (255, 0, 0)},
"set_rgb",
{switchbotColorMode.RGB},
switchbotColorMode.RGB,
),
(
SERVICE_TURN_ON,
{ATTR_COLOR_TEMP_KELVIN: 4000},
"set_color_temp",
{switchbotColorMode.COLOR_TEMP},
switchbotColorMode.COLOR_TEMP,
),
],
)
async def test_exception_handling_light_service(
hass: HomeAssistant,
mock_entry_factory: Callable[[str], MockConfigEntry],
service: str,
service_data: dict,
mock_method: str,
color_modes: set | None,
color_mode: switchbotColorMode | None,
exception: Exception,
error_message: str,
) -> None:
"""Test exception handling for light service with exception."""
inject_bluetooth_service_info(hass, WOSTRIP_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="light_strip")
entry.add_to_hass(hass)
entity_id = "light.test_name"
with patch.multiple(
"homeassistant.components.switchbot.light.switchbot.SwitchbotLightStrip",
color_modes=color_modes,
color_mode=color_mode,
update=AsyncMock(return_value=None),
**{mock_method: AsyncMock(side_effect=exception)},
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
LIGHT_DOMAIN,
service,
{**service_data, ATTR_ENTITY_ID: entity_id},
blocking=True,
)

51
tests/components/switchbot/test_lock.py
View File

@ -4,6 +4,7 @@ from collections.abc import Callable
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from switchbot.devices.device import SwitchbotOperationError
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN
@ -14,6 +15,7 @@ from homeassistant.const import (
SERVICE_UNLOCK,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from . import LOCK_SERVICE_INFO, WOLOCKPRO_SERVICE_INFO
@ -103,3 +105,52 @@ async def test_lock_services_with_night_latch_enabled(
)
mocked_instance.assert_awaited_once()
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
SwitchbotOperationError("Operation failed"),
"An error occurred while performing the action: Operation failed",
),
],
)
@pytest.mark.parametrize(
("service", "mock_method"),
[
(SERVICE_LOCK, "lock"),
(SERVICE_OPEN, "unlock"),
(SERVICE_UNLOCK, "unlock_without_unlatch"),
],
)
async def test_exception_handling_lock_service(
hass: HomeAssistant,
mock_entry_encrypted_factory: Callable[[str], MockConfigEntry],
service: str,
mock_method: str,
exception: Exception,
error_message: str,
) -> None:
"""Test exception handling for lock service with exception."""
inject_bluetooth_service_info(hass, LOCK_SERVICE_INFO)
entry = mock_entry_encrypted_factory(sensor_type="lock")
entry.add_to_hass(hass)
entity_id = "lock.test_name"
with patch.multiple(
"homeassistant.components.switchbot.lock.switchbot.SwitchbotLock",
is_night_latch_enabled=MagicMock(return_value=True),
**{mock_method: AsyncMock(side_effect=exception)},
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
LOCK_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)

62
tests/components/switchbot/test_switch.py
View File

@ -1,10 +1,20 @@
"""Test the switchbot switches."""
from collections.abc import Callable
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from homeassistant.components.switch import STATE_ON
import pytest
from switchbot.devices.device import SwitchbotOperationError
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_ON,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from . import WOHAND_SERVICE_INFO
@ -45,3 +55,51 @@ async def test_switchbot_switch_with_restore_state(
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes["last_run_success"] is True
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
SwitchbotOperationError("Operation failed"),
"An error occurred while performing the action: Operation failed",
),
],
)
@pytest.mark.parametrize(
("service", "mock_method"),
[
(SERVICE_TURN_ON, "turn_on"),
(SERVICE_TURN_OFF, "turn_off"),
],
)
async def test_exception_handling_switch(
hass: HomeAssistant,
mock_entry_factory: Callable[[str], MockConfigEntry],
service: str,
mock_method: str,
exception: Exception,
error_message: str,
) -> None:
"""Test exception handling for switch service with exception."""
inject_bluetooth_service_info(hass, WOHAND_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="bot")
entry.add_to_hass(hass)
entity_id = "switch.test_name"
patch_target = (
f"homeassistant.components.switchbot.switch.switchbot.Switchbot.{mock_method}"
)
with patch(patch_target, new=AsyncMock(side_effect=exception)):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)