Add LoRa add-on firmware update for Shelly (#169607)

homeassistant/components/shelly/strings.json

@@ -616,6 +616,9 @@
     "update": {
       "beta_firmware": {
         "name": "Beta firmware"
+      },
+      "lora_firmware": {
+        "name": "LoRa add-on firmware"
       }
     },
     "valve": {

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,
+    ),
 }
 
 

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,