Add repair flow for Shelly BLE scanner with unsupported firmware (#143850)

This commit is contained in:
Maciej Bieniek 2025-04-30 15:43:07 +02:00 committed by GitHub
parent d8122d149b
commit fa1dc75517
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 310 additions and 1 deletions

View File

@ -56,6 +56,7 @@ from .coordinator import (
ShellyRpcCoordinator,
ShellyRpcPollingCoordinator,
)
from .repairs import async_manage_ble_scanner_firmware_unsupported_issue
from .utils import (
async_create_issue_unsupported_firmware,
get_coap_context,
@ -320,6 +321,10 @@ async def _async_setup_rpc_entry(hass: HomeAssistant, entry: ShellyConfigEntry)
await hass.config_entries.async_forward_entry_setups(
entry, runtime_data.platforms
)
async_manage_ble_scanner_firmware_unsupported_issue(
hass,
entry,
)
elif (
sleep_period is None
or device_entry is None

View File

@ -227,6 +227,8 @@ class BLEScannerMode(StrEnum):
PASSIVE = "passive"
BLE_SCANNER_MIN_FIRMWARE = "1.5.1"
MAX_PUSH_UPDATE_FAILURES = 5
PUSH_UPDATE_ISSUE_ID = "push_update_{unique}"
@ -234,6 +236,8 @@ NOT_CALIBRATED_ISSUE_ID = "not_calibrated_{unique}"
FIRMWARE_UNSUPPORTED_ISSUE_ID = "firmware_unsupported_{unique}"
BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID = "ble_scanner_firmware_unsupported_{unique}"
GAS_VALVE_OPEN_STATES = ("opening", "opened")
OTA_BEGIN = "ota_begin"

View File

@ -0,0 +1,127 @@
"""Repairs flow for Shelly."""
from __future__ import annotations
from typing import TYPE_CHECKING
from aioshelly.const import MODEL_OUT_PLUG_S_G3, MODEL_PLUG_S_G3
from aioshelly.exceptions import DeviceConnectionError, RpcCallError
from aioshelly.rpc_device import RpcDevice
from awesomeversion import AwesomeVersion
import voluptuous as vol
from homeassistant import data_entry_flow
from homeassistant.components.repairs import RepairsFlow
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import issue_registry as ir
from .const import (
BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID,
BLE_SCANNER_MIN_FIRMWARE,
CONF_BLE_SCANNER_MODE,
DOMAIN,
BLEScannerMode,
)
from .coordinator import ShellyConfigEntry
@callback
def async_manage_ble_scanner_firmware_unsupported_issue(
hass: HomeAssistant,
entry: ShellyConfigEntry,
) -> None:
"""Manage the BLE scanner firmware unsupported issue."""
issue_id = BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID.format(unique=entry.unique_id)
if TYPE_CHECKING:
assert entry.runtime_data.rpc is not None
device = entry.runtime_data.rpc.device
supports_scripts = entry.runtime_data.rpc_supports_scripts
if supports_scripts and device.model not in (MODEL_PLUG_S_G3, MODEL_OUT_PLUG_S_G3):
firmware = AwesomeVersion(device.shelly["ver"])
if (
firmware < BLE_SCANNER_MIN_FIRMWARE
and entry.options.get(CONF_BLE_SCANNER_MODE) == BLEScannerMode.ACTIVE
):
ir.async_create_issue(
hass,
DOMAIN,
issue_id,
is_fixable=True,
is_persistent=True,
severity=ir.IssueSeverity.WARNING,
translation_key="ble_scanner_firmware_unsupported",
translation_placeholders={
"device_name": device.name,
"ip_address": device.ip_address,
"firmware": firmware,
},
data={"entry_id": entry.entry_id},
)
return
ir.async_delete_issue(hass, DOMAIN, issue_id)
class BleScannerFirmwareUpdateFlow(RepairsFlow):
"""Handler for an issue fixing flow."""
def __init__(self, device: RpcDevice) -> None:
"""Initialize."""
self._device = device
async def async_step_init(
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Handle the first step of a fix flow."""
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Handle the confirm step of a fix flow."""
if user_input is not None:
return await self.async_step_update_firmware()
issue_registry = ir.async_get(self.hass)
description_placeholders = None
if issue := issue_registry.async_get_issue(self.handler, self.issue_id):
description_placeholders = issue.translation_placeholders
return self.async_show_form(
step_id="confirm",
data_schema=vol.Schema({}),
description_placeholders=description_placeholders,
)
async def async_step_update_firmware(
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Handle the confirm step of a fix flow."""
if not self._device.status["sys"]["available_updates"]:
return self.async_abort(reason="update_not_available")
try:
await self._device.trigger_ota_update()
except (DeviceConnectionError, RpcCallError):
return self.async_abort(reason="cannot_connect")
return self.async_create_entry(title="", data={})
async def async_create_fix_flow(
hass: HomeAssistant, issue_id: str, data: dict[str, str] | None
) -> RepairsFlow:
"""Create flow."""
if TYPE_CHECKING:
assert isinstance(data, dict)
entry_id = data["entry_id"]
entry = hass.config_entries.async_get_entry(entry_id)
if TYPE_CHECKING:
assert entry is not None
device = entry.runtime_data.rpc.device
return BleScannerFirmwareUpdateFlow(device)

View File

@ -262,6 +262,21 @@
}
},
"issues": {
"ble_scanner_firmware_unsupported": {
"title": "{device_name} is running unsupported firmware",
"fix_flow": {
"step": {
"confirm": {
"title": "{device_name} is running unsupported firmware",
"description": "Your Shelly device {device_name} with IP address {ip_address} is running firmware {firmware} and acts as BLE scanner with active mode. This firmware version is not supported for BLE scanner active mode.\n\nSelect **Submit** button to start the OTA update to the latest stable firmware version."
}
},
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"update_not_available": "Device does not offer firmware update. Check internet connectivity (gateway, DNS, time) and restart the device."
}
}
},
"device_not_calibrated": {
"title": "Shelly device {device_name} is not calibrated",
"description": "Shelly device {device_name} with IP address {ip_address} requires calibration. To calibrate the device, it must be rebooted after proper installation on the valve. You can reboot the device in its web panel, go to 'Settings' > 'Device Reboot'."

View File

@ -499,6 +499,7 @@ def _mock_rpc_device(version: str | None = None):
),
xmod_info={},
zigbee_enabled=False,
ip_address="10.10.10.10",
)
type(device).name = PropertyMock(return_value="Test name")
return device

View File

@ -16,6 +16,8 @@ from aioshelly.rpc_device.utils import bluetooth_mac_from_primary_mac
import pytest
from homeassistant.components.shelly.const import (
BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID,
BLE_SCANNER_MIN_FIRMWARE,
BLOCK_EXPECTED_SLEEP_PERIOD,
BLOCK_WRONG_SLEEP_PERIOD,
CONF_BLE_SCANNER_MODE,
@ -38,7 +40,7 @@ from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.device_registry import DeviceRegistry, format_mac
from homeassistant.setup import async_setup_component
from . import init_integration, mutate_rpc_device_status
from . import MOCK_MAC, init_integration, mutate_rpc_device_status
async def test_custom_coap_port(
@ -579,3 +581,27 @@ async def test_device_script_getcode_error(
entry = await init_integration(hass, 2)
assert entry.state is ConfigEntryState.SETUP_RETRY
async def test_ble_scanner_unsupported_firmware_fixed(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test device init with unsupported firmware."""
issue_id = BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID.format(unique=MOCK_MAC)
entry = await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 1
monkeypatch.setitem(mock_rpc_device.shelly, "ver", BLE_SCANNER_MIN_FIRMWARE)
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
assert not issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 0

View File

@ -0,0 +1,131 @@
"""Test repairs handling for Shelly."""
from unittest.mock import Mock
from aioshelly.exceptions import DeviceConnectionError, RpcCallError
import pytest
from homeassistant.components.shelly.const import (
BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID,
CONF_BLE_SCANNER_MODE,
DOMAIN,
BLEScannerMode,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from . import MOCK_MAC, init_integration
from tests.components.repairs import (
async_process_repairs_platforms,
process_repair_fix_flow,
start_repair_fix_flow,
)
from tests.typing import ClientSessionGenerator
async def test_ble_scanner_unsupported_firmware_issue(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_rpc_device: Mock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test repair issues handling for BLE scanner with unsupported firmware."""
issue_id = BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID.format(unique=MOCK_MAC)
assert await async_setup_component(hass, "repairs", {})
await hass.async_block_till_done()
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 1
await async_process_repairs_platforms(hass)
client = await hass_client()
result = await start_repair_fix_flow(client, DOMAIN, issue_id)
flow_id = result["flow_id"]
assert result["step_id"] == "confirm"
result = await process_repair_fix_flow(client, flow_id)
assert result["type"] == "create_entry"
assert mock_rpc_device.trigger_ota_update.call_count == 1
# Assert the issue is no longer present
assert not issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 0
async def test_unsupported_firmware_issue_update_not_available(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test repair issues handling when firmware update is not available."""
issue_id = BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID.format(unique=MOCK_MAC)
assert await async_setup_component(hass, "repairs", {})
await hass.async_block_till_done()
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 1
await async_process_repairs_platforms(hass)
client = await hass_client()
result = await start_repair_fix_flow(client, DOMAIN, issue_id)
flow_id = result["flow_id"]
assert result["step_id"] == "confirm"
monkeypatch.setitem(mock_rpc_device.status, "sys", {"available_updates": {}})
result = await process_repair_fix_flow(client, flow_id)
assert result["type"] == "abort"
assert result["reason"] == "update_not_available"
assert mock_rpc_device.trigger_ota_update.call_count == 0
assert issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 1
@pytest.mark.parametrize(
"exception", [DeviceConnectionError, RpcCallError(999, "Unknown error")]
)
async def test_unsupported_firmware_issue_exc(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_rpc_device: Mock,
issue_registry: ir.IssueRegistry,
exception: Exception,
) -> None:
"""Test repair issues handling when OTA update ends with an exception."""
issue_id = BLE_SCANNER_FIRMWARE_UNSUPPORTED_ISSUE_ID.format(unique=MOCK_MAC)
assert await async_setup_component(hass, "repairs", {})
await hass.async_block_till_done()
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 1
await async_process_repairs_platforms(hass)
client = await hass_client()
result = await start_repair_fix_flow(client, DOMAIN, issue_id)
flow_id = result["flow_id"]
assert result["step_id"] == "confirm"
mock_rpc_device.trigger_ota_update.side_effect = exception
result = await process_repair_fix_flow(client, flow_id)
assert result["type"] == "abort"
assert result["reason"] == "cannot_connect"
assert mock_rpc_device.trigger_ota_update.call_count == 1
assert issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 1