evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Wrap tplink exceptions caused by user actions inside HomeAssistantError (#114919)

Browse source
This commit is contained in:
Steven B 2024-04-09 09:35:25 +01:00 • committed by GitHub
parent 0a57641f3f
commit bf3eb463ae
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 184 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
homeassistant/components/tplink/coordinator.py
View file

@ -7,6 +7,7 @@ import logging
from kasa import AuthenticationException, SmartDevice, SmartDeviceException
from homeassistant import config_entries
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.debounce import Debouncer
@ -20,6 +21,8 @@ REQUEST_REFRESH_DELAY = 0.35
class TPLinkDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""DataUpdateCoordinator to gather data for a specific TPLink device."""
config_entry: config_entries.ConfigEntry
def __init__(
self,
hass: HomeAssistant,

41
homeassistant/components/tplink/entity.py
View file

@ -5,8 +5,14 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from typing import Any, Concatenate, ParamSpec, TypeVar
from kasa import SmartDevice
from kasa import (
AuthenticationException,
SmartDevice,
SmartDeviceException,
TimeoutException,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -21,10 +27,39 @@ _P = ParamSpec("_P")
def async_refresh_after(
func: Callable[Concatenate[_T, _P], Awaitable[None]],
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]:
"""Define a wrapper to refresh after."""
"""Define a wrapper to raise HA errors and refresh after."""
async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None:
await func(self, *args, **kwargs)
try:
await func(self, *args, **kwargs)
except AuthenticationException as ex:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="device_authentication",
translation_placeholders={
"func": func.__name__,
"exc": str(ex),
},
) from ex
except TimeoutException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="device_timeout",
translation_placeholders={
"func": func.__name__,
"exc": str(ex),
},
) from ex
except SmartDeviceException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="device_error",
translation_placeholders={
"func": func.__name__,
"exc": str(ex),
},
) from ex
await self.coordinator.async_request_refresh()
return _async_wrap

11
homeassistant/components/tplink/strings.json
View file

@ -184,5 +184,16 @@
}
}
}
},
"exceptions": {
"device_timeout": {
"message": "Timeout communicating with the device {func}: {exc}"
},
"device_error": {
"message": "Unable to communicate with the device {func}: {exc}"
},
"device_authentication": {
"message": "Device authentication error {func}: {exc}"
}
}
}

66
tests/components/tplink/test_light.py
View file

@ -5,6 +5,7 @@ from __future__ import annotations
from datetime import timedelta
from unittest.mock import MagicMock, PropertyMock
from kasa import AuthenticationException, SmartDeviceException, TimeoutException
import pytest
from homeassistant.components import tplink
@ -24,8 +25,10 @@ from homeassistant.components.light import (
DOMAIN as LIGHT_DOMAIN,
)
from homeassistant.components.tplink.const import DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.const import ATTR_ENTITY_ID, CONF_HOST, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -730,3 +733,66 @@ async def test_smart_strip_custom_sequence_effect(hass: HomeAssistant) -> None:
}
)
strip.set_custom_effect.reset_mock()
@pytest.mark.parametrize(
("exception_type", "msg", "reauth_expected"),
[
(
AuthenticationException,
"Device authentication error async_turn_on: test error",
True,
),
(
TimeoutException,
"Timeout communicating with the device async_turn_on: test error",
False,
),
(
SmartDeviceException,
"Unable to communicate with the device async_turn_on: test error",
False,
),
],
ids=["Authentication", "Timeout", "Other"],
)
async def test_light_errors_when_turned_on(
hass: HomeAssistant,
exception_type,
msg,
reauth_expected,
) -> None:
"""Tests the light wraps errors correctly."""
already_migrated_config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: "127.0.0.1"}, unique_id=MAC_ADDRESS
)
already_migrated_config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
bulb.turn_on.side_effect = exception_type(msg)
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "light.my_bulb"
assert not any(
already_migrated_config_entry.async_get_active_flows(hass, {SOURCE_REAUTH})
)
with pytest.raises(HomeAssistantError, match=msg):
await hass.services.async_call(
LIGHT_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.async_block_till_done()
assert bulb.turn_on.call_count == 1
assert (
any(
flow
for flow in already_migrated_config_entry.async_get_active_flows(
hass, {SOURCE_REAUTH}
)
if flow["handler"] == tplink.DOMAIN
)
== reauth_expected
)

67
tests/components/tplink/test_switch.py
View file

@ -3,12 +3,13 @@
from datetime import timedelta
from unittest.mock import AsyncMock
from kasa import SmartDeviceException
from kasa import AuthenticationException, SmartDeviceException, TimeoutException
import pytest
from homeassistant.components import tplink
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.tplink.const import DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
@ -17,6 +18,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util, slugify
@ -202,3 +204,66 @@ async def test_strip_unique_ids(hass: HomeAssistant) -> None:
assert (
entity_registry.async_get(entity_id).unique_id == f"PLUG{plug_id}DEVICEID"
)
@pytest.mark.parametrize(
("exception_type", "msg", "reauth_expected"),
[
(
AuthenticationException,
"Device authentication error async_turn_on: test error",
True,
),
(
TimeoutException,
"Timeout communicating with the device async_turn_on: test error",
False,
),
(
SmartDeviceException,
"Unable to communicate with the device async_turn_on: test error",
False,
),
],
ids=["Authentication", "Timeout", "Other"],
)
async def test_plug_errors_when_turned_on(
hass: HomeAssistant,
exception_type,
msg,
reauth_expected,
) -> None:
"""Tests the plug wraps errors correctly."""
already_migrated_config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: "127.0.0.1"}, unique_id=MAC_ADDRESS
)
already_migrated_config_entry.add_to_hass(hass)
plug = _mocked_plug()
plug.turn_on.side_effect = exception_type("test error")
with _patch_discovery(device=plug), _patch_connect(device=plug):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "switch.my_plug"
assert not any(
already_migrated_config_entry.async_get_active_flows(hass, {SOURCE_REAUTH})
)
with pytest.raises(HomeAssistantError, match=msg):
await hass.services.async_call(
SWITCH_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.async_block_till_done()
assert plug.turn_on.call_count == 1
assert (
any(
flow
for flow in already_migrated_config_entry.async_get_active_flows(
hass, {SOURCE_REAUTH}
)
if flow["handler"] == tplink.DOMAIN
)
== reauth_expected
)