Automatic migration from multi-PAN back to Zigbee firmware (#93831)

* Initial implementation of migration back to Zigbee firmware

* Fix typo in `BACKUP_RETRIES` constant name

* Name potentially long-running tasks

* Add an explicit timeout to `_async_wait_until_addon_state`

* Guard against the addon not being installed when uninstalling

* Do not launch the progress flow unless the addon is being installed

* Use a separate translation key for confirmation before disabling multi-PAN

* Disable the bellows UART thread within the ZHA config flow radio manager

* Enhance config flow progress keys for flasher addon installation

* Allow `zha.async_unload_entry` to succeed when ZHA is not loaded

* Do not endlessly spawn task when uninstalling addon synchronously

* Include `uninstall_addon.data.*` in SkyConnect and Yellow translations

* Make `homeassistant_hardware` unit tests pass

* Fix SkyConnect unit test USB mock

* Fix unit tests in related integrations

* Use a separate constant for connection retrying

* Unit test ZHA migration from multi-PAN

* Test ZHA multi-PAN migration helper changes

* Fix flaky SkyConnect unit test being affected by system USB devices

* Unit test the synchronous addon uninstall helper

* Test failure when flasher addon is already running

* Test failure where flasher addon fails to install

* Test ZHA migration failures

* Rename `get_addon_manager` to `get_multiprotocol_addon_manager`

* Remove stray "addon uninstall" comment

* Use better variable names for the two addon managers

* Remove extraneous `self.install_task = None`

* Use the addon manager's `addon_name` instead of constants

* Migrate synchronous addon operations into a new class

* Remove wrapper functions with `finally` clause

* Use a more descriptive error message when the flasher addon is stalled

* Fix existing unit tests

* Remove `wait_until_done`

* Fully replace all addon name constants with those from managers

* Fix OTBR breakage

* Simplify `is_hassio` mocking

* Add missing tests for `check_multi_pan_addon`

* Add missing tests for `multi_pan_addon_using_device`

* Use `waiting` instead of `sync` in class name and methods
This commit is contained in:
puddly 2023-08-28 17:26:34 -04:00 committed by GitHub
parent 23839a7f10
commit c8ef3f9393
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 1383 additions and 252 deletions

View File

@ -5,3 +5,4 @@ import logging
LOGGER = logging.getLogger(__package__) LOGGER = logging.getLogger(__package__)
SILABS_MULTIPROTOCOL_ADDON_SLUG = "core_silabs_multiprotocol" SILABS_MULTIPROTOCOL_ADDON_SLUG = "core_silabs_multiprotocol"
SILABS_FLASHER_ADDON_SLUG = "core_silabs_flasher"

View File

@ -3,10 +3,12 @@ from __future__ import annotations
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
import asyncio import asyncio
from collections.abc import Awaitable
import dataclasses import dataclasses
import logging import logging
from typing import Any, Protocol from typing import Any, Protocol
import async_timeout
import voluptuous as vol import voluptuous as vol
import yarl import yarl
@ -33,17 +35,19 @@ from homeassistant.helpers.selector import (
from homeassistant.helpers.singleton import singleton from homeassistant.helpers.singleton import singleton
from homeassistant.helpers.storage import Store from homeassistant.helpers.storage import Store
from .const import LOGGER, SILABS_MULTIPROTOCOL_ADDON_SLUG from .const import LOGGER, SILABS_FLASHER_ADDON_SLUG, SILABS_MULTIPROTOCOL_ADDON_SLUG
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DATA_ADDON_MANAGER = "silabs_multiprotocol_addon_manager" DATA_MULTIPROTOCOL_ADDON_MANAGER = "silabs_multiprotocol_addon_manager"
DATA_FLASHER_ADDON_MANAGER = "silabs_flasher"
ADDON_SETUP_TIMEOUT = 5 ADDON_STATE_POLL_INTERVAL = 3
ADDON_SETUP_TIMEOUT_ROUNDS = 40 ADDON_INFO_POLL_TIMEOUT = 15 * 60
CONF_ADDON_AUTOFLASH_FW = "autoflash_firmware" CONF_ADDON_AUTOFLASH_FW = "autoflash_firmware"
CONF_ADDON_DEVICE = "device" CONF_ADDON_DEVICE = "device"
CONF_DISABLE_MULTI_PAN = "disable_multi_pan"
CONF_ENABLE_MULTI_PAN = "enable_multi_pan" CONF_ENABLE_MULTI_PAN = "enable_multi_pan"
DEFAULT_CHANNEL = 15 DEFAULT_CHANNEL = 15
@ -55,15 +59,64 @@ STORAGE_VERSION_MINOR = 1
SAVE_DELAY = 10 SAVE_DELAY = 10
@singleton(DATA_ADDON_MANAGER) @singleton(DATA_MULTIPROTOCOL_ADDON_MANAGER)
async def get_addon_manager(hass: HomeAssistant) -> MultiprotocolAddonManager: async def get_multiprotocol_addon_manager(
hass: HomeAssistant,
) -> MultiprotocolAddonManager:
"""Get the add-on manager.""" """Get the add-on manager."""
manager = MultiprotocolAddonManager(hass) manager = MultiprotocolAddonManager(hass)
await manager.async_setup() await manager.async_setup()
return manager return manager
class MultiprotocolAddonManager(AddonManager): class WaitingAddonManager(AddonManager):
"""Addon manager which supports waiting operations for managing an addon."""
async def async_wait_until_addon_state(self, *states: AddonState) -> None:
"""Poll an addon's info until it is in a specific state."""
async with async_timeout.timeout(ADDON_INFO_POLL_TIMEOUT):
while True:
try:
info = await self.async_get_addon_info()
except AddonError:
info = None
_LOGGER.debug("Waiting for addon to be in state %s: %s", states, info)
if info is not None and info.state in states:
break
await asyncio.sleep(ADDON_STATE_POLL_INTERVAL)
async def async_start_addon_waiting(self) -> None:
"""Start an add-on."""
await self.async_schedule_start_addon()
await self.async_wait_until_addon_state(AddonState.RUNNING)
async def async_install_addon_waiting(self) -> None:
"""Install an add-on."""
await self.async_schedule_install_addon()
await self.async_wait_until_addon_state(
AddonState.RUNNING,
AddonState.NOT_RUNNING,
)
async def async_uninstall_addon_waiting(self) -> None:
"""Uninstall an add-on."""
try:
info = await self.async_get_addon_info()
except AddonError:
info = None
# Do not try to uninstall an addon if it is already uninstalled
if info is not None and info.state == AddonState.NOT_INSTALLED:
return
await self.async_uninstall_addon()
await self.async_wait_until_addon_state(AddonState.NOT_INSTALLED)
class MultiprotocolAddonManager(WaitingAddonManager):
"""Silicon Labs Multiprotocol add-on manager.""" """Silicon Labs Multiprotocol add-on manager."""
def __init__(self, hass: HomeAssistant) -> None: def __init__(self, hass: HomeAssistant) -> None:
@ -207,6 +260,18 @@ class MultipanProtocol(Protocol):
""" """
@singleton(DATA_FLASHER_ADDON_MANAGER)
@callback
def get_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager:
"""Get the flasher add-on manager."""
return WaitingAddonManager(
hass,
LOGGER,
"Silicon Labs Flasher",
SILABS_FLASHER_ADDON_SLUG,
)
@dataclasses.dataclass @dataclasses.dataclass
class SerialPortSettings: class SerialPortSettings:
"""Serial port settings.""" """Serial port settings."""
@ -242,9 +307,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
ZhaMultiPANMigrationHelper, ZhaMultiPANMigrationHelper,
) )
# If we install the add-on we should uninstall it on entry remove.
self.install_task: asyncio.Task | None = None self.install_task: asyncio.Task | None = None
self.start_task: asyncio.Task | None = None self.start_task: asyncio.Task | None = None
self.stop_task: asyncio.Task | None = None
self._zha_migration_mgr: ZhaMultiPANMigrationHelper | None = None self._zha_migration_mgr: ZhaMultiPANMigrationHelper | None = None
self.config_entry = config_entry self.config_entry = config_entry
self.original_addon_config: dict[str, Any] | None = None self.original_addon_config: dict[str, Any] | None = None
@ -275,37 +340,37 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
"""Return the correct flow manager.""" """Return the correct flow manager."""
return self.hass.config_entries.options return self.hass.config_entries.options
async def _async_get_addon_info(self) -> AddonInfo: async def _resume_flow_when_done(self, awaitable: Awaitable) -> None:
try:
await awaitable
finally:
self.hass.async_create_task(
self.flow_manager.async_configure(flow_id=self.flow_id)
)
async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo:
"""Return and cache Silicon Labs Multiprotocol add-on info.""" """Return and cache Silicon Labs Multiprotocol add-on info."""
addon_manager: AddonManager = await get_addon_manager(self.hass)
try: try:
addon_info: AddonInfo = await addon_manager.async_get_addon_info() addon_info: AddonInfo = await addon_manager.async_get_addon_info()
except AddonError as err: except AddonError as err:
_LOGGER.error(err) _LOGGER.error(err)
raise AbortFlow("addon_info_failed") from err raise AbortFlow(
"addon_info_failed",
description_placeholders={"addon_name": addon_manager.addon_name},
) from err
return addon_info return addon_info
async def _async_set_addon_config(self, config: dict) -> None: async def _async_set_addon_config(
self, config: dict, addon_manager: AddonManager
) -> None:
"""Set Silicon Labs Multiprotocol add-on config.""" """Set Silicon Labs Multiprotocol add-on config."""
addon_manager: AddonManager = await get_addon_manager(self.hass)
try: try:
await addon_manager.async_set_addon_options(config) await addon_manager.async_set_addon_options(config)
except AddonError as err: except AddonError as err:
_LOGGER.error(err) _LOGGER.error(err)
raise AbortFlow("addon_set_config_failed") from err raise AbortFlow("addon_set_config_failed") from err
async def _async_install_addon(self) -> None:
"""Install the Silicon Labs Multiprotocol add-on."""
addon_manager: AddonManager = await get_addon_manager(self.hass)
try:
await addon_manager.async_schedule_install_addon()
finally:
# Continue the flow after show progress when the task is done.
self.hass.async_create_task(
self.flow_manager.async_configure(flow_id=self.flow_id)
)
async def async_step_init( async def async_step_init(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
@ -319,7 +384,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Handle logic when on Supervisor host.""" """Handle logic when on Supervisor host."""
addon_info = await self._async_get_addon_info() multipan_manager = await get_multiprotocol_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(multipan_manager)
if addon_info.state == AddonState.NOT_INSTALLED: if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_addon_not_installed() return await self.async_step_addon_not_installed()
@ -347,19 +413,26 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
) -> FlowResult: ) -> FlowResult:
"""Install Silicon Labs Multiprotocol add-on.""" """Install Silicon Labs Multiprotocol add-on."""
if not self.install_task: if not self.install_task:
self.install_task = self.hass.async_create_task(self._async_install_addon()) multipan_manager = await get_multiprotocol_addon_manager(self.hass)
self.install_task = self.hass.async_create_task(
self._resume_flow_when_done(
multipan_manager.async_install_addon_waiting()
),
"SiLabs Multiprotocol addon install",
)
return self.async_show_progress( return self.async_show_progress(
step_id="install_addon", progress_action="install_addon" step_id="install_addon",
progress_action="install_addon",
description_placeholders={"addon_name": multipan_manager.addon_name},
) )
try: try:
await self.install_task await self.install_task
except AddonError as err: except AddonError as err:
self.install_task = None
_LOGGER.error(err) _LOGGER.error(err)
return self.async_show_progress_done(next_step_id="install_failed") return self.async_show_progress_done(next_step_id="install_failed")
finally:
self.install_task = None self.install_task = None
return self.async_show_progress_done(next_step_id="configure_addon") return self.async_show_progress_done(next_step_id="configure_addon")
@ -367,7 +440,11 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Add-on installation failed.""" """Add-on installation failed."""
return self.async_abort(reason="addon_install_failed") multipan_manager = await get_multiprotocol_addon_manager(self.hass)
return self.async_abort(
reason="addon_install_failed",
description_placeholders={"addon_name": multipan_manager.addon_name},
)
async def async_step_configure_addon( async def async_step_configure_addon(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -386,7 +463,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
async_get_channel as async_get_zha_channel, async_get_channel as async_get_zha_channel,
) )
addon_info = await self._async_get_addon_info() multipan_manager = await get_multiprotocol_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(multipan_manager)
addon_config = addon_info.options addon_config = addon_info.options
@ -426,14 +504,14 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
multipan_channel = zha_channel multipan_channel = zha_channel
# Initialize the shared channel # Initialize the shared channel
multipan_manager = await get_addon_manager(self.hass) multipan_manager = await get_multiprotocol_addon_manager(self.hass)
multipan_manager.async_set_channel(multipan_channel) multipan_manager.async_set_channel(multipan_channel)
if new_addon_config != addon_config: if new_addon_config != addon_config:
# Copy the add-on config to keep the objects separate. # Copy the add-on config to keep the objects separate.
self.original_addon_config = dict(addon_config) self.original_addon_config = dict(addon_config)
_LOGGER.debug("Reconfiguring addon with %s", new_addon_config) _LOGGER.debug("Reconfiguring addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config) await self._async_set_addon_config(new_addon_config, multipan_manager)
return await self.async_step_start_addon() return await self.async_step_start_addon()
@ -442,9 +520,16 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
) -> FlowResult: ) -> FlowResult:
"""Start Silicon Labs Multiprotocol add-on.""" """Start Silicon Labs Multiprotocol add-on."""
if not self.start_task: if not self.start_task:
self.start_task = self.hass.async_create_task(self._async_start_addon()) multipan_manager = await get_multiprotocol_addon_manager(self.hass)
self.start_task = self.hass.async_create_task(
self._resume_flow_when_done(
multipan_manager.async_start_addon_waiting()
)
)
return self.async_show_progress( return self.async_show_progress(
step_id="start_addon", progress_action="start_addon" step_id="start_addon",
progress_action="start_addon",
description_placeholders={"addon_name": multipan_manager.addon_name},
) )
try: try:
@ -461,18 +546,11 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Add-on start failed.""" """Add-on start failed."""
return self.async_abort(reason="addon_start_failed") multipan_manager = await get_multiprotocol_addon_manager(self.hass)
return self.async_abort(
async def _async_start_addon(self) -> None: reason="addon_start_failed",
"""Start Silicon Labs Multiprotocol add-on.""" description_placeholders={"addon_name": multipan_manager.addon_name},
addon_manager: AddonManager = await get_addon_manager(self.hass) )
try:
await addon_manager.async_schedule_start_addon()
finally:
# Continue the flow after show progress when the task is done.
self.hass.async_create_task(
self.flow_manager.async_configure(flow_id=self.flow_id)
)
async def async_step_finish_addon_setup( async def async_step_finish_addon_setup(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -493,16 +571,25 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
return self.async_create_entry(title="", data={}) return self.async_create_entry(title="", data={})
async def async_step_addon_installed_other_device(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Show dialog explaining the addon is in use by another device."""
if user_input is None:
return self.async_show_form(step_id="addon_installed_other_device")
return self.async_create_entry(title="", data={})
async def async_step_addon_installed( async def async_step_addon_installed(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Handle logic when the addon is already installed.""" """Handle logic when the addon is already installed."""
addon_info = await self._async_get_addon_info() multipan_manager = await get_multiprotocol_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(multipan_manager)
serial_device = (await self._async_serial_port_settings()).device serial_device = (await self._async_serial_port_settings()).device
if addon_info.options.get(CONF_ADDON_DEVICE) == serial_device: if addon_info.options.get(CONF_ADDON_DEVICE) != serial_device:
return await self.async_step_show_addon_menu() return await self.async_step_addon_installed_other_device()
return await self.async_step_addon_installed_other_device() return await self.async_step_show_addon_menu()
async def async_step_show_addon_menu( async def async_step_show_addon_menu(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -520,7 +607,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Reconfigure the addon.""" """Reconfigure the addon."""
multipan_manager = await get_addon_manager(self.hass) multipan_manager = await get_multiprotocol_addon_manager(self.hass)
active_platforms = await multipan_manager.async_active_platforms() active_platforms = await multipan_manager.async_active_platforms()
if set(active_platforms) != {"otbr", "zha"}: if set(active_platforms) != {"otbr", "zha"}:
return await self.async_step_notify_unknown_multipan_user() return await self.async_step_notify_unknown_multipan_user()
@ -540,7 +627,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Change the channel.""" """Change the channel."""
multipan_manager = await get_addon_manager(self.hass) multipan_manager = await get_multiprotocol_addon_manager(self.hass)
if user_input is None: if user_input is None:
channels = [str(x) for x in range(11, 27)] channels = [str(x) for x in range(11, 27)]
suggested_channel = DEFAULT_CHANNEL suggested_channel = DEFAULT_CHANNEL
@ -584,23 +671,217 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC):
async def async_step_uninstall_addon( async def async_step_uninstall_addon(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Uninstall the addon (not implemented).""" """Uninstall the addon and revert the firmware."""
return await self.async_step_show_revert_guide() if user_input is None:
return self.async_show_form(
step_id="uninstall_addon",
data_schema=vol.Schema(
{vol.Required(CONF_DISABLE_MULTI_PAN, default=False): bool}
),
description_placeholders={"hardware_name": self._hardware_name()},
)
if not user_input[CONF_DISABLE_MULTI_PAN]:
return self.async_create_entry(title="", data={})
async def async_step_show_revert_guide( return await self.async_step_firmware_revert()
async def async_step_firmware_revert(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Link to a guide for reverting to Zigbee firmware.""" """Install the flasher addon, if necessary."""
if user_input is None:
return self.async_show_form(step_id="show_revert_guide")
return self.async_create_entry(title="", data={})
async def async_step_addon_installed_other_device( flasher_manager = get_flasher_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(flasher_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_flasher_addon()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_configure_flasher_addon()
# If the addon is already installed and running, fail
return self.async_abort(
reason="addon_already_running",
description_placeholders={"addon_name": flasher_manager.addon_name},
)
async def async_step_install_flasher_addon(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Show dialog explaining the addon is in use by another device.""" """Show progress dialog for installing flasher addon."""
if user_input is None: flasher_manager = get_flasher_addon_manager(self.hass)
return self.async_show_form(step_id="addon_installed_other_device") addon_info = await self._async_get_addon_info(flasher_manager)
_LOGGER.debug("Flasher addon state: %s", addon_info)
if not self.install_task:
self.install_task = self.hass.async_create_task(
self._resume_flow_when_done(
flasher_manager.async_install_addon_waiting()
),
"SiLabs Flasher addon install",
)
return self.async_show_progress(
step_id="install_flasher_addon",
progress_action="install_addon",
description_placeholders={"addon_name": flasher_manager.addon_name},
)
try:
await self.install_task
except AddonError as err:
_LOGGER.error(err)
return self.async_show_progress_done(next_step_id="install_failed")
finally:
self.install_task = None
return self.async_show_progress_done(next_step_id="configure_flasher_addon")
async def async_step_configure_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Perform initial backup and reconfigure ZHA."""
# pylint: disable-next=import-outside-toplevel
from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN
# pylint: disable-next=import-outside-toplevel
from homeassistant.components.zha.radio_manager import (
ZhaMultiPANMigrationHelper,
)
zha_entries = self.hass.config_entries.async_entries(ZHA_DOMAIN)
new_settings = await self._async_serial_port_settings()
_LOGGER.debug("Using new ZHA settings: %s", new_settings)
if zha_entries:
zha_migration_mgr = ZhaMultiPANMigrationHelper(self.hass, zha_entries[0])
migration_data = {
"new_discovery_info": {
"name": self._hardware_name(),
"port": {
"path": new_settings.device,
"baudrate": int(new_settings.baudrate),
"flow_control": (
"hardware" if new_settings.flow_control else None
),
},
"radio_type": "ezsp",
},
"old_discovery_info": {
"hw": {
"name": self._zha_name(),
"port": {"path": get_zigbee_socket()},
"radio_type": "ezsp",
}
},
}
_LOGGER.debug("Starting ZHA migration with: %s", migration_data)
try:
if await zha_migration_mgr.async_initiate_migration(migration_data):
self._zha_migration_mgr = zha_migration_mgr
except Exception as err:
_LOGGER.exception("Unexpected exception during ZHA migration")
raise AbortFlow("zha_migration_failed") from err
flasher_manager = get_flasher_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(flasher_manager)
new_addon_config = {
**addon_info.options,
"device": new_settings.device,
"flow_control": new_settings.flow_control,
}
_LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config, flasher_manager)
return await self.async_step_uninstall_multiprotocol_addon()
async def async_step_uninstall_multiprotocol_addon(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Uninstall Silicon Labs Multiprotocol add-on."""
if not self.stop_task:
multipan_manager = await get_multiprotocol_addon_manager(self.hass)
self.stop_task = self.hass.async_create_task(
self._resume_flow_when_done(
multipan_manager.async_uninstall_addon_waiting()
),
"SiLabs Multiprotocol addon uninstall",
)
return self.async_show_progress(
step_id="uninstall_multiprotocol_addon",
progress_action="uninstall_multiprotocol_addon",
description_placeholders={"addon_name": multipan_manager.addon_name},
)
try:
await self.stop_task
finally:
self.stop_task = None
return self.async_show_progress_done(next_step_id="start_flasher_addon")
async def async_step_start_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Start Silicon Labs Flasher add-on."""
if not self.start_task:
flasher_manager = get_flasher_addon_manager(self.hass)
async def start_and_wait_until_done() -> None:
await flasher_manager.async_start_addon_waiting()
# Now that the addon is running, wait for it to finish
await flasher_manager.async_wait_until_addon_state(
AddonState.NOT_RUNNING
)
self.start_task = self.hass.async_create_task(
self._resume_flow_when_done(start_and_wait_until_done())
)
return self.async_show_progress(
step_id="start_flasher_addon",
progress_action="start_flasher_addon",
description_placeholders={"addon_name": flasher_manager.addon_name},
)
try:
await self.start_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
return self.async_show_progress_done(next_step_id="flasher_failed")
finally:
self.start_task = None
return self.async_show_progress_done(next_step_id="flashing_complete")
async def async_step_flasher_failed(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Flasher add-on start failed."""
flasher_manager = get_flasher_addon_manager(self.hass)
return self.async_abort(
reason="addon_start_failed",
description_placeholders={"addon_name": flasher_manager.addon_name},
)
async def async_step_flashing_complete(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Finish flashing and update the config entry."""
flasher_manager = get_flasher_addon_manager(self.hass)
await flasher_manager.async_uninstall_addon_waiting()
# Finish ZHA migration if needed
if self._zha_migration_mgr:
try:
await self._zha_migration_mgr.async_finish_migration()
except Exception as err:
_LOGGER.exception("Unexpected exception during ZHA migration")
raise AbortFlow("zha_migration_failed") from err
return self.async_create_entry(title="", data={}) return self.async_create_entry(title="", data={})
@ -613,18 +894,18 @@ async def check_multi_pan_addon(hass: HomeAssistant) -> None:
if not is_hassio(hass): if not is_hassio(hass):
return return
addon_manager: AddonManager = await get_addon_manager(hass) multipan_manager = await get_multiprotocol_addon_manager(hass)
try: try:
addon_info: AddonInfo = await addon_manager.async_get_addon_info() addon_info: AddonInfo = await multipan_manager.async_get_addon_info()
except AddonError as err: except AddonError as err:
_LOGGER.error(err) _LOGGER.error(err)
raise HomeAssistantError from err raise HomeAssistantError from err
# Request the addon to start if it's not started # Request the addon to start if it's not started
# addon_manager.async_start_addon returns as soon as the start request has been sent # `async_start_addon` returns as soon as the start request has been sent
# and does not wait for the addon to be started, so we raise below # and does not wait for the addon to be started, so we raise below
if addon_info.state == AddonState.NOT_RUNNING: if addon_info.state == AddonState.NOT_RUNNING:
await addon_manager.async_start_addon() await multipan_manager.async_start_addon()
if addon_info.state not in (AddonState.NOT_INSTALLED, AddonState.RUNNING): if addon_info.state not in (AddonState.NOT_INSTALLED, AddonState.RUNNING):
_LOGGER.debug("Multi pan addon installed and in state %s", addon_info.state) _LOGGER.debug("Multi pan addon installed and in state %s", addon_info.state)
@ -640,8 +921,8 @@ async def multi_pan_addon_using_device(hass: HomeAssistant, device_path: str) ->
if not is_hassio(hass): if not is_hassio(hass):
return False return False
addon_manager: AddonManager = await get_addon_manager(hass) multipan_manager = await get_multiprotocol_addon_manager(hass)
addon_info: AddonInfo = await addon_manager.async_get_addon_info() addon_info: AddonInfo = await multipan_manager.async_get_addon_info()
if addon_info.state != AddonState.RUNNING: if addon_info.state != AddonState.RUNNING:
return False return False

View File

@ -39,31 +39,42 @@
"reconfigure_addon": { "reconfigure_addon": {
"title": "Reconfigure IEEE 802.15.4 radio multiprotocol support" "title": "Reconfigure IEEE 802.15.4 radio multiprotocol support"
}, },
"show_revert_guide": {
"title": "Multiprotocol support is enabled for this device",
"description": "If you want to change to Zigbee only firmware, please complete the following manual steps:\n\n * Remove the Silicon Labs Multiprotocol addon\n\n * Flash the Zigbee only firmware, follow the guide at https://github.com/NabuCasa/silabs-firmware/wiki/Flash-Silicon-Labs-radio-firmware-manually.\n\n * Reconfigure ZHA to migrate settings to the reflashed radio"
},
"start_addon": { "start_addon": {
"title": "The Silicon Labs Multiprotocol add-on is starting." "title": "The Silicon Labs Multiprotocol add-on is starting."
}, },
"uninstall_addon": { "uninstall_addon": {
"title": "Remove IEEE 802.15.4 radio multiprotocol support." "title": "Remove IEEE 802.15.4 radio multiprotocol support",
"description": "Disabling multiprotocol support will revert your {hardware_name}'s radio back to Zigbee-only firmware and will disable Thread support provided by the {hardware_name}. Your Thread devices will continue working only if you have another Thread border router nearby.\n\nIt will take a few minutes to install the Zigbee firmware and restoring a backup.",
"data": {
"disable_multi_pan": "Disable multiprotocol support"
}
},
"install_flasher_addon": {
"title": "The Silicon Labs Flasher add-on installation has started"
},
"configure_flasher_addon": {
"title": "The Silicon Labs Flasher add-on installation has started"
},
"start_flasher_addon": {
"title": "Installing firmware",
"description": "Zigbee firmware is now being installed. This will take a few minutes."
} }
}, },
"error": { "error": {
"unknown": "[%key:common::config_flow::error::unknown%]" "unknown": "[%key:common::config_flow::error::unknown%]"
}, },
"abort": { "abort": {
"addon_info_failed": "Failed to get Silicon Labs Multiprotocol add-on info.", "addon_info_failed": "Failed to get {addon_name} add-on info.",
"addon_install_failed": "Failed to install the Silicon Labs Multiprotocol add-on.", "addon_install_failed": "Failed to install the {addon_name} add-on.",
"addon_set_config_failed": "Failed to set Silicon Labs Multiprotocol configuration.", "addon_already_running": "Failed to start the {addon_name} add-on because it is already running.",
"addon_start_failed": "Failed to start the Silicon Labs Multiprotocol add-on.", "addon_set_config_failed": "Failed to set {addon_name} configuration.",
"addon_start_failed": "Failed to start the {addon_name} add-on.",
"not_hassio": "The hardware options can only be configured on HassOS installations.", "not_hassio": "The hardware options can only be configured on HassOS installations.",
"zha_migration_failed": "The ZHA migration did not succeed." "zha_migration_failed": "The ZHA migration did not succeed."
}, },
"progress": { "progress": {
"install_addon": "Please wait while the Silicon Labs Multiprotocol add-on installation finishes. This can take several minutes.", "install_addon": "Please wait while the {addon_name} add-on installation finishes. This can take several minutes.",
"start_addon": "Please wait while the Silicon Labs Multiprotocol add-on start completes. This may take some seconds." "start_addon": "Please wait while the {addon_name} add-on start completes. This may take some seconds."
} }
} }
} }

View File

@ -38,15 +38,25 @@
"reconfigure_addon": { "reconfigure_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]" "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]"
}, },
"show_revert_guide": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::description%]"
},
"start_addon": { "start_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]" "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]"
}, },
"uninstall_addon": { "uninstall_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::description%]",
"data": {
"disable_multi_pan": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::data::disable_multi_pan%]"
}
},
"install_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_flasher_addon::title%]"
},
"configure_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::configure_flasher_addon::title%]"
},
"start_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]"
} }
}, },
"error": { "error": {
@ -55,6 +65,7 @@
"abort": { "abort": {
"addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]", "addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]",
"addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]", "addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]",
"addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]",
"addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]", "addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]",
"addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]", "addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]",
"not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]", "not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]",

View File

@ -60,15 +60,25 @@
"reconfigure_addon": { "reconfigure_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]" "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]"
}, },
"show_revert_guide": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::description%]"
},
"start_addon": { "start_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]" "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]"
}, },
"uninstall_addon": { "uninstall_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::description%]",
"data": {
"disable_multi_pan": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::data::disable_multi_pan%]"
}
},
"install_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_flasher_addon::title%]"
},
"configure_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::configure_flasher_addon::title%]"
},
"start_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]"
} }
}, },
"error": { "error": {
@ -77,6 +87,7 @@
"abort": { "abort": {
"addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]", "addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]",
"addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]", "addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]",
"addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]",
"addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]", "addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]",
"addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]", "addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]",
"not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]", "not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]",

View File

@ -14,7 +14,7 @@ from python_otbr_api.tlv_parser import MeshcopTLVType
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
MultiprotocolAddonManager, MultiprotocolAddonManager,
get_addon_manager, get_multiprotocol_addon_manager,
is_multiprotocol_url, is_multiprotocol_url,
multi_pan_addon_using_device, multi_pan_addon_using_device,
) )
@ -146,8 +146,10 @@ async def get_allowed_channel(hass: HomeAssistant, otbr_url: str) -> int | None:
# The OTBR is not sharing the radio, no restriction # The OTBR is not sharing the radio, no restriction
return None return None
addon_manager: MultiprotocolAddonManager = await get_addon_manager(hass) multipan_manager: MultiprotocolAddonManager = await get_multiprotocol_addon_manager(
return addon_manager.async_get_channel() hass
)
return multipan_manager.async_get_channel()
async def _warn_on_channel_collision( async def _warn_on_channel_collision(

View File

@ -166,7 +166,11 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload ZHA config entry.""" """Unload ZHA config entry."""
zha_gateway: ZHAGateway = hass.data[DATA_ZHA].pop(DATA_ZHA_GATEWAY) try:
zha_gateway: ZHAGateway = hass.data[DATA_ZHA].pop(DATA_ZHA_GATEWAY)
except KeyError:
return False
await zha_gateway.shutdown() await zha_gateway.shutdown()
GROUP_PROBE.cleanup() GROUP_PROBE.cleanup()

View File

@ -96,10 +96,12 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]:
yellow_radio.manufacturer = "Nabu Casa" yellow_radio.manufacturer = "Nabu Casa"
# Present the multi-PAN addon as a setup option, if it's available # Present the multi-PAN addon as a setup option, if it's available
addon_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager(
hass
)
try: try:
addon_info = await addon_manager.async_get_addon_info() addon_info = await multipan_manager.async_get_addon_info()
except (AddonError, KeyError): except (AddonError, KeyError):
addon_info = None addon_info = None

View File

@ -9,6 +9,7 @@ import logging
import os import os
from typing import Any from typing import Any
from bellows.config import CONF_USE_THREAD
import voluptuous as vol import voluptuous as vol
from zigpy.application import ControllerApplication from zigpy.application import ControllerApplication
import zigpy.backups import zigpy.backups
@ -47,7 +48,9 @@ RECOMMENDED_RADIOS = (
) )
CONNECT_DELAY_S = 1.0 CONNECT_DELAY_S = 1.0
RETRY_DELAY_S = 1.0
BACKUP_RETRIES = 5
MIGRATION_RETRIES = 100 MIGRATION_RETRIES = 100
HARDWARE_DISCOVERY_SCHEMA = vol.Schema( HARDWARE_DISCOVERY_SCHEMA = vol.Schema(
@ -134,6 +137,7 @@ class ZhaRadioManager:
app_config[CONF_DATABASE] = database_path app_config[CONF_DATABASE] = database_path
app_config[CONF_DEVICE] = self.device_settings app_config[CONF_DEVICE] = self.device_settings
app_config[CONF_NWK_BACKUP_ENABLED] = False app_config[CONF_NWK_BACKUP_ENABLED] = False
app_config[CONF_USE_THREAD] = False
app_config = self.radio_type.controller.SCHEMA(app_config) app_config = self.radio_type.controller.SCHEMA(app_config)
app = await self.radio_type.controller.new( app = await self.radio_type.controller.new(
@ -341,7 +345,24 @@ class ZhaMultiPANMigrationHelper:
old_radio_mgr.device_path = config_entry_data[CONF_DEVICE][CONF_DEVICE_PATH] old_radio_mgr.device_path = config_entry_data[CONF_DEVICE][CONF_DEVICE_PATH]
old_radio_mgr.device_settings = config_entry_data[CONF_DEVICE] old_radio_mgr.device_settings = config_entry_data[CONF_DEVICE]
old_radio_mgr.radio_type = RadioType[config_entry_data[CONF_RADIO_TYPE]] old_radio_mgr.radio_type = RadioType[config_entry_data[CONF_RADIO_TYPE]]
backup = await old_radio_mgr.async_load_network_settings(create_backup=True)
for retry in range(BACKUP_RETRIES):
try:
backup = await old_radio_mgr.async_load_network_settings(
create_backup=True
)
break
except OSError as err:
if retry >= BACKUP_RETRIES - 1:
raise
_LOGGER.debug(
"Failed to create backup %r, retrying in %s seconds",
err,
RETRY_DELAY_S,
)
await asyncio.sleep(RETRY_DELAY_S)
# Then configure the radio manager for the new radio to use the new settings # Then configure the radio manager for the new radio to use the new settings
self._radio_mgr.chosen_backup = backup self._radio_mgr.chosen_backup = backup
@ -381,10 +402,10 @@ class ZhaMultiPANMigrationHelper:
_LOGGER.debug( _LOGGER.debug(
"Failed to restore backup %r, retrying in %s seconds", "Failed to restore backup %r, retrying in %s seconds",
err, err,
CONNECT_DELAY_S, RETRY_DELAY_S,
) )
await asyncio.sleep(CONNECT_DELAY_S) await asyncio.sleep(RETRY_DELAY_S)
_LOGGER.debug("Restored backup after %s retries", retry) _LOGGER.debug("Restored backup after %s retries", retry)

View File

@ -147,3 +147,21 @@ def start_addon_fixture():
"homeassistant.components.hassio.addon_manager.async_start_addon" "homeassistant.components.hassio.addon_manager.async_start_addon"
) as start_addon: ) as start_addon:
yield start_addon yield start_addon
@pytest.fixture(name="stop_addon")
def stop_addon_fixture():
"""Mock stop add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_stop_addon"
) as stop_addon:
yield stop_addon
@pytest.fixture(name="uninstall_addon")
def uninstall_addon_fixture():
"""Mock uninstall add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_uninstall_addon"
) as uninstall_addon:
yield uninstall_addon

View File

@ -9,7 +9,7 @@ import pytest
def mock_usb_serial_by_id_fixture() -> Generator[MagicMock, None, None]: def mock_usb_serial_by_id_fixture() -> Generator[MagicMock, None, None]:
"""Mock usb serial by id.""" """Mock usb serial by id."""
with patch( with patch(
"homeassistant.components.zwave_js.config_flow.usb.get_serial_by_id" "homeassistant.components.zha.config_flow.usb.get_serial_by_id"
) as mock_usb_serial_by_id: ) as mock_usb_serial_by_id:
mock_usb_serial_by_id.side_effect = lambda x: x mock_usb_serial_by_id.side_effect = lambda x: x
yield mock_usb_serial_by_id yield mock_usb_serial_by_id
@ -149,3 +149,21 @@ def start_addon_fixture():
"homeassistant.components.hassio.addon_manager.async_start_addon" "homeassistant.components.hassio.addon_manager.async_start_addon"
) as start_addon: ) as start_addon:
yield start_addon yield start_addon
@pytest.fixture(name="stop_addon")
def stop_addon_fixture():
"""Mock stop add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_stop_addon"
) as stop_addon:
yield stop_addon
@pytest.fixture(name="uninstall_addon")
def uninstall_addon_fixture():
"""Mock uninstall add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_uninstall_addon"
) as uninstall_addon:
yield uninstall_addon

View File

@ -1,7 +1,10 @@
"""Test the Home Assistant SkyConnect config flow.""" """Test the Home Assistant SkyConnect config flow."""
from collections.abc import Generator
import copy import copy
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
import pytest
from homeassistant.components import homeassistant_sky_connect, usb from homeassistant.components import homeassistant_sky_connect, usb
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.components.zha.core.const import ( from homeassistant.components.zha.core.const import (
@ -25,6 +28,15 @@ USB_DATA = usb.UsbServiceInfo(
) )
@pytest.fixture(autouse=True)
def config_flow_handler(hass: HomeAssistant) -> Generator[None, None, None]:
"""Fixture for a test config flow."""
with patch(
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.WaitingAddonManager.async_wait_until_addon_state"
):
yield
async def test_config_flow(hass: HomeAssistant) -> None: async def test_config_flow(hass: HomeAssistant) -> None:
"""Test the config flow.""" """Test the config flow."""
with patch( with patch(

View File

@ -24,6 +24,13 @@ CONFIG_ENTRY_DATA = {
} }
@pytest.fixture(autouse=True)
def disable_usb_probing() -> Generator[None, None, None]:
"""Disallow touching of system USB devices during unit tests."""
with patch("homeassistant.components.usb.comports", return_value=[]):
yield
@pytest.fixture @pytest.fixture
def mock_zha_config_flow_setup() -> Generator[None, None, None]: def mock_zha_config_flow_setup() -> Generator[None, None, None]:
"""Mock the radio connection and probing of the ZHA config flow.""" """Mock the radio connection and probing of the ZHA config flow."""

View File

@ -1,4 +1,5 @@
"""Test the Home Assistant Yellow config flow.""" """Test the Home Assistant Yellow config flow."""
from collections.abc import Generator
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
import pytest import pytest
@ -11,6 +12,15 @@ from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry, MockModule, mock_integration from tests.common import MockConfigEntry, MockModule, mock_integration
@pytest.fixture(autouse=True)
def config_flow_handler(hass: HomeAssistant) -> Generator[None, None, None]:
"""Fixture for a test config flow."""
with patch(
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.WaitingAddonManager.async_wait_until_addon_state"
):
yield
@pytest.fixture(name="get_yellow_settings") @pytest.fixture(name="get_yellow_settings")
def mock_get_yellow_settings(): def mock_get_yellow_settings():
"""Mock getting yellow settings.""" """Mock getting yellow settings."""

View File

@ -31,7 +31,9 @@ def disable_platform_only():
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def reduce_reconnect_timeout(): def reduce_reconnect_timeout():
"""Reduces reconnect timeout to speed up tests.""" """Reduces reconnect timeout to speed up tests."""
with patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.0001): with patch(
"homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.0001
), patch("homeassistant.components.zha.radio_manager.RETRY_DELAY_S", 0.0001):
yield yield
@ -83,7 +85,7 @@ def com_port(device="/dev/ttyUSB1234"):
@pytest.fixture @pytest.fixture
def mock_connect_zigpy_app() -> Generator[None, None, None]: def mock_connect_zigpy_app() -> Generator[MagicMock, None, None]:
"""Mock the radio connection.""" """Mock the radio connection."""
mock_connect_app = MagicMock() mock_connect_app = MagicMock()
@ -96,7 +98,7 @@ def mock_connect_zigpy_app() -> Generator[None, None, None]:
"homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app", "homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
return_value=mock_connect_app, return_value=mock_connect_app,
): ):
yield yield mock_connect_app
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
@ -370,3 +372,52 @@ async def test_migrate_non_matching_port(
"radio_type": "ezsp", "radio_type": "ezsp",
} }
assert config_entry.title == "Test" assert config_entry.title == "Test"
async def test_migrate_initiate_failure(
hass: HomeAssistant,
mock_connect_zigpy_app,
) -> None:
"""Test retries with failure."""
# Set up the config entry
config_entry = MockConfigEntry(
data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
domain=DOMAIN,
options={},
title="Test",
)
config_entry.add_to_hass(hass)
config_entry.state = config_entries.ConfigEntryState.SETUP_IN_PROGRESS
migration_data = {
"new_discovery_info": {
"name": "Test Updated",
"port": {
"path": "socket://some/virtual_port",
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "efr32",
},
"old_discovery_info": {
"hw": {
"name": "Test",
"port": {
"path": "/dev/ttyTEST123",
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "efr32",
}
},
}
mock_load_info = AsyncMock(side_effect=OSError())
mock_connect_zigpy_app.__aenter__.return_value.load_network_info = mock_load_info
migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry)
with pytest.raises(OSError):
await migration_helper.async_initiate_migration(migration_data)
assert len(mock_load_info.mock_calls) == radio_manager.BACKUP_RETRIES