From dd14d7982ec45453fdf616cfef3d2db2653fad77 Mon Sep 17 00:00:00 2001 From: Simone Chemelli Date: Mon, 27 Jul 2026 13:15:23 +0200 Subject: [PATCH] Add LoRa add-on firmware update for Shelly (#169607) --- homeassistant/components/shelly/strings.json | 3 + homeassistant/components/shelly/update.py | 94 ++++++++ tests/components/shelly/test_update.py | 234 +++++++++++++++++++ 3 files changed, 331 insertions(+) diff --git a/homeassistant/components/shelly/strings.json b/homeassistant/components/shelly/strings.json index 56a2712c0031..cf6d25176cef 100644 --- a/homeassistant/components/shelly/strings.json +++ b/homeassistant/components/shelly/strings.json @@ -616,6 +616,9 @@ "update": { "beta_firmware": { "name": "Beta firmware" + }, + "lora_firmware": { + "name": "LoRa add-on firmware" } }, "valve": { diff --git a/homeassistant/components/shelly/update.py b/homeassistant/components/shelly/update.py index ca570a80b526..ea0fb12b1f7d 100644 --- a/homeassistant/components/shelly/update.py +++ b/homeassistant/components/shelly/update.py @@ -81,6 +81,90 @@ REST_UPDATES: Final = { ), } + +class RpcLoraAddOnUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity): + """Represent a RPC LoRa add-on update entity.""" + + _attr_supported_features = ( + UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS + ) + entity_description: RpcUpdateDescription + + @property + def _update_status(self) -> dict[str, Any]: + """Status of the LoRa add-on update, reported by the lora component.""" + return self.status.get("update") or {} + + @property + @override + def installed_version(self) -> str | None: + """Version currently in use.""" + return cast(str | None, self.status.get("fw_version")) + + @property + @override + def latest_version(self) -> str | None: + """Latest version available for install.""" + status = self.entity_description.latest_version(self.status) + # After firmware update, available_updates can be None, for a brief moment. + new_version = (status.get("available_updates") or {}).get( + "stable", {"version": ""} + )["version"] + if new_version: + return cast(str, new_version) + + return self.installed_version + + @property + @override + def in_progress(self) -> bool: + """Update installation in progress.""" + return bool(self._update_status.get("state") in ("started", "updating")) + + @property + @override + def update_percentage(self) -> int | None: + """Update installation progress.""" + return cast(int | None, self._update_status.get("progress")) + + @override + async def async_install( + self, version: str | None, backup: bool, **kwargs: Any + ) -> None: + """Install the latest firmware version.""" + update_data = self.status["available_updates"] + LOGGER.debug("LoRa add-on OTA update service - update_data: %s", update_data) + + new_version = update_data.get("stable", {"version": ""})["version"] + + LOGGER.info( + "Starting OTA update of LoRa add-on on device %s from '%s' to '%s'", + self.coordinator.name, + self.installed_version, + new_version, + ) + try: + await self.coordinator.device.trigger_add_on_ota_update() + except DeviceConnectionError as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="ota_update_connection_error", + translation_placeholders={"device": self.coordinator.name}, + ) from err + except RpcCallError as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="ota_update_rpc_error", + translation_placeholders={"device": self.coordinator.name}, + ) from err + except InvalidAuthError: + await self.coordinator.async_shutdown_device_and_start_reauth() + else: + LOGGER.debug( + "LoRa add-on OTA update call for %s successful", self.coordinator.name + ) + + RPC_UPDATES: Final = { "fwupdate": RpcUpdateDescription( key="sys", @@ -100,6 +184,16 @@ RPC_UPDATES: Final = { entity_category=EntityCategory.CONFIG, entity_registry_enabled_default=False, ), + "loraupdate": RpcUpdateDescription( + key="lora", + translation_key="lora_firmware", + # Right after firmware update, lora status can be None + latest_version=lambda status: status or {}, + beta=False, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + entity_class=RpcLoraAddOnUpdateEntity, + ), } diff --git a/tests/components/shelly/test_update.py b/tests/components/shelly/test_update.py index b75412b8dcda..16dae97fd384 100644 --- a/tests/components/shelly/test_update.py +++ b/tests/components/shelly/test_update.py @@ -403,6 +403,240 @@ async def test_rpc_update( assert entry.unique_id == "123456789ABC-sys-fwupdate" +async def test_rpc_lora_update( + hass: HomeAssistant, + mock_rpc_device: Mock, + entity_registry: EntityRegistry, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test RPC device LoRa update entity, idle state.""" + entity_id = "update.test_name_lora_add_on_firmware" + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "1", + "available_updates": { + "stable": {"version": "2"}, + }, + }, + ) + await init_integration(hass, 2) + + assert (state := hass.states.get(entity_id)) + 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 + assert state.attributes[ATTR_UPDATE_PERCENTAGE] is None + assert state.attributes[ATTR_RELEASE_URL] is None + supported_feat = state.attributes[ATTR_SUPPORTED_FEATURES] + assert supported_feat == UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS + + assert (entry := entity_registry.async_get(entity_id)) + assert entry.unique_id == "123456789ABC-lora-loraupdate" + + +async def test_rpc_lora_update_no_update_available( + hass: HomeAssistant, + mock_rpc_device: Mock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test RPC device LoRa update entity when no update is available.""" + entity_id = "update.test_name_lora_add_on_firmware" + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "1", + "available_updates": {}, + }, + ) + await init_integration(hass, 2) + + assert (state := hass.states.get(entity_id)) + assert state.state == STATE_OFF + assert state.attributes[ATTR_INSTALLED_VERSION] == "1" + assert state.attributes[ATTR_LATEST_VERSION] == "1" + + +async def test_rpc_lora_update_in_progress( + hass: HomeAssistant, + mock_rpc_device: Mock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test RPC device LoRa update entity, install and updating state.""" + entity_id = "update.test_name_lora_add_on_firmware" + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "1", + "available_updates": { + "stable": {"version": "2"}, + }, + }, + ) + await init_integration(hass, 2) + + await hass.services.async_call( + UPDATE_DOMAIN, + SERVICE_INSTALL, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + + assert mock_rpc_device.trigger_add_on_ota_update.call_count == 1 + + monkeypatch.setitem( + mock_rpc_device.status["lora"], + "update", + {"state": "updating", "progress": 50}, + ) + mock_rpc_device.mock_update() + + assert (state := hass.states.get(entity_id)) + assert state.state == STATE_ON + assert state.attributes[ATTR_IN_PROGRESS] is True + assert state.attributes[ATTR_UPDATE_PERCENTAGE] == 50 + + +async def test_rpc_lora_update_success( + hass: HomeAssistant, + mock_rpc_device: Mock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test RPC device LoRa update entity, update completed.""" + entity_id = "update.test_name_lora_add_on_firmware" + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "1", + "update": {"state": "updating", "progress": 90}, + "available_updates": { + "stable": {"version": "2"}, + }, + }, + ) + await init_integration(hass, 2) + + assert (state := hass.states.get(entity_id)) + assert state.attributes[ATTR_IN_PROGRESS] is True + assert state.attributes[ATTR_UPDATE_PERCENTAGE] == 90 + + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "2", + "update": {"state": "idle"}, + "available_updates": { + "stable": {"version": "2"}, + }, + }, + ) + mock_rpc_device.mock_update() + + assert (state := hass.states.get(entity_id)) + 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 + assert state.attributes[ATTR_UPDATE_PERCENTAGE] is None + + +@pytest.mark.parametrize( + ("exc", "error"), + [ + ( + DeviceConnectionError, + "Device communication error occurred while triggering" + " OTA update for Test name", + ), + ( + RpcCallError(-1, "error"), + "RPC call error occurred while triggering OTA update for Test name", + ), + ], +) +async def test_rpc_lora_update_errors( + hass: HomeAssistant, + exc: Exception, + error: str, + mock_rpc_device: Mock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test RPC device LoRa update connection/call errors.""" + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "1", + "available_updates": { + "stable": {"version": "2"}, + }, + }, + ) + monkeypatch.setattr( + mock_rpc_device, "trigger_add_on_ota_update", AsyncMock(side_effect=exc) + ) + await init_integration(hass, 2) + + with pytest.raises(HomeAssistantError, match=error): + await hass.services.async_call( + UPDATE_DOMAIN, + SERVICE_INSTALL, + {ATTR_ENTITY_ID: "update.test_name_lora_add_on_firmware"}, + blocking=True, + ) + + +async def test_rpc_lora_update_auth_error( + hass: HomeAssistant, + mock_rpc_device: Mock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test RPC device LoRa update authentication error.""" + monkeypatch.setitem( + mock_rpc_device.status, + "lora", + { + "fw_version": "1", + "available_updates": { + "stable": {"version": "2"}, + }, + }, + ) + monkeypatch.setattr( + mock_rpc_device, + "trigger_add_on_ota_update", + AsyncMock(side_effect=InvalidAuthError), + ) + entry = await init_integration(hass, 2) + + assert entry.state is ConfigEntryState.LOADED + + await hass.services.async_call( + UPDATE_DOMAIN, + SERVICE_INSTALL, + {ATTR_ENTITY_ID: "update.test_name_lora_add_on_firmware"}, + 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 + + async def test_rpc_sleeping_update( hass: HomeAssistant, mock_rpc_device: Mock,