Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup Shelly update platform (#80845)

pull/80924/head
Shay Levy 2022-10-25 00:20:26 +03:00 committed by GitHub
parent 2ddf1f9416
commit dce4753510
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 437 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
homeassistant/components/shelly/coordinator.py
View File

@ -14,13 +14,11 @@ from aioshelly.rpc_device import RpcDevice
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_DEVICE_ID, CONF_HOST, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
ATTR_BETA,
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
@ -249,43 +247,6 @@ class ShellyBlockCoordinator(DataUpdateCoordinator):
self.device_id = entry.id
self.device.subscribe_updates(self.async_set_updated_data)
async def async_trigger_ota_update(self, beta: bool = False) -> None:
"""Trigger or schedule an ota update."""
update_data = self.device.status["update"]
LOGGER.debug("OTA update service - update_data: %s", update_data)
if not update_data["has_update"] and not beta:
LOGGER.warning("No OTA update available for device %s", self.name)
return
if beta and not update_data.get("beta_version"):
LOGGER.warning(
"No OTA update on beta channel available for device %s", self.name
)
return
if update_data["status"] == "updating":
LOGGER.warning("OTA update already in progress for %s", self.name)
return
new_version = update_data["new_version"]
if beta:
new_version = update_data["beta_version"]
LOGGER.info(
"Start OTA update of device %s from '%s' to '%s'",
self.name,
self.device.firmware_version,
new_version,
)
try:
result = await self.device.trigger_ota_update(beta=beta)
except DeviceConnectionError as err:
raise HomeAssistantError(f"Error starting OTA update: {repr(err)}") from err
except InvalidAuthError:
self.entry.async_start_reauth(self.hass)
else:
LOGGER.debug("Result of OTA update call: %s", result)
def shutdown(self) -> None:
"""Shutdown the coordinator."""
self.device.shutdown()
@ -480,46 +441,6 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
self.device_id = entry.id
self.device.subscribe_updates(self.async_set_updated_data)
async def async_trigger_ota_update(self, beta: bool = False) -> None:
"""Trigger an ota update."""
update_data = self.device.status["sys"]["available_updates"]
LOGGER.debug("OTA update service - update_data: %s", update_data)
if not bool(update_data) or (not update_data.get("stable") and not beta):
LOGGER.warning("No OTA update available for device %s", self.name)
return
if beta and not update_data.get(ATTR_BETA):
LOGGER.warning(
"No OTA update on beta channel available for device %s", self.name
)
return
new_version = update_data.get("stable", {"version": ""})["version"]
if beta:
new_version = update_data.get(ATTR_BETA, {"version": ""})["version"]
assert self.device.shelly
LOGGER.info(
"Start OTA update of device %s from '%s' to '%s'",
self.name,
self.device.firmware_version,
new_version,
)
try:
await self.device.trigger_ota_update(beta=beta)
except DeviceConnectionError as err:
raise HomeAssistantError(
f"OTA update connection error: {repr(err)}"
) from err
except RpcCallError as err:
raise HomeAssistantError(f"OTA update request error: {repr(err)}") from err
except InvalidAuthError:
self.entry.async_start_reauth(self.hass)
else:
LOGGER.debug("OTA update call successful")
async def shutdown(self) -> None:
"""Shutdown the coordinator."""
await self.device.shutdown()

89
homeassistant/components/shelly/update.py
View File

@ -6,6 +6,8 @@ from dataclasses import dataclass
import logging
from typing import Any, Final, cast
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
from homeassistant.components.update import (
UpdateDeviceClass,
UpdateEntity,
@ -14,11 +16,12 @@ from homeassistant.components.update import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import CONF_SLEEP_PERIOD
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator, get_entry_data
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator
from .entity import (
RestEntityDescription,
RpcEntityDescription,
@ -37,7 +40,7 @@ class RpcUpdateRequiredKeysMixin:
"""Class for RPC update required keys."""
latest_version: Callable[[dict], Any]
install: Callable
beta: bool
@dataclass
@ -45,7 +48,7 @@ class RestUpdateRequiredKeysMixin:
"""Class for REST update required keys."""
latest_version: Callable[[dict], Any]
install: Callable
beta: bool
@dataclass
@ -67,7 +70,7 @@ REST_UPDATES: Final = {
name="Firmware Update",
key="fwupdate",
latest_version=lambda status: status["update"]["new_version"],
install=lambda coordinator: coordinator.async_trigger_ota_update(),
beta=False,
device_class=UpdateDeviceClass.FIRMWARE,
entity_category=EntityCategory.CONFIG,
entity_registry_enabled_default=False,
@ -76,7 +79,7 @@ REST_UPDATES: Final = {
name="Beta Firmware Update",
key="fwupdate",
latest_version=lambda status: status["update"].get("beta_version"),
install=lambda coordinator: coordinator.async_trigger_ota_update(beta=True),
beta=True,
device_class=UpdateDeviceClass.FIRMWARE,
entity_category=EntityCategory.CONFIG,
entity_registry_enabled_default=False,
@ -88,10 +91,8 @@ RPC_UPDATES: Final = {
name="Firmware Update",
key="sys",
sub_key="available_updates",
latest_version=lambda status: status.get("stable", {"version": None})[
"version"
],
install=lambda coordinator: coordinator.async_trigger_ota_update(),
latest_version=lambda status: status.get("stable", {"version": ""})["version"],
beta=False,
device_class=UpdateDeviceClass.FIRMWARE,
entity_category=EntityCategory.CONFIG,
entity_registry_enabled_default=False,
@ -100,8 +101,8 @@ RPC_UPDATES: Final = {
name="Beta Firmware Update",
key="sys",
sub_key="available_updates",
latest_version=lambda status: status.get("beta", {"version": None})["version"],
install=lambda coordinator: coordinator.async_trigger_ota_update(beta=True),
latest_version=lambda status: status.get("beta", {"version": ""})["version"],
beta=True,
device_class=UpdateDeviceClass.FIRMWARE,
entity_category=EntityCategory.CONFIG,
entity_registry_enabled_default=False,
@ -151,11 +152,7 @@ class RestUpdateEntity(ShellyRestAttributeEntity, UpdateEntity):
@property
def installed_version(self) -> str | None:
"""Version currently in use."""
version = self.block_coordinator.device.status["update"]["old_version"]
if version is None:
return None
return cast(str, version)
return cast(str, self.block_coordinator.device.status["update"]["old_version"])
@property
def latest_version(self) -> str | None:
@ -163,7 +160,7 @@ class RestUpdateEntity(ShellyRestAttributeEntity, UpdateEntity):
new_version = self.entity_description.latest_version(
self.block_coordinator.device.status,
)
if new_version not in (None, ""):
if new_version:
return cast(str, new_version)
return self.installed_version
@ -177,10 +174,29 @@ class RestUpdateEntity(ShellyRestAttributeEntity, UpdateEntity):
self, version: str | None, backup: bool, **kwargs: Any
) -> None:
"""Install the latest firmware version."""
config_entry = self.block_coordinator.entry
coordinator = get_entry_data(self.hass)[config_entry.entry_id].block
self._in_progress_old_version = self.installed_version
await self.entity_description.install(coordinator)
beta = self.entity_description.beta
update_data = self.coordinator.device.status["update"]
LOGGER.debug("OTA update service - update_data: %s", update_data)
new_version = update_data["new_version"]
if beta:
new_version = update_data["beta_version"]
LOGGER.info(
"Starting OTA update of device %s from '%s' to '%s'",
self.name,
self.coordinator.device.firmware_version,
new_version,
)
try:
result = await self.coordinator.device.trigger_ota_update(beta=beta)
except DeviceConnectionError as err:
raise HomeAssistantError(f"Error starting OTA update: {repr(err)}") from err
except InvalidAuthError:
self.coordinator.entry.async_start_reauth(self.hass)
else:
LOGGER.debug("Result of OTA update call: %s", result)
class RpcUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity):
@ -205,9 +221,7 @@ class RpcUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity):
@property
def installed_version(self) -> str | None:
"""Version currently in use."""
if self.coordinator.device.shelly is None:
return None
assert self.coordinator.device.shelly
return cast(str, self.coordinator.device.shelly["ver"])
@property
@ -216,7 +230,7 @@ class RpcUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity):
new_version = self.entity_description.latest_version(
self.coordinator.device.status[self.key][self.entity_description.sub_key],
)
if new_version is not None:
if new_version:
return cast(str, new_version)
return self.installed_version
@ -231,4 +245,29 @@ class RpcUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity):
) -> None:
"""Install the latest firmware version."""
self._in_progress_old_version = self.installed_version
await self.entity_description.install(self.coordinator)
beta = self.entity_description.beta
update_data = self.coordinator.device.status["sys"]["available_updates"]
LOGGER.debug("OTA update service - update_data: %s", update_data)
new_version = update_data.get("stable", {"version": ""})["version"]
if beta:
new_version = update_data.get("beta", {"version": ""})["version"]
LOGGER.info(
"Starting OTA update of device %s from '%s' to '%s'",
self.coordinator.name,
self.coordinator.device.firmware_version,
new_version,
)
try:
await self.coordinator.device.trigger_ota_update(beta=beta)
except DeviceConnectionError as err:
raise HomeAssistantError(
f"OTA update connection error: {repr(err)}"
) from err
except RpcCallError as err:
raise HomeAssistantError(f"OTA update request error: {repr(err)}") from err
except InvalidAuthError:
self.coordinator.entry.async_start_reauth(self.hass)
else:
LOGGER.debug("OTA update call successful")

391
tests/components/shelly/test_update.py
View File

@ -1,13 +1,29 @@
"""Tests for Shelly update platform."""
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.update import DOMAIN as UPDATE_DOMAIN, SERVICE_INSTALL
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON, STATE_UNKNOWN
from datetime import timedelta
from unittest.mock import AsyncMock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
import pytest
from homeassistant.components.shelly.const import DOMAIN, REST_SENSORS_UPDATE_INTERVAL
from homeassistant.components.update import (
ATTR_IN_PROGRESS,
ATTR_INSTALLED_VERSION,
ATTR_LATEST_VERSION,
DOMAIN as UPDATE_DOMAIN,
SERVICE_INSTALL,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_registry import async_get
from homeassistant.util import dt
from . import MOCK_MAC, init_integration
from tests.common import async_fire_time_changed
async def test_block_update(hass: HomeAssistant, mock_block_device, monkeypatch):
"""Test block device update entity."""
@ -19,9 +35,15 @@ async def test_block_update(hass: HomeAssistant, mock_block_device, monkeypatch)
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "1")
monkeypatch.setitem(mock_block_device.status["update"], "new_version", "2")
await init_integration(hass, 1)
assert hass.states.get("update.test_name_firmware_update").state == STATE_ON
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2"
assert state.attributes[ATTR_IN_PROGRESS] is False
await hass.services.async_call(
UPDATE_DOMAIN,
@ -31,17 +53,162 @@ async def test_block_update(hass: HomeAssistant, mock_block_device, monkeypatch)
)
assert mock_block_device.trigger_ota_update.call_count == 1
monkeypatch.setitem(mock_block_device.status["update"], "old_version", None)
monkeypatch.setitem(mock_block_device.status["update"], "new_version", None)
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2"
assert state.attributes[ATTR_IN_PROGRESS] is True
# update entity
await async_update_entity(hass, "update.test_name_firmware_update")
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "2")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
assert hass.states.get("update.test_name_firmware_update").state == STATE_UNKNOWN
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "2"
assert state.attributes[ATTR_LATEST_VERSION] == "2"
assert state.attributes[ATTR_IN_PROGRESS] is False
async def test_block_beta_update(hass: HomeAssistant, mock_block_device, monkeypatch):
"""Test block device beta update entity."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
f"{MOCK_MAC}-fwupdate_beta",
suggested_object_id="test_name_beta_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "1")
monkeypatch.setitem(mock_block_device.status["update"], "new_version", "2")
monkeypatch.setitem(mock_block_device.status["update"], "beta_version", "")
await init_integration(hass, 1)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "1"
assert state.attributes[ATTR_IN_PROGRESS] is False
monkeypatch.setitem(mock_block_device.status["update"], "beta_version", "2b")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2b"
assert state.attributes[ATTR_IN_PROGRESS] is False
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: "update.test_name_beta_firmware_update"},
blocking=True,
)
assert mock_block_device.trigger_ota_update.call_count == 1
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2b"
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "2b")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "2b"
assert state.attributes[ATTR_LATEST_VERSION] == "2b"
assert state.attributes[ATTR_IN_PROGRESS] is False
async def test_block_update_connection_error(
hass: HomeAssistant, mock_block_device, monkeypatch, caplog
):
"""Test block device update connection error."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
f"{MOCK_MAC}-fwupdate",
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "1")
monkeypatch.setitem(mock_block_device.status["update"], "new_version", "2")
monkeypatch.setattr(
mock_block_device,
"trigger_ota_update",
AsyncMock(side_effect=DeviceConnectionError),
)
await init_integration(hass, 1)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: "update.test_name_firmware_update"},
blocking=True,
)
assert "Error starting OTA update" in caplog.text
async def test_block_update_auth_error(
hass: HomeAssistant, mock_block_device, monkeypatch
):
"""Test block device update authentication error."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
f"{MOCK_MAC}-fwupdate",
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "1")
monkeypatch.setitem(mock_block_device.status["update"], "new_version", "2")
monkeypatch.setattr(
mock_block_device,
"trigger_ota_update",
AsyncMock(side_effect=InvalidAuthError),
)
entry = await init_integration(hass, 1)
assert entry.state == ConfigEntryState.LOADED
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: "update.test_name_firmware_update"},
blocking=True,
)
assert entry.state == 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
async def test_rpc_update(hass: HomeAssistant, mock_rpc_device, monkeypatch):
"""Test rpc device update entity."""
"""Test RPC device update entity."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
@ -50,9 +217,21 @@ async def test_rpc_update(hass: HomeAssistant, mock_rpc_device, monkeypatch):
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "1")
monkeypatch.setitem(
mock_rpc_device.status["sys"],
"available_updates",
{
"stable": {"version": "2"},
},
)
await init_integration(hass, 2)
assert hass.states.get("update.test_name_firmware_update").state == STATE_ON
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2"
assert state.attributes[ATTR_IN_PROGRESS] is False
await hass.services.async_call(
UPDATE_DOMAIN,
@ -60,13 +239,189 @@ async def test_rpc_update(hass: HomeAssistant, mock_rpc_device, monkeypatch):
{ATTR_ENTITY_ID: "update.test_name_firmware_update"},
blocking=True,
)
await hass.async_block_till_done()
assert mock_rpc_device.trigger_ota_update.call_count == 1
monkeypatch.setitem(mock_rpc_device.status["sys"], "available_updates", {})
monkeypatch.setattr(mock_rpc_device, "shelly", None)
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2"
assert state.attributes[ATTR_IN_PROGRESS] is True
# update entity
await async_update_entity(hass, "update.test_name_firmware_update")
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "2")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
assert hass.states.get("update.test_name_firmware_update").state == STATE_UNKNOWN
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "2"
assert state.attributes[ATTR_LATEST_VERSION] == "2"
assert state.attributes[ATTR_IN_PROGRESS] is False
async def test_rpc_beta_update(hass: HomeAssistant, mock_rpc_device, monkeypatch):
"""Test RPC device beta update entity."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
f"{MOCK_MAC}-sys-fwupdate_beta",
suggested_object_id="test_name_beta_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "1")
monkeypatch.setitem(
mock_rpc_device.status["sys"],
"available_updates",
{
"stable": {"version": "2"},
"beta": {"version": ""},
},
)
await init_integration(hass, 2)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "1"
assert state.attributes[ATTR_IN_PROGRESS] is False
monkeypatch.setitem(
mock_rpc_device.status["sys"],
"available_updates",
{
"stable": {"version": "2"},
"beta": {"version": "2b"},
},
)
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2b"
assert state.attributes[ATTR_IN_PROGRESS] is False
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: "update.test_name_beta_firmware_update"},
blocking=True,
)
assert mock_rpc_device.trigger_ota_update.call_count == 1
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "1"
assert state.attributes[ATTR_LATEST_VERSION] == "2b"
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "2b")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "2b"
assert state.attributes[ATTR_LATEST_VERSION] == "2b"
assert state.attributes[ATTR_IN_PROGRESS] is False
@pytest.mark.parametrize(
"exc, error",
[
(DeviceConnectionError, "Error starting OTA update"),
(RpcCallError(-1, "error"), "OTA update request error"),
],
)
async def test_rpc_update__errors(
hass: HomeAssistant, exc, error, mock_rpc_device, monkeypatch, caplog
):
"""Test RPC device update connection/call errors."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
f"{MOCK_MAC}-sys-fwupdate",
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "1")
monkeypatch.setitem(
mock_rpc_device.status["sys"],
"available_updates",
{
"stable": {"version": "2"},
"beta": {"version": ""},
},
)
monkeypatch.setattr(
mock_rpc_device, "trigger_ota_update", AsyncMock(side_effect=exc)
)
await init_integration(hass, 2)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: "update.test_name_firmware_update"},
blocking=True,
)
assert error in caplog.text
async def test_rpc_update_auth_error(
hass: HomeAssistant, mock_rpc_device, monkeypatch, caplog
):
"""Test RPC device update authentication error."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
f"{MOCK_MAC}-sys-fwupdate",
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "1")
monkeypatch.setitem(
mock_rpc_device.status["sys"],
"available_updates",
{
"stable": {"version": "2"},
"beta": {"version": ""},
},
)
monkeypatch.setattr(
mock_rpc_device,
"trigger_ota_update",
AsyncMock(side_effect=InvalidAuthError),
)
entry = await init_integration(hass, 2)
assert entry.state == ConfigEntryState.LOADED
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: "update.test_name_firmware_update"},
blocking=True,
)
assert entry.state == 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