From 4c212bdcd441a64394647d7ccf69ce51bdba576e Mon Sep 17 00:00:00 2001 From: Petar Petrov Date: Thu, 18 Sep 2025 18:33:06 +0300 Subject: [PATCH] Enable thread migration for ZBT integration (#152550) --- .../firmware_config_flow.py | 93 +++++++++---------- .../test_config_flow_failures.py | 41 -------- 2 files changed, 44 insertions(+), 90 deletions(-) diff --git a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py index 69f29098208..7f57350cc99 100644 --- a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py +++ b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py @@ -288,6 +288,45 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): return self.async_show_progress_done(next_step_id=next_step_id) + async def _configure_and_start_otbr_addon(self) -> None: + """Configure and start the OTBR addon.""" + + # Before we start the addon, confirm that the correct firmware is running + # and populate `self._probed_firmware_info` with the correct information + if not await self._probe_firmware_info(probe_methods=(ApplicationType.SPINEL,)): + raise AbortFlow( + "unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + + otbr_manager = get_otbr_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(otbr_manager) + + assert self._device is not None + new_addon_config = { + **addon_info.options, + "device": self._device, + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": False, + } + + _LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config) + + try: + await otbr_manager.async_set_addon_options(new_addon_config) + except AddonError as err: + _LOGGER.error(err) + raise AbortFlow( + "addon_set_config_failed", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": otbr_manager.addon_name, + }, + ) from err + + await otbr_manager.async_start_addon_waiting() + async def async_step_firmware_download_failed( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -473,18 +512,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): return await self.async_step_install_otbr_addon() if addon_info.state == AddonState.RUNNING: - # We only fail setup if we have an instance of OTBR running *and* it's - # pointing to different hardware - if addon_info.options["device"] != self._device: - return self.async_abort( - reason="otbr_addon_already_running", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": otbr_manager.addon_name, - }, - ) - - # Otherwise, stop the addon before continuing to flash firmware + # Stop the addon before continuing to flash firmware await otbr_manager.async_stop_addon() return None @@ -553,43 +581,8 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): otbr_manager = get_otbr_addon_manager(self.hass) if not self.addon_start_task: - # Before we start the addon, confirm that the correct firmware is running - # and populate `self._probed_firmware_info` with the correct information - if not await self._probe_firmware_info( - probe_methods=(ApplicationType.SPINEL,) - ): - return self.async_abort( - reason="unsupported_firmware", - description_placeholders=self._get_translation_placeholders(), - ) - - addon_info = await self._async_get_addon_info(otbr_manager) - - assert self._device is not None - new_addon_config = { - **addon_info.options, - "device": self._device, - "baudrate": 460800, - "flow_control": True, - "autoflash_firmware": False, - } - - _LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config) - - try: - await otbr_manager.async_set_addon_options(new_addon_config) - except AddonError as err: - _LOGGER.error(err) - raise AbortFlow( - "addon_set_config_failed", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": otbr_manager.addon_name, - }, - ) from err - self.addon_start_task = self.hass.async_create_task( - otbr_manager.async_start_addon_waiting() + self._configure_and_start_otbr_addon() ) if not self.addon_start_task.done(): @@ -608,7 +601,9 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): except (AddonError, AbortFlow) as err: _LOGGER.error(err) self._failed_addon_name = otbr_manager.addon_name - self._failed_addon_reason = "addon_start_failed" + self._failed_addon_reason = ( + err.reason if isinstance(err, AbortFlow) else "addon_start_failed" + ) return self.async_show_progress_done(next_step_id="addon_operation_failed") finally: self.addon_start_task = None diff --git a/tests/components/homeassistant_hardware/test_config_flow_failures.py b/tests/components/homeassistant_hardware/test_config_flow_failures.py index 9ad3977394a..e02faf97ced 100644 --- a/tests/components/homeassistant_hardware/test_config_flow_failures.py +++ b/tests/components/homeassistant_hardware/test_config_flow_failures.py @@ -277,47 +277,6 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None: assert result["reason"] == "addon_info_failed" -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_thread_addon_already_configured(hass: HomeAssistant) -> None: - """Test failure case when the Thread addon is already running.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_firmware_info( - hass, - probe_app_type=ApplicationType.EZSP, - otbr_addon_info=AddonInfo( - available=True, - hostname=None, - options={ - "device": TEST_DEVICE + "2", # A different device - }, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ), - ) as (mock_otbr_manager, _): - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, - ) - - # Cannot install addon - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "otbr_addon_already_running" - - @pytest.mark.parametrize( "ignore_translations_for_mock_domains", ["test_firmware_domain"],