Remove @progress_step decorator from ZHA and Hardware integration (home-assistant#155867)

Co-authored-by: TheJulianJES <[email protected]>

Author: puddly

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -28,7 +28,7 @@
     OptionsFlow,
 )
 from homeassistant.core import callback
-from homeassistant.data_entry_flow import AbortFlow, progress_step
+from homeassistant.data_entry_flow import AbortFlow
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 from homeassistant.helpers.hassio import is_hassio
@@ -97,6 +97,12 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.addon_uninstall_task: asyncio.Task | None = None
         self.firmware_install_task: asyncio.Task[None] | None = None
         self.installing_firmware_name: str | None = None
+        self._install_otbr_addon_task: asyncio.Task[None] | None = None
+        self._start_otbr_addon_task: asyncio.Task[None] | None = None
+
+        # Progress flow steps cannot abort so we need to store the abort reason and then
+        # re-raise it in a dedicated step
+        self._progress_error: AbortFlow | None = None
 
     def _get_translation_placeholders(self) -> dict[str, str]:
         """Shared translation placeholders."""
@@ -106,6 +112,11 @@ def _get_translation_placeholders(self) -> dict[str, str]:
                 if self._probed_firmware_info is not None
                 else "unknown"
             ),
+            "firmware_name": (
+                self.installing_firmware_name
+                if self.installing_firmware_name is not None
+                else "unknown"
+            ),
             "model": self._hardware_name,
         }
 
@@ -182,22 +193,22 @@ async def _install_firmware_step(
             return self.async_show_progress(
                 step_id=step_id,
                 progress_action="install_firmware",
-                description_placeholders={
-                    **self._get_translation_placeholders(),
-                    "firmware_name": firmware_name,
-                },
+                description_placeholders=self._get_translation_placeholders(),
                 progress_task=self.firmware_install_task,
             )
 
         try:
             await self.firmware_install_task
         except AbortFlow as err:
-            return self.async_show_progress_done(
-                next_step_id=err.reason,
-            )
+            self._progress_error = err
+            return self.async_show_progress_done(next_step_id="progress_failed")
         except HomeAssistantError:
             _LOGGER.exception("Failed to flash firmware")
-            return self.async_show_progress_done(next_step_id="firmware_install_failed")
+            self._progress_error = AbortFlow(
+                reason="fw_install_failed",
+                description_placeholders=self._get_translation_placeholders(),
+            )
+            return self.async_show_progress_done(next_step_id="progress_failed")
         finally:
             self.firmware_install_task = None
 
@@ -241,7 +252,10 @@ async def _install_firmware(
                 _LOGGER.debug("Skipping firmware upgrade due to index download failure")
                 return
 
-            raise AbortFlow(reason="firmware_download_failed") from err
+            raise AbortFlow(
+                reason="fw_download_failed",
+                description_placeholders=self._get_translation_placeholders(),
+            ) from err
 
         if not firmware_install_required:
             assert self._probed_firmware_info is not None
@@ -270,7 +284,10 @@ async def _install_firmware(
                 return
 
             # Otherwise, fail
-            raise AbortFlow(reason="firmware_download_failed") from err
+            raise AbortFlow(
+                reason="fw_download_failed",
+                description_placeholders=self._get_translation_placeholders(),
+            ) from err
 
         self._probed_firmware_info = await async_flash_silabs_firmware(
             hass=self.hass,
@@ -313,41 +330,6 @@ async def _configure_and_start_otbr_addon(self) -> None:
 
         await otbr_manager.async_start_addon_waiting()
 
-    async def async_step_firmware_download_failed(
-        self, user_input: dict[str, Any] | None = None
-    ) -> ConfigFlowResult:
-        """Abort when firmware download failed."""
-        assert self.installing_firmware_name is not None
-        return self.async_abort(
-            reason="fw_download_failed",
-            description_placeholders={
-                **self._get_translation_placeholders(),
-                "firmware_name": self.installing_firmware_name,
-            },
-        )
-
-    async def async_step_firmware_install_failed(
-        self, user_input: dict[str, Any] | None = None
-    ) -> ConfigFlowResult:
-        """Abort when firmware install failed."""
-        assert self.installing_firmware_name is not None
-        return self.async_abort(
-            reason="fw_install_failed",
-            description_placeholders={
-                **self._get_translation_placeholders(),
-                "firmware_name": self.installing_firmware_name,
-            },
-        )
-
-    async def async_step_unsupported_firmware(
-        self, user_input: dict[str, Any] | None = None
-    ) -> ConfigFlowResult:
-        """Abort when unsupported firmware is detected."""
-        return self.async_abort(
-            reason="unsupported_firmware",
-            description_placeholders=self._get_translation_placeholders(),
-        )
-
     async def async_step_zigbee_installation_type(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -511,16 +493,15 @@ async def async_step_install_thread_firmware(
         """Install Thread firmware."""
         raise NotImplementedError
 
-    @progress_step(
-        description_placeholders=lambda self: {
-            **self._get_translation_placeholders(),
-            "addon_name": get_otbr_addon_manager(self.hass).addon_name,
-        }
-    )
-    async def async_step_install_otbr_addon(
+    async def async_step_progress_failed(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
-        """Show progress dialog for installing the OTBR addon."""
+        """Abort when progress step failed."""
+        assert self._progress_error is not None
+        raise self._progress_error
+
+    async def _async_install_otbr_addon(self) -> None:
+        """Do the work of installing the OTBR addon."""
         addon_manager = get_otbr_addon_manager(self.hass)
         addon_info = await self._async_get_addon_info(addon_manager)
 
@@ -538,18 +519,39 @@ async def async_step_install_otbr_addon(
                 },
             ) from err
 
-        return await self.async_step_finish_thread_installation()
-
-    @progress_step(
-        description_placeholders=lambda self: {
-            **self._get_translation_placeholders(),
-            "addon_name": get_otbr_addon_manager(self.hass).addon_name,
-        }
-    )
-    async def async_step_start_otbr_addon(
+    async def async_step_install_otbr_addon(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
-        """Configure OTBR to point to the SkyConnect and run the addon."""
+        """Show progress dialog for installing the OTBR addon."""
+        if self._install_otbr_addon_task is None:
+            self._install_otbr_addon_task = self.hass.async_create_task(
+                self._async_install_otbr_addon(),
+                "Install OTBR addon",
+            )
+
+        if not self._install_otbr_addon_task.done():
+            return self.async_show_progress(
+                step_id="install_otbr_addon",
+                progress_action="install_otbr_addon",
+                description_placeholders={
+                    **self._get_translation_placeholders(),
+                    "addon_name": get_otbr_addon_manager(self.hass).addon_name,
+                },
+                progress_task=self._install_otbr_addon_task,
+            )
+
+        try:
+            await self._install_otbr_addon_task
+        except AbortFlow as err:
+            self._progress_error = err
+            return self.async_show_progress_done(next_step_id="progress_failed")
+        finally:
+            self._install_otbr_addon_task = None
+
+        return self.async_show_progress_done(next_step_id="finish_thread_installation")
+
+    async def _async_start_otbr_addon(self) -> None:
+        """Do the work of starting the OTBR addon."""
         try:
             await self._configure_and_start_otbr_addon()
         except AddonError as err:
@@ -562,7 +564,36 @@ async def async_step_start_otbr_addon(
                 },
             ) from err
 
-        return await self.async_step_pre_confirm_otbr()
+    async def async_step_start_otbr_addon(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Configure OTBR to point to the SkyConnect and run the addon."""
+        if self._start_otbr_addon_task is None:
+            self._start_otbr_addon_task = self.hass.async_create_task(
+                self._async_start_otbr_addon(),
+                "Start OTBR addon",
+            )
+
+        if not self._start_otbr_addon_task.done():
+            return self.async_show_progress(
+                step_id="start_otbr_addon",
+                progress_action="start_otbr_addon",
+                description_placeholders={
+                    **self._get_translation_placeholders(),
+                    "addon_name": get_otbr_addon_manager(self.hass).addon_name,
+                },
+                progress_task=self._start_otbr_addon_task,
+            )
+
+        try:
+            await self._start_otbr_addon_task
+        except AbortFlow as err:
+            self._progress_error = err
+            return self.async_show_progress_done(next_step_id="progress_failed")
+        finally:
+            self._start_otbr_addon_task = None
+
+        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