Fix hardware unit tests

2025-08-01 03:35:09 +02:00 · 2025-05-14 13:50:19 -04:00
parent 208d0e9ee3
commit 8d96736003
2 changed files with 111 additions and 134 deletions
--- a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py
+++ b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py
@@ -24,7 +24,6 @@ from homeassistant.core import callback
 from homeassistant.data_entry_flow import AbortFlow
 from homeassistant.helpers.hassio import is_hassio

-from . import silabs_multiprotocol_addon
 from .const import OTBR_DOMAIN, ZHA_DOMAIN
 from .util import (
    ApplicationType,
@@ -76,22 +75,6 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):

        return placeholders

-    async def _async_set_addon_config(
-        self, config: dict, addon_manager: AddonManager
-    ) -> None:
-        """Set add-on config."""
-        try:
-            await addon_manager.async_set_addon_options(config)
-        except AddonError as err:
-            _LOGGER.error(err)
-            raise AbortFlow(
-                "addon_set_config_failed",
-                description_placeholders={
-                    **self._get_translation_placeholders(),
-                    "addon_name": addon_manager.addon_name,
-                },
-            ) from err
-
    async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo:
        """Return add-on info."""
        try:
@@ -174,46 +157,6 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
        """Install Zigbee firmware."""
        raise NotImplementedError

-    async def _install_addon(
-        self,
-        addon_manager: silabs_multiprotocol_addon.WaitingAddonManager,
-        step_id: str,
-        next_step_id: str,
-    ) -> ConfigFlowResult:
-        """Show progress dialog for installing an addon."""
-        addon_info = await self._async_get_addon_info(addon_manager)
-
-        _LOGGER.debug("Flasher addon state: %s", addon_info)
-
-        if not self.addon_install_task:
-            self.addon_install_task = self.hass.async_create_task(
-                addon_manager.async_install_addon_waiting(),
-                "Addon install",
-            )
-
-        if not self.addon_install_task.done():
-            return self.async_show_progress(
-                step_id=step_id,
-                progress_action="install_addon",
-                description_placeholders={
-                    **self._get_translation_placeholders(),
-                    "addon_name": addon_manager.addon_name,
-                },
-                progress_task=self.addon_install_task,
-            )
-
-        try:
-            await self.addon_install_task
-        except AddonError as err:
-            _LOGGER.error(err)
-            self._failed_addon_name = addon_manager.addon_name
-            self._failed_addon_reason = "addon_install_failed"
-            return self.async_show_progress_done(next_step_id="addon_operation_failed")
-        finally:
-            self.addon_install_task = None
-
-        return self.async_show_progress_done(next_step_id=next_step_id)
-
    async def async_step_addon_operation_failed(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
@@ -311,11 +254,39 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Show progress dialog for installing the OTBR addon."""
-        return await self._install_addon(
-            addon_manager=get_otbr_addon_manager(self.hass),
-            step_id="install_otbr_addon",
-            next_step_id="pick_firmware_thread",
-        )
+        addon_manager = get_otbr_addon_manager(self.hass)
+        addon_info = await self._async_get_addon_info(addon_manager)
+
+        _LOGGER.debug("OTBR addon info: %s", addon_info)
+
+        if not self.addon_install_task:
+            self.addon_install_task = self.hass.async_create_task(
+                addon_manager.async_install_addon_waiting(),
+                "OTBR addon install",
+            )
+
+        if not self.addon_install_task.done():
+            return self.async_show_progress(
+                step_id="install_otbr_addon",
+                progress_action="install_addon",
+                description_placeholders={
+                    **self._get_translation_placeholders(),
+                    "addon_name": addon_manager.addon_name,
+                },
+                progress_task=self.addon_install_task,
+            )
+
+        try:
+            await self.addon_install_task
+        except AddonError as err:
+            _LOGGER.error(err)
+            self._failed_addon_name = addon_manager.addon_name
+            self._failed_addon_reason = "addon_install_failed"
+            return self.async_show_progress_done(next_step_id="addon_operation_failed")
+        finally:
+            self.addon_install_task = None
+
+        return self.async_show_progress_done(next_step_id="pick_firmware_thread")

    async def async_step_start_otbr_addon(
        self, user_input: dict[str, Any] | None = None
@@ -334,7 +305,18 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
        }

        _LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config)
-        await self._async_set_addon_config(new_addon_config, otbr_manager)
+
+        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

        if not self.addon_start_task:
            self.addon_start_task = self.hass.async_create_task(
--- a/tests/components/homeassistant_hardware/test_config_flow_failures.py
+++ b/tests/components/homeassistant_hardware/test_config_flow_failures.py
@@ -163,7 +163,7 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None:
    "ignore_translations_for_mock_domains",
    ["test_firmware_domain"],
 )
-async def test_config_flow_thread_addon_already_running(hass: HomeAssistant) -> None:
+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"}
@@ -175,7 +175,9 @@ async def test_config_flow_thread_addon_already_running(hass: HomeAssistant) ->
        otbr_addon_info=AddonInfo(
            available=True,
            hostname=None,
-            options={},
+            options={
+                "device": TEST_DEVICE + "2",  # A different device
+            },
            state=AddonState.RUNNING,
            update_available=False,
            version="1.0.0",
@@ -235,7 +237,7 @@ async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> No
 )
 async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> None:
    """Test failure case when flasher addon cannot be configured."""
-    result = await hass.config_entries.flow.async_init(
+    init_result = await hass.config_entries.flow.async_init(
        TEST_DOMAIN, context={"source": "hardware"}
    )

@@ -243,21 +245,33 @@ async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) ->
        hass,
        app_type=ApplicationType.EZSP,
    ) as mock_otbr_manager:
+
+        async def install_addon() -> None:
+            mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
+                available=True,
+                hostname=None,
+                options={"device": TEST_DEVICE},
+                state=AddonState.NOT_RUNNING,
+                update_available=False,
+                version="1.0.0",
+            )
+
+        mock_otbr_manager.async_install_addon_waiting = AsyncMock(
+            side_effect=install_addon
+        )
        mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError())

-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"], user_input={}
+        confirm_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"], user_input={}
        )
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"],
+
+        pick_thread_result = await hass.config_entries.flow.async_configure(
+            confirm_result["flow_id"],
            user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
        )
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        await hass.async_block_till_done(wait_background_tasks=True)

-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        assert result["type"] == FlowResultType.ABORT
-        assert result["reason"] == "addon_set_config_failed"
+        assert pick_thread_result["type"] == FlowResultType.ABORT
+        assert pick_thread_result["reason"] == "addon_set_config_failed"


@pytest.mark.parametrize(
@@ -266,63 +280,35 @@ async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) ->
 )
 async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None:
    """Test failure case when flasher addon fails to run."""
-    result = await hass.config_entries.flow.async_init(
+    init_result = await hass.config_entries.flow.async_init(
        TEST_DOMAIN, context={"source": "hardware"}
    )

    with mock_addon_info(
        hass,
        app_type=ApplicationType.EZSP,
+        otbr_addon_info=AddonInfo(
+            available=True,
+            hostname=None,
+            options={"device": TEST_DEVICE},
+            state=AddonState.NOT_RUNNING,
+            update_available=False,
+            version="1.0.0",
+        ),
    ) as mock_otbr_manager:
        mock_otbr_manager.async_start_addon_waiting = AsyncMock(
            side_effect=AddonError()
        )
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"], user_input={}
+        confirm_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"], user_input={}
        )
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"],
+        pick_thread_result = await hass.config_entries.flow.async_configure(
+            confirm_result["flow_id"],
            user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
        )
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        await hass.async_block_till_done(wait_background_tasks=True)

-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        assert result["type"] == FlowResultType.ABORT
-        assert result["reason"] == "addon_start_failed"
-
-
-async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) -> None:
-    """Test failure case when flasher addon uninstall fails."""
-    result = await hass.config_entries.flow.async_init(
-        TEST_DOMAIN, context={"source": "hardware"}
-    )
-
-    with mock_addon_info(
-        hass,
-        app_type=ApplicationType.EZSP,
-    ) as mock_otbr_manager:
-        mock_otbr_manager.async_uninstall_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},
-        )
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        await hass.async_block_till_done(wait_background_tasks=True)
-
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        await hass.async_block_till_done(wait_background_tasks=True)
-
-        # Uninstall failure isn't critical
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        assert result["type"] is FlowResultType.FORM
-        assert result["step_id"] == "confirm_otbr"
+        assert pick_thread_result["type"] == FlowResultType.ABORT
+        assert pick_thread_result["reason"] == "addon_start_failed"


@pytest.mark.parametrize(
@@ -331,40 +317,49 @@ async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) -
 )
 async def test_config_flow_thread_confirmation_fails(hass: HomeAssistant) -> None:
    """Test the config flow failing due to OpenThread firmware not being detected."""
-    result = await hass.config_entries.flow.async_init(
+    init_result = await hass.config_entries.flow.async_init(
        TEST_DOMAIN, context={"source": "hardware"}
    )

    with mock_addon_info(
        hass,
        app_type=ApplicationType.EZSP,
+        otbr_addon_info=AddonInfo(
+            available=True,
+            hostname=None,
+            options={"device": TEST_DEVICE},
+            state=AddonState.RUNNING,
+            update_available=False,
+            version="1.0.0",
+        ),
    ):
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"], user_input={}
+        confirm_result = await hass.config_entries.flow.async_configure(
+            init_result["flow_id"], user_input={}
        )
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"],
+        pick_thread_result = await hass.config_entries.flow.async_configure(
+            confirm_result["flow_id"],
            user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
        )
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        install_progress_result = await hass.config_entries.flow.async_configure(
+            pick_thread_result["flow_id"]
+        )
        await hass.async_block_till_done(wait_background_tasks=True)

-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        await hass.async_block_till_done(wait_background_tasks=True)
-
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-        assert result["type"] is FlowResultType.FORM
-        assert result["step_id"] == "confirm_otbr"
+        confirm_result = await hass.config_entries.flow.async_configure(
+            install_progress_result["flow_id"]
+        )
+        assert confirm_result["type"] is FlowResultType.FORM
+        assert confirm_result["step_id"] == "confirm_otbr"

    with mock_addon_info(
        hass,
        app_type=None,  # Probing fails
    ):
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"], user_input={}
+        error_result = await hass.config_entries.flow.async_configure(
+            confirm_result["flow_id"], user_input={}
        )
-        assert result["type"] is FlowResultType.ABORT
-        assert result["reason"] == "unsupported_firmware"
+        assert error_result["type"] is FlowResultType.ABORT
+        assert error_result["reason"] == "unsupported_firmware"


@pytest.mark.parametrize(