Allow setup of Zigbee/Thread for ZBT-1 and Yellow without internet access (#147549)

Co-authored-by: Norbert Rittel <norbert@rittel.de>

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -7,7 +7,10 @@ import asyncio
 import logging
 from typing import Any
 
-from ha_silabs_firmware_client import FirmwareUpdateClient
+from aiohttp import ClientError
+from ha_silabs_firmware_client import FirmwareUpdateClient, ManifestMissing
+from universal_silabs_flasher.common import Version
+from universal_silabs_flasher.firmware import NabuCasaMetadata
 
 from homeassistant.components.hassio import (
     AddonError,
@@ -149,15 +152,78 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
         assert self._device is not None
 
         if not self.firmware_install_task:
-            session = async_get_clientsession(self.hass)
-            client = FirmwareUpdateClient(fw_update_url, session)
-            manifest = await client.async_update_data()
-
-            fw_meta = next(
-                fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
+            # We 100% need to install new firmware only if the wrong firmware is
+            # currently installed
+            firmware_install_required = self._probed_firmware_info is None or (
+                self._probed_firmware_info.firmware_type
+                != expected_installed_firmware_type
             )
 
-            fw_data = await client.async_fetch_firmware(fw_meta)
+            session = async_get_clientsession(self.hass)
+            client = FirmwareUpdateClient(fw_update_url, session)
+
+            try:
+                manifest = await client.async_update_data()
+                fw_manifest = next(
+                    fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
+                )
+            except (StopIteration, TimeoutError, ClientError, ManifestMissing) as err:
+                _LOGGER.warning(
+                    "Failed to fetch firmware update manifest", exc_info=True
+                )
+
+                # Not having internet access should not prevent setup
+                if not firmware_install_required:
+                    _LOGGER.debug(
+                        "Skipping firmware upgrade due to index download failure"
+                    )
+                    return self.async_show_progress_done(next_step_id=next_step_id)
+
+                raise AbortFlow(
+                    "fw_download_failed",
+                    description_placeholders={
+                        **self._get_translation_placeholders(),
+                        "firmware_name": firmware_name,
+                    },
+                ) from err
+
+            if not firmware_install_required:
+                assert self._probed_firmware_info is not None
+
+                # Make sure we do not downgrade the firmware
+                fw_metadata = NabuCasaMetadata.from_json(fw_manifest.metadata)
+                fw_version = fw_metadata.get_public_version()
+                probed_fw_version = Version(self._probed_firmware_info.firmware_version)
+
+                if probed_fw_version >= fw_version:
+                    _LOGGER.debug(
+                        "Not downgrading firmware, installed %s is newer than available %s",
+                        probed_fw_version,
+                        fw_version,
+                    )
+                    return self.async_show_progress_done(next_step_id=next_step_id)
+
+            try:
+                fw_data = await client.async_fetch_firmware(fw_manifest)
+            except (TimeoutError, ClientError, ValueError) as err:
+                _LOGGER.warning("Failed to fetch firmware update", exc_info=True)
+
+                # If we cannot download new firmware, we shouldn't block setup
+                if not firmware_install_required:
+                    _LOGGER.debug(
+                        "Skipping firmware upgrade due to image download failure"
+                    )
+                    return self.async_show_progress_done(next_step_id=next_step_id)
+
+                # Otherwise, fail
+                raise AbortFlow(
+                    "fw_download_failed",
+                    description_placeholders={
+                        **self._get_translation_placeholders(),
+                        "firmware_name": firmware_name,
+                    },
+                ) from err
+
             self.firmware_install_task = self.hass.async_create_task(
                 async_flash_silabs_firmware(
                     hass=self.hass,
@@ -215,6 +281,14 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
             },
         )
 
+    async def async_step_pre_confirm_zigbee(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Pre-confirm Zigbee setup."""
+
+        # This step is necessary to prevent `user_input` from being passed through
+        return await self.async_step_confirm_zigbee()
+
     async def async_step_confirm_zigbee(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -409,7 +483,15 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
         finally:
             self.addon_start_task = None
 
-        return self.async_show_progress_done(next_step_id="confirm_otbr")
+        return self.async_show_progress_done(next_step_id="pre_confirm_otbr")
+
+    async def async_step_pre_confirm_otbr(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Pre-confirm OTBR setup."""
+
+        # This step is necessary to prevent `user_input` from being passed through
+        return await self.async_step_confirm_otbr()
 
     async def async_step_confirm_otbr(
         self, user_input: dict[str, Any] | None = None

homeassistant/components/homeassistant_hardware/strings.json

@@ -36,7 +36,8 @@
         "otbr_addon_already_running": "The OpenThread Border Router add-on is already running, it cannot be installed again.",
         "zha_still_using_stick": "This {model} is in use by the Zigbee Home Automation integration. Please migrate your Zigbee network to another adapter or delete the integration and try again.",
         "otbr_still_using_stick": "This {model} is in use by the OpenThread Border Router add-on. If you use the Thread network, make sure you have alternative border routers. Uninstall the add-on and try again.",
-        "unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or add-on is currently trying to communicate with the device. If you are running Home Assistant OS in a virtual machine or in Docker, please make sure that permissions are set correctly for the device."
+        "unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or add-on is currently trying to communicate with the device. If you are running Home Assistant OS in a virtual machine or in Docker, please make sure that permissions are set correctly for the device.",
+        "fw_download_failed": "{firmware_name} firmware for your {model} failed to download. Make sure Home Assistant has internet access and try again."
       },
       "progress": {
         "install_firmware": "Please wait while {firmware_name} firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes."

homeassistant/components/homeassistant_sky_connect/config_flow.py

@@ -93,7 +93,7 @@ class SkyConnectFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol):
             firmware_name="Zigbee",
             expected_installed_firmware_type=ApplicationType.EZSP,
             step_id="install_zigbee_firmware",
-            next_step_id="confirm_zigbee",
+            next_step_id="pre_confirm_zigbee",
         )
 
     async def async_step_install_thread_firmware(

homeassistant/components/homeassistant_sky_connect/strings.json

@@ -92,7 +92,8 @@
       "otbr_addon_already_running": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_addon_already_running%]",
       "zha_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::zha_still_using_stick%]",
       "otbr_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_still_using_stick%]",
-      "unsupported_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::unsupported_firmware%]"
+      "unsupported_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::unsupported_firmware%]",
+      "fw_download_failed": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::fw_download_failed%]"
     },
     "progress": {
       "install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",
@@ -145,7 +146,8 @@
       "otbr_addon_already_running": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_addon_already_running%]",
       "zha_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::zha_still_using_stick%]",
       "otbr_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_still_using_stick%]",
-      "unsupported_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::unsupported_firmware%]"
+      "unsupported_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::unsupported_firmware%]",
+      "fw_download_failed": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::fw_download_failed%]"
     },
     "progress": {
       "install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",

homeassistant/components/homeassistant_yellow/strings.json

@@ -117,7 +117,8 @@
       "otbr_addon_already_running": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_addon_already_running%]",
       "zha_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::zha_still_using_stick%]",
       "otbr_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_still_using_stick%]",
-      "unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or addon is currently trying to communicate with the device."
+      "unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or add-on is currently trying to communicate with the device.",
+      "fw_download_failed": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::fw_download_failed%]"
     },
     "progress": {
       "install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",

tests/components/homeassistant_hardware/test_config_flow.py

@@ -6,6 +6,7 @@ import contextlib
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, Mock, call, patch
 
+from aiohttp import ClientError
 from ha_silabs_firmware_client import (
     FirmwareManifest,
     FirmwareMetadata,
@@ -80,7 +81,7 @@ class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN):
             firmware_name="Zigbee",
             expected_installed_firmware_type=ApplicationType.EZSP,
             step_id="install_zigbee_firmware",
-            next_step_id="confirm_zigbee",
+            next_step_id="pre_confirm_zigbee",
         )
 
     async def async_step_install_thread_firmware(
@@ -137,7 +138,7 @@ class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow):
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Install Zigbee firmware."""
-        return await self.async_step_confirm_zigbee()
+        return await self.async_step_pre_confirm_zigbee()
 
     async def async_step_install_thread_firmware(
         self, user_input: dict[str, Any] | None = None
@@ -208,6 +209,7 @@ def mock_firmware_info(
     *,
     is_hassio: bool = True,
     probe_app_type: ApplicationType | None = ApplicationType.EZSP,
+    probe_fw_version: str | None = "2.4.4.0",
     otbr_addon_info: AddonInfo = AddonInfo(
         available=True,
         hostname=None,
@@ -217,6 +219,7 @@ def mock_firmware_info(
         version=None,
     ),
     flash_app_type: ApplicationType = ApplicationType.EZSP,
+    flash_fw_version: str | None = "7.4.4.0",
 ) -> Iterator[tuple[Mock, Mock]]:
     """Mock the main addon states for the config flow."""
     mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
@@ -243,7 +246,14 @@ def mock_firmware_info(
                 checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
                 size=123,
                 release_notes="Some release notes",
-                metadata={},
+                metadata={
+                    "baudrate": 460800,
+                    "fw_type": "openthread_rcp",
+                    "fw_variant": None,
+                    "metadata_version": 2,
+                    "ot_rcp_version": "SL-OPENTHREAD/2.4.4.0_GitHub-7074a43e4",
+                    "sdk_version": "4.4.4",
+                },
                 url=TEST_RELEASES_URL / "fake_openthread_rcp_7.4.4.0_variant.gbl",
             ),
             FirmwareMetadata(
@@ -251,7 +261,14 @@ def mock_firmware_info(
                 checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
                 size=123,
                 release_notes="Some release notes",
-                metadata={},
+                metadata={
+                    "baudrate": 115200,
+                    "ezsp_version": "7.4.4.0",
+                    "fw_type": "zigbee_ncp",
+                    "fw_variant": None,
+                    "metadata_version": 2,
+                    "sdk_version": "4.4.4",
+                },
                 url=TEST_RELEASES_URL / "fake_zigbee_ncp_7.4.4.0_variant.gbl",
             ),
         ],
@@ -263,7 +280,7 @@ def mock_firmware_info(
         probed_firmware_info = FirmwareInfo(
             device="/dev/ttyUSB0",  # Not used
             firmware_type=probe_app_type,
-            firmware_version=None,
+            firmware_version=probe_fw_version,
             owners=[],
             source="probe",
         )
@@ -274,7 +291,7 @@ def mock_firmware_info(
         flashed_firmware_info = FirmwareInfo(
             device=TEST_DEVICE,
             firmware_type=flash_app_type,
-            firmware_version="7.4.4.0",
+            firmware_version=flash_fw_version,
             owners=[create_mock_owner()],
             source="probe",
         )
@@ -333,7 +350,7 @@ def mock_firmware_info(
             side_effect=mock_flash_firmware,
         ),
     ):
-        yield mock_otbr_manager
+        yield mock_otbr_manager, mock_update_client
 
 
 async def consume_progress_flow(
@@ -411,6 +428,91 @@ async def test_config_flow_zigbee(hass: HomeAssistant) -> None:
     assert zha_flow["step_id"] == "confirm"
 
 
+async def test_config_flow_firmware_index_download_fails_but_not_required(
+    hass: HomeAssistant,
+) -> None:
+    """Test flow continues if index download fails but install is not required."""
+    init_result = await hass.config_entries.flow.async_init(
+        TEST_DOMAIN, context={"source": "hardware"}
+    )
+
+    with mock_firmware_info(
+        hass,
+        # The correct firmware is already installed
+        probe_app_type=ApplicationType.EZSP,
+        # An older version is probed, so an upgrade is attempted
+        probe_fw_version="7.4.3.0",
+    ) as (_, mock_update_client):
+        # Mock the firmware download to fail
+        mock_update_client.async_update_data.side_effect = ClientError()
+
+        pick_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"],
+            user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
+        )
+
+        assert pick_result["type"] is FlowResultType.FORM
+        assert pick_result["step_id"] == "confirm_zigbee"
+
+
+async def test_config_flow_firmware_download_fails_but_not_required(
+    hass: HomeAssistant,
+) -> None:
+    """Test flow continues if firmware download fails but install is not required."""
+    init_result = await hass.config_entries.flow.async_init(
+        TEST_DOMAIN, context={"source": "hardware"}
+    )
+
+    with (
+        mock_firmware_info(
+            hass,
+            # The correct firmware is already installed so installation isn't required
+            probe_app_type=ApplicationType.EZSP,
+            # An older version is probed, so an upgrade is attempted
+            probe_fw_version="7.4.3.0",
+        ) as (_, mock_update_client),
+    ):
+        mock_update_client.async_fetch_firmware.side_effect = ClientError()
+
+        pick_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"],
+            user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
+        )
+
+        assert pick_result["type"] is FlowResultType.FORM
+        assert pick_result["step_id"] == "confirm_zigbee"
+
+
+async def test_config_flow_doesnt_downgrade(
+    hass: HomeAssistant,
+) -> None:
+    """Test flow exits early, without downgrading firmware."""
+    init_result = await hass.config_entries.flow.async_init(
+        TEST_DOMAIN, context={"source": "hardware"}
+    )
+
+    with (
+        mock_firmware_info(
+            hass,
+            probe_app_type=ApplicationType.EZSP,
+            # An newer version is probed than what we offer
+            probe_fw_version="7.5.0.0",
+        ),
+        patch(
+            "homeassistant.components.homeassistant_hardware.firmware_config_flow.async_flash_silabs_firmware"
+        ) as mock_async_flash_silabs_firmware,
+    ):
+        pick_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"],
+            user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
+        )
+
+        assert pick_result["type"] is FlowResultType.FORM
+        assert pick_result["step_id"] == "confirm_zigbee"
+
+        assert len(mock_async_flash_silabs_firmware.mock_calls) == 0
+
+
 async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) -> None:
     """Test the config flow, skip installing the addon if necessary."""
     result = await hass.config_entries.flow.async_init(
@@ -480,7 +582,7 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None:
         hass,
         probe_app_type=ApplicationType.EZSP,
         flash_app_type=ApplicationType.SPINEL,
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         # Pick the menu option
         pick_result = await hass.config_entries.flow.async_configure(
             init_result["flow_id"],
@@ -564,7 +666,7 @@ async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) -
             update_available=False,
             version=None,
         ),
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         # Pick the menu option
         pick_result = await hass.config_entries.flow.async_configure(
             init_result["flow_id"],
@@ -631,7 +733,7 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
         hass,
         probe_app_type=ApplicationType.EZSP,
         flash_app_type=ApplicationType.SPINEL,
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         # First step is confirmation
         result = await hass.config_entries.options.async_init(config_entry.entry_id)
         assert result["type"] is FlowResultType.MENU

tests/components/homeassistant_hardware/test_config_flow_failures.py

@@ -2,6 +2,7 @@
 
 from unittest.mock import AsyncMock, patch
 
+from aiohttp import ClientError
 import pytest
 
 from homeassistant.components.hassio import AddonError, AddonInfo, AddonState
@@ -109,7 +110,7 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None:
     with mock_firmware_info(
         hass,
         probe_app_type=ApplicationType.EZSP,
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         mock_otbr_manager.async_get_addon_info.side_effect = AddonError()
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"], user_input={}
@@ -147,7 +148,7 @@ async def test_config_flow_thread_addon_already_configured(hass: HomeAssistant)
             update_available=False,
             version="1.0.0",
         ),
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         mock_otbr_manager.async_install_addon_waiting = AsyncMock(
             side_effect=AddonError()
         )
@@ -178,7 +179,7 @@ async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> No
     with mock_firmware_info(
         hass,
         probe_app_type=ApplicationType.EZSP,
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         mock_otbr_manager.async_install_addon_waiting = AsyncMock(
             side_effect=AddonError()
         )
@@ -209,7 +210,7 @@ async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) ->
     with mock_firmware_info(
         hass,
         probe_app_type=ApplicationType.EZSP,
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
 
         async def install_addon() -> None:
             mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
@@ -270,7 +271,7 @@ async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None
             update_available=False,
             version="1.0.0",
         ),
-    ) as mock_otbr_manager:
+    ) as (mock_otbr_manager, _):
         mock_otbr_manager.async_start_addon_waiting = AsyncMock(
             side_effect=AddonError()
         )
@@ -341,6 +342,64 @@ async def test_config_flow_thread_confirmation_fails(hass: HomeAssistant) -> Non
         assert pick_thread_progress_result["reason"] == "unsupported_firmware"
 
 
+@pytest.mark.parametrize(
+    "ignore_translations_for_mock_domains", ["test_firmware_domain"]
+)
+async def test_config_flow_firmware_index_download_fails_and_required(
+    hass: HomeAssistant,
+) -> None:
+    """Test flow aborts if OTA index download fails and install is required."""
+    init_result = await hass.config_entries.flow.async_init(
+        TEST_DOMAIN, context={"source": "hardware"}
+    )
+
+    with (
+        mock_firmware_info(
+            hass,
+            # The wrong firmware is installed, so a new install is required
+            probe_app_type=ApplicationType.SPINEL,
+        ) as (_, mock_update_client),
+    ):
+        mock_update_client.async_update_data.side_effect = ClientError()
+
+        pick_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"],
+            user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
+        )
+
+        assert pick_result["type"] is FlowResultType.ABORT
+        assert pick_result["reason"] == "fw_download_failed"
+
+
+@pytest.mark.parametrize(
+    "ignore_translations_for_mock_domains", ["test_firmware_domain"]
+)
+async def test_config_flow_firmware_download_fails_and_required(
+    hass: HomeAssistant,
+) -> None:
+    """Test flow aborts if firmware download fails and install is required."""
+    init_result = await hass.config_entries.flow.async_init(
+        TEST_DOMAIN, context={"source": "hardware"}
+    )
+
+    with (
+        mock_firmware_info(
+            hass,
+            # The wrong firmware is installed, so a new install is required
+            probe_app_type=ApplicationType.SPINEL,
+        ) as (_, mock_update_client),
+    ):
+        mock_update_client.async_fetch_firmware.side_effect = ClientError()
+
+        pick_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"],
+            user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
+        )
+
+        assert pick_result["type"] is FlowResultType.ABORT
+        assert pick_result["reason"] == "fw_download_failed"
+
+
 @pytest.mark.parametrize(
     "ignore_translations_for_mock_domains",
     ["test_firmware_domain"],

tests/components/homeassistant_sky_connect/test_config_flow.py

@@ -75,7 +75,7 @@ async def test_config_flow(
         next_step_id: str,
     ) -> ConfigFlowResult:
         if next_step_id == "start_otbr_addon":
-            next_step_id = "confirm_otbr"
+            next_step_id = "pre_confirm_otbr"
 
         return await getattr(self, f"async_step_{next_step_id}")(user_input={})
 
@@ -100,14 +100,22 @@ async def test_config_flow(
             ),
         ),
     ):
-        result = await hass.config_entries.flow.async_configure(
+        confirm_result = await hass.config_entries.flow.async_configure(
             result["flow_id"],
             user_input={"next_step_id": step},
         )
 
-    assert result["type"] is FlowResultType.CREATE_ENTRY
+        assert confirm_result["type"] is FlowResultType.FORM
+        assert confirm_result["step_id"] == (
+            "confirm_zigbee" if step == STEP_PICK_FIRMWARE_ZIGBEE else "confirm_otbr"
+        )
 
-    config_entry = result["result"]
+        create_result = await hass.config_entries.flow.async_configure(
+            confirm_result["flow_id"], user_input={}
+        )
+
+    assert create_result["type"] is FlowResultType.CREATE_ENTRY
+    config_entry = create_result["result"]
     assert config_entry.data == {
         "firmware": fw_type.value,
         "firmware_version": fw_version,
@@ -171,7 +179,7 @@ async def test_options_flow(
     assert result["description_placeholders"]["model"] == model
 
     async def mock_async_step_pick_firmware_zigbee(self, data):
-        return await self.async_step_confirm_zigbee(user_input={})
+        return await self.async_step_pre_confirm_zigbee()
 
     with (
         patch(
@@ -190,13 +198,20 @@ async def test_options_flow(
             ),
         ),
     ):
-        result = await hass.config_entries.options.async_configure(
+        confirm_result = await hass.config_entries.options.async_configure(
             result["flow_id"],
             user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
         )
 
-    assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["result"] is True
+        assert confirm_result["type"] is FlowResultType.FORM
+        assert confirm_result["step_id"] == "confirm_zigbee"
+
+        create_result = await hass.config_entries.options.async_configure(
+            confirm_result["flow_id"], user_input={}
+        )
+
+    assert create_result["type"] is FlowResultType.CREATE_ENTRY
+    assert create_result["result"] is True
 
     assert config_entry.data == {
         "firmware": "ezsp",

tests/components/homeassistant_yellow/test_config_flow.py

@@ -348,7 +348,7 @@ async def test_firmware_options_flow(
     assert result["description_placeholders"]["model"] == "Home Assistant Yellow"
 
     async def mock_async_step_pick_firmware_zigbee(self, data):
-        return await self.async_step_confirm_zigbee(user_input={})
+        return await self.async_step_pre_confirm_zigbee()
 
     async def mock_install_firmware_step(
         self,
@@ -360,11 +360,16 @@ async def test_firmware_options_flow(
         next_step_id: str,
     ) -> ConfigFlowResult:
         if next_step_id == "start_otbr_addon":
-            next_step_id = "confirm_otbr"
+            next_step_id = "pre_confirm_otbr"
 
         return await getattr(self, f"async_step_{next_step_id}")(user_input={})
 
     with (
+        patch(
+            "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow.async_step_pick_firmware_zigbee",
+            autospec=True,
+            side_effect=mock_async_step_pick_firmware_zigbee,
+        ),
         patch(
             "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup",
             return_value=None,
@@ -385,13 +390,22 @@ async def test_firmware_options_flow(
             ),
         ),
     ):
-        result = await hass.config_entries.options.async_configure(
+        confirm_result = await hass.config_entries.options.async_configure(
             result["flow_id"],
             user_input={"next_step_id": step},
         )
 
-    assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["result"] is True
+        assert confirm_result["type"] is FlowResultType.FORM
+        assert confirm_result["step_id"] == (
+            "confirm_zigbee" if step == STEP_PICK_FIRMWARE_ZIGBEE else "confirm_otbr"
+        )
+
+        create_result = await hass.config_entries.options.async_configure(
+            confirm_result["flow_id"], user_input={}
+        )
+
+    assert create_result["type"] is FlowResultType.CREATE_ENTRY
+    assert create_result["result"] is True
 
     assert config_entry.data == {
         "firmware": fw_type.value,