Use aioshelly methods with Shelly RPC text and select entities (#143464)

This commit is contained in:
Maciej Bieniek 2025-04-23 10:07:05 +02:00 committed by GitHub
parent d86d7b8843
commit b785d5297a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 217 additions and 4 deletions

View File

@ -20,6 +20,7 @@ from .entity import (
RpcEntityDescription,
ShellyRpcAttributeEntity,
async_setup_entry_rpc,
rpc_call,
)
from .utils import (
async_remove_orphaned_entities,
@ -75,6 +76,7 @@ class RpcSelect(ShellyRpcAttributeEntity, SelectEntity):
"""Represent a RPC select entity."""
entity_description: RpcSelectDescription
_id: int
def __init__(
self,
@ -96,8 +98,9 @@ class RpcSelect(ShellyRpcAttributeEntity, SelectEntity):
return self.option_map[self.attribute_value]
@rpc_call
async def async_select_option(self, option: str) -> None:
"""Change the value."""
await self.call_rpc(
"Enum.Set", {"id": self._id, "value": self.reversed_option_map[option]}
await self.coordinator.device.enum_set(
self._id, self.reversed_option_map[option]
)

View File

@ -20,6 +20,7 @@ from .entity import (
RpcEntityDescription,
ShellyRpcAttributeEntity,
async_setup_entry_rpc,
rpc_call,
)
from .utils import (
async_remove_orphaned_entities,
@ -75,6 +76,7 @@ class RpcText(ShellyRpcAttributeEntity, TextEntity):
"""Represent a RPC text entity."""
entity_description: RpcTextDescription
_id: int
@property
def native_value(self) -> str | None:
@ -84,6 +86,7 @@ class RpcText(ShellyRpcAttributeEntity, TextEntity):
return self.attribute_value
@rpc_call
async def async_set_value(self, value: str) -> None:
"""Change the value."""
await self.call_rpc("Text.Set", {"id": self._id, "value": value})
await self.coordinator.device.text_set(self._id, value)

View File

@ -3,6 +3,7 @@
from copy import deepcopy
from unittest.mock import Mock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
import pytest
from homeassistant.components.select import (
@ -11,8 +12,11 @@ from homeassistant.components.select import (
DOMAIN as SELECT_PLATFORM,
SERVICE_SELECT_OPTION,
)
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
@ -81,7 +85,7 @@ async def test_rpc_device_virtual_enum(
blocking=True,
)
# 'Title 1' corresponds to 'option 1'
assert mock_rpc_device.call_rpc.call_args[0][1] == {"id": 203, "value": "option 1"}
mock_rpc_device.enum_set.assert_called_once_with(203, "option 1")
mock_rpc_device.mock_update()
assert (state := hass.states.get(entity_id))
@ -149,3 +153,108 @@ async def test_rpc_remove_virtual_enum_when_orphaned(
await hass.async_block_till_done()
assert entity_registry.async_get(entity_id) is None
@pytest.mark.parametrize(
("exception", "error"),
[
(
DeviceConnectionError,
"Device communication error occurred while calling action for select.test_name_enum_203 of Test name",
),
(
RpcCallError(999),
"RPC call error occurred while calling action for select.test_name_enum_203 of Test name",
),
],
)
async def test_select_set_exc(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
exception: Exception,
error: str,
) -> None:
"""Test select setting with exception."""
config = deepcopy(mock_rpc_device.config)
config["enum:203"] = {
"name": None,
"options": ["option 1", "option 2", "option 3"],
"meta": {
"ui": {
"view": "dropdown",
"titles": {"option 1": "Title 1", "option 2": None},
}
},
}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["enum:203"] = {"value": "option 1"}
monkeypatch.setattr(mock_rpc_device, "status", status)
await init_integration(hass, 3)
mock_rpc_device.enum_set.side_effect = exception
with pytest.raises(HomeAssistantError, match=error):
await hass.services.async_call(
SELECT_PLATFORM,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: f"{SELECT_PLATFORM}.test_name_enum_203",
ATTR_OPTION: "option 2",
},
blocking=True,
)
async def test_select_set_reauth_error(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test select setting with authentication error."""
config = deepcopy(mock_rpc_device.config)
config["enum:203"] = {
"name": None,
"options": ["option 1", "option 2", "option 3"],
"meta": {
"ui": {
"view": "dropdown",
"titles": {"option 1": "Title 1", "option 2": None},
}
},
}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["enum:203"] = {"value": "option 1"}
monkeypatch.setattr(mock_rpc_device, "status", status)
entry = await init_integration(hass, 3)
mock_rpc_device.enum_set.side_effect = InvalidAuthError
await hass.services.async_call(
SELECT_PLATFORM,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: f"{SELECT_PLATFORM}.test_name_enum_203",
ATTR_OPTION: "option 2",
},
blocking=True,
)
assert entry.state is ConfigEntryState.LOADED
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == entry.entry_id

View File

@ -3,15 +3,19 @@
from copy import deepcopy
from unittest.mock import Mock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
import pytest
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.text import (
ATTR_VALUE,
DOMAIN as TEXT_PLATFORM,
SERVICE_SET_VALUE,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
@ -67,6 +71,7 @@ async def test_rpc_device_virtual_text(
blocking=True,
)
mock_rpc_device.mock_update()
mock_rpc_device.text_set.assert_called_once_with(203, "sed do eiusmod")
assert (state := hass.states.get(entity_id))
assert state.state == "sed do eiusmod"
@ -127,3 +132,96 @@ async def test_rpc_remove_virtual_text_when_orphaned(
await hass.async_block_till_done()
assert entity_registry.async_get(entity_id) is None
@pytest.mark.parametrize(
("exception", "error"),
[
(
DeviceConnectionError,
"Device communication error occurred while calling action for text.test_name_text_203 of Test name",
),
(
RpcCallError(999),
"RPC call error occurred while calling action for text.test_name_text_203 of Test name",
),
],
)
async def test_text_set_exc(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
exception: Exception,
error: str,
) -> None:
"""Test text setting with exception."""
config = deepcopy(mock_rpc_device.config)
config["text:203"] = {
"name": None,
"meta": {"ui": {"view": "field"}},
}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["text:203"] = {"value": "lorem ipsum"}
monkeypatch.setattr(mock_rpc_device, "status", status)
await init_integration(hass, 3)
mock_rpc_device.text_set.side_effect = exception
with pytest.raises(HomeAssistantError, match=error):
await hass.services.async_call(
TEXT_PLATFORM,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: f"{TEXT_PLATFORM}.test_name_text_203",
ATTR_VALUE: "new value",
},
blocking=True,
)
async def test_text_set_reauth_error(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test text setting with authentication error."""
config = deepcopy(mock_rpc_device.config)
config["text:203"] = {
"name": None,
"meta": {"ui": {"view": "field"}},
}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["text:203"] = {"value": "lorem ipsum"}
monkeypatch.setattr(mock_rpc_device, "status", status)
entry = await init_integration(hass, 3)
mock_rpc_device.text_set.side_effect = InvalidAuthError
await hass.services.async_call(
TEXT_PLATFORM,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: f"{TEXT_PLATFORM}.test_name_text_203",
ATTR_VALUE: "new value",
},
blocking=True,
)
assert entry.state is ConfigEntryState.LOADED
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == entry.entry_id