From dd7f7be6adee76f2add98dcca8d3ff87bceabf70 Mon Sep 17 00:00:00 2001 From: Martin Hjelmare Date: Tue, 23 Sep 2025 10:47:30 +0100 Subject: [PATCH] Move hardware thread add-on install after firmware install (#152800) --- .../homeassistant_connect_zbt2/config_flow.py | 2 +- .../firmware_config_flow.py | 56 ++-- .../homeassistant_sky_connect/config_flow.py | 2 +- .../homeassistant_yellow/config_flow.py | 2 +- .../test_config_flow.py | 56 ++-- .../test_config_flow.py | 288 ++++++++---------- .../test_config_flow_failures.py | 232 ++++++-------- .../test_config_flow.py | 56 ++-- .../homeassistant_yellow/test_config_flow.py | 40 +-- 9 files changed, 345 insertions(+), 389 deletions(-) diff --git a/homeassistant/components/homeassistant_connect_zbt2/config_flow.py b/homeassistant/components/homeassistant_connect_zbt2/config_flow.py index 19b7763cfd7..49243e5a97d 100644 --- a/homeassistant/components/homeassistant_connect_zbt2/config_flow.py +++ b/homeassistant/components/homeassistant_connect_zbt2/config_flow.py @@ -90,7 +90,7 @@ class ZBT2FirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol): firmware_name="OpenThread", expected_installed_firmware_type=ApplicationType.SPINEL, step_id="install_thread_firmware", - next_step_id="start_otbr_addon", + next_step_id="finish_thread_installation", ) diff --git a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py index 7f57350cc99..6df3e697fef 100644 --- a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py +++ b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py @@ -415,11 +415,39 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): if self._picked_firmware_type == PickedFirmwareType.ZIGBEE: return await self.async_step_install_zigbee_firmware() - if result := await self._ensure_thread_addon_setup(): - return result + return await self.async_step_prepare_thread_installation() + + async def async_step_prepare_thread_installation( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Prepare for Thread installation by stopping the OTBR addon if needed.""" + if not is_hassio(self.hass): + return self.async_abort( + reason="not_hassio_thread", + description_placeholders=self._get_translation_placeholders(), + ) + + otbr_manager = get_otbr_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(otbr_manager) + + if addon_info.state == AddonState.RUNNING: + # Stop the addon before continuing to flash firmware + await otbr_manager.async_stop_addon() return await self.async_step_install_thread_firmware() + async def async_step_finish_thread_installation( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Finish Thread installation by starting the OTBR addon.""" + otbr_manager = get_otbr_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(otbr_manager) + + if addon_info.state == AddonState.NOT_INSTALLED: + return await self.async_step_install_otbr_addon() + + return await self.async_step_start_otbr_addon() + async def async_step_pick_firmware_zigbee( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -495,28 +523,6 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): """Continue the ZHA flow.""" raise NotImplementedError - async def _ensure_thread_addon_setup(self) -> ConfigFlowResult | None: - """Ensure the OTBR addon is set up and not running.""" - - # We install the OTBR addon no matter what, since it is required to use Thread - if not is_hassio(self.hass): - return self.async_abort( - reason="not_hassio_thread", - description_placeholders=self._get_translation_placeholders(), - ) - - otbr_manager = get_otbr_addon_manager(self.hass) - addon_info = await self._async_get_addon_info(otbr_manager) - - if addon_info.state == AddonState.NOT_INSTALLED: - return await self.async_step_install_otbr_addon() - - if addon_info.state == AddonState.RUNNING: - # Stop the addon before continuing to flash firmware - await otbr_manager.async_stop_addon() - - return None - async def async_step_pick_firmware_thread( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -572,7 +578,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): finally: self.addon_install_task = None - return self.async_show_progress_done(next_step_id="install_thread_firmware") + return self.async_show_progress_done(next_step_id="finish_thread_installation") async def async_step_start_otbr_addon( self, user_input: dict[str, Any] | None = None diff --git a/homeassistant/components/homeassistant_sky_connect/config_flow.py b/homeassistant/components/homeassistant_sky_connect/config_flow.py index 197cb2ff2ce..7a9eff0b741 100644 --- a/homeassistant/components/homeassistant_sky_connect/config_flow.py +++ b/homeassistant/components/homeassistant_sky_connect/config_flow.py @@ -106,7 +106,7 @@ class SkyConnectFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol): firmware_name="OpenThread", expected_installed_firmware_type=ApplicationType.SPINEL, step_id="install_thread_firmware", - next_step_id="start_otbr_addon", + next_step_id="finish_thread_installation", ) diff --git a/homeassistant/components/homeassistant_yellow/config_flow.py b/homeassistant/components/homeassistant_yellow/config_flow.py index 7f84d0ddeb3..efc218caeaa 100644 --- a/homeassistant/components/homeassistant_yellow/config_flow.py +++ b/homeassistant/components/homeassistant_yellow/config_flow.py @@ -105,7 +105,7 @@ class YellowFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol): firmware_name="OpenThread", expected_installed_firmware_type=ApplicationType.SPINEL, step_id="install_thread_firmware", - next_step_id="start_otbr_addon", + next_step_id="finish_thread_installation", ) diff --git a/tests/components/homeassistant_connect_zbt2/test_config_flow.py b/tests/components/homeassistant_connect_zbt2/test_config_flow.py index 399361d453f..e3b4f7a66f5 100644 --- a/tests/components/homeassistant_connect_zbt2/test_config_flow.py +++ b/tests/components/homeassistant_connect_zbt2/test_config_flow.py @@ -1,6 +1,7 @@ """Test the Home Assistant Connect ZBT-2 config flow.""" -from unittest.mock import patch +from collections.abc import Generator +from unittest.mock import AsyncMock, call, patch import pytest @@ -23,6 +24,16 @@ from .common import USB_DATA_ZBT2 from tests.common import MockConfigEntry +@pytest.fixture(name="supervisor") +def mock_supervisor_fixture() -> Generator[None]: + """Mock Supervisor.""" + with patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio", + return_value=True, + ): + yield + + async def test_config_flow_zigbee( hass: HomeAssistant, ) -> None: @@ -51,16 +62,9 @@ async def test_config_flow_zigbee( step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._install_firmware_step", autospec=True, @@ -113,8 +117,10 @@ async def test_config_flow_zigbee( assert zha_flow["step_id"] == "confirm" +@pytest.mark.usefixtures("addon_installed", "supervisor") async def test_config_flow_thread( hass: HomeAssistant, + start_addon: AsyncMock, ) -> None: """Test Thread config flow for Connect ZBT-2.""" fw_type = ApplicationType.SPINEL @@ -141,16 +147,9 @@ async def test_config_flow_thread( step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._install_firmware_step", autospec=True, @@ -167,11 +166,23 @@ async def test_config_flow_thread( ), ), ): - confirm_result = await hass.config_entries.flow.async_configure( + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_otbr_addon" + + # Make sure the flow continues when the progress task is done. + await hass.async_block_till_done() + + confirm_result = await hass.config_entries.flow.async_configure( + result["flow_id"] + ) + + assert start_addon.call_count == 1 + assert start_addon.call_args == call("core_openthread_border_router") assert confirm_result["type"] is FlowResultType.FORM assert confirm_result["step_id"] == "confirm_otbr" @@ -244,20 +255,13 @@ async def test_options_flow( step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners", return_value=[], ), - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow._install_firmware_step", autospec=True, diff --git a/tests/components/homeassistant_hardware/test_config_flow.py b/tests/components/homeassistant_hardware/test_config_flow.py index 4040386562d..c7c2535e372 100644 --- a/tests/components/homeassistant_hardware/test_config_flow.py +++ b/tests/components/homeassistant_hardware/test_config_flow.py @@ -1,11 +1,12 @@ """Test the Home Assistant hardware firmware config flow.""" import asyncio -from collections.abc import Awaitable, Callable, Generator, Iterator +from collections.abc import AsyncGenerator, Awaitable, Callable, Iterator import contextlib from typing import Any from unittest.mock import AsyncMock, MagicMock, Mock, call, patch +from aiohasupervisor.models import AddonsOptions from aiohttp import ClientError from ha_silabs_firmware_client import ( FirmwareManifest, @@ -15,7 +16,6 @@ from ha_silabs_firmware_client import ( import pytest from yarl import URL -from homeassistant.components.hassio import AddonInfo, AddonState from homeassistant.components.homeassistant_hardware.firmware_config_flow import ( STEP_PICK_FIRMWARE_THREAD, STEP_PICK_FIRMWARE_ZIGBEE, @@ -25,7 +25,6 @@ from homeassistant.components.homeassistant_hardware.firmware_config_flow import from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, FirmwareInfo, - get_otbr_addon_manager, ) from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow from homeassistant.core import HomeAssistant, callback @@ -77,7 +76,7 @@ class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN): ) -> ConfigFlowResult: """Install Zigbee firmware.""" return await self._install_firmware_step( - fw_update_url=TEST_RELEASES_URL, + fw_update_url=str(TEST_RELEASES_URL), fw_type="fake_zigbee_ncp", firmware_name="Zigbee", expected_installed_firmware_type=ApplicationType.EZSP, @@ -90,12 +89,12 @@ class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN): ) -> ConfigFlowResult: """Install Thread firmware.""" return await self._install_firmware_step( - fw_update_url=TEST_RELEASES_URL, + fw_update_url=str(TEST_RELEASES_URL), fw_type="fake_openthread_rcp", firmware_name="Thread", expected_installed_firmware_type=ApplicationType.SPINEL, step_id="install_thread_firmware", - next_step_id="start_otbr_addon", + next_step_id="finish_thread_installation", ) def _async_flow_finished(self) -> ConfigFlowResult: @@ -139,13 +138,27 @@ class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow): self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Install Zigbee firmware.""" - return await self.async_step_pre_confirm_zigbee() + return await self._install_firmware_step( + fw_update_url=str(TEST_RELEASES_URL), + fw_type="fake_zigbee_ncp", + firmware_name="Zigbee", + expected_installed_firmware_type=ApplicationType.EZSP, + step_id="install_zigbee_firmware", + next_step_id="pre_confirm_zigbee", + ) async def async_step_install_thread_firmware( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Install Thread firmware.""" - return await self.async_step_start_otbr_addon() + return await self._install_firmware_step( + fw_update_url=str(TEST_RELEASES_URL), + fw_type="fake_openthread_rcp", + firmware_name="Thread", + expected_installed_firmware_type=ApplicationType.SPINEL, + step_id="install_thread_firmware", + next_step_id="finish_thread_installation", + ) def _async_flow_finished(self) -> ConfigFlowResult: """Create the config entry.""" @@ -166,7 +179,7 @@ class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow): @pytest.fixture(autouse=True) async def mock_test_firmware_platform( hass: HomeAssistant, -) -> Generator[None]: +) -> AsyncGenerator[None]: """Fixture for a test config flow.""" mock_module = MockModule( TEST_DOMAIN, async_setup_entry=AsyncMock(return_value=True) @@ -206,42 +219,20 @@ def create_mock_owner() -> Mock: @contextlib.contextmanager def mock_firmware_info( - hass: HomeAssistant, *, 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, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - 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)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_get_addon_info.return_value = otbr_addon_info - +) -> Iterator[Mock]: + """Mock the firmware info.""" mock_update_client = AsyncMock(spec_set=FirmwareUpdateClient) mock_update_client.async_update_data.return_value = FirmwareManifest( url=TEST_RELEASES_URL, html_url=TEST_RELEASES_URL / "html", created_at=utcnow(), - firmwares=[ + firmwares=( FirmwareMetadata( filename="fake_openthread_rcp_7.4.4.0_variant.gbl", checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef", @@ -272,7 +263,7 @@ def mock_firmware_info( }, url=TEST_RELEASES_URL / "fake_zigbee_ncp_7.4.4.0_variant.gbl", ), - ], + ), ) if probe_app_type is None: @@ -318,14 +309,6 @@ def mock_firmware_info( return flashed_firmware_info with ( - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio", return_value=is_hassio, @@ -351,7 +334,7 @@ def mock_firmware_info( side_effect=mock_flash_firmware, ), ): - yield mock_otbr_manager, mock_update_client + yield mock_update_client async def consume_progress_flow( @@ -385,7 +368,6 @@ async def test_config_flow_recommended(hass: HomeAssistant) -> None: assert init_result["step_id"] == "pick_firmware" with mock_firmware_info( - hass, probe_app_type=ApplicationType.SPINEL, flash_app_type=ApplicationType.EZSP, ): @@ -469,7 +451,6 @@ async def test_config_flow_zigbee_custom( assert init_result["step_id"] == "pick_firmware" with mock_firmware_info( - hass, probe_app_type=ApplicationType.SPINEL, flash_app_type=ApplicationType.EZSP, ): @@ -531,12 +512,11 @@ async def test_config_flow_firmware_index_download_fails_but_not_required( assert init_result["step_id"] == "pick_firmware" 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): + ) as mock_update_client: # Mock the firmware download to fail mock_update_client.async_update_data.side_effect = ClientError() @@ -567,15 +547,12 @@ async def test_config_flow_firmware_download_fails_but_not_required( assert init_result["type"] is FlowResultType.MENU assert init_result["step_id"] == "pick_firmware" - 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), - ): + with mock_firmware_info( + # 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( @@ -607,7 +584,6 @@ async def test_config_flow_doesnt_downgrade( 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", @@ -642,7 +618,9 @@ async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) -> assert result["type"] is FlowResultType.MENU assert result["step_id"] == "pick_firmware" - with mock_firmware_info(hass, probe_app_type=ApplicationType.SPINEL): + with mock_firmware_info( + probe_app_type=ApplicationType.SPINEL, + ): # Pick the menu option: we skip installation, instead we directly run it result = await hass.config_entries.flow.async_configure( result["flow_id"], @@ -659,7 +637,6 @@ async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) -> # Done with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, ): await hass.async_block_till_done(wait_background_tasks=True) @@ -693,7 +670,12 @@ async def test_config_flow_auto_confirm_if_running(hass: HomeAssistant) -> None: } -async def test_config_flow_thread(hass: HomeAssistant) -> None: +@pytest.mark.usefixtures("addon_installed") +async def test_config_flow_thread( + hass: HomeAssistant, + set_addon_options: AsyncMock, + start_addon: AsyncMock, +) -> None: """Test the config flow.""" init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} @@ -703,10 +685,9 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None: assert init_result["step_id"] == "pick_firmware" with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, flash_app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, _): + ): # Pick the menu option pick_result = await hass.config_entries.flow.async_configure( init_result["flow_id"], @@ -714,27 +695,15 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None: ) assert pick_result["type"] is FlowResultType.SHOW_PROGRESS - assert pick_result["progress_action"] == "install_addon" - assert pick_result["step_id"] == "install_otbr_addon" - assert pick_result["description_placeholders"]["firmware_type"] == "ezsp" - assert pick_result["description_placeholders"]["model"] == TEST_HARDWARE_NAME + assert pick_result["progress_action"] == "install_firmware" + assert pick_result["step_id"] == "install_thread_firmware" + description_placeholders = pick_result["description_placeholders"] + assert description_placeholders is not None + assert description_placeholders["firmware_type"] == "ezsp" + assert description_placeholders["model"] == TEST_HARDWARE_NAME await hass.async_block_till_done(wait_background_tasks=True) - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={ - "device": "", - "baudrate": 460800, - "flow_control": True, - "autoflash_firmware": False, - }, - state=AddonState.NOT_RUNNING, - update_available=False, - version="1.2.3", - ) - # Progress the flow, it is now installing firmware confirm_otbr_result = await consume_progress_flow( hass, @@ -760,37 +729,36 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None: "hardware": TEST_HARDWARE_NAME, } - assert mock_otbr_manager.async_set_addon_options.mock_calls == [ - call( - { - "device": TEST_DEVICE, + assert set_addon_options.call_args == call( + "core_openthread_border_router", + AddonsOptions( + config={ + "device": "/dev/SomeDevice123", "baudrate": 460800, "flow_control": True, "autoflash_firmware": False, - } - ) - ] + }, + ), + ) + assert start_addon.call_count == 1 + assert start_addon.call_args == call("core_openthread_border_router") -async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) -> None: +@pytest.mark.usefixtures("addon_installed") +async def test_config_flow_thread_addon_already_installed( + hass: HomeAssistant, + set_addon_options: AsyncMock, + start_addon: AsyncMock, +) -> None: """Test the Thread config flow, addon is already installed.""" init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, flash_app_type=ApplicationType.SPINEL, - otbr_addon_info=AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_RUNNING, - update_available=False, - version=None, - ), - ) as (mock_otbr_manager, _): + ): # Pick the menu option pick_result = await hass.config_entries.flow.async_configure( init_result["flow_id"], @@ -813,16 +781,19 @@ async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) - assert confirm_otbr_result["step_id"] == "confirm_otbr" # The addon has been installed - assert mock_otbr_manager.async_set_addon_options.mock_calls == [ - call( - { - "device": TEST_DEVICE, + assert set_addon_options.call_args == call( + "core_openthread_border_router", + AddonsOptions( + config={ + "device": "/dev/SomeDevice123", "baudrate": 460800, "flow_control": True, - "autoflash_firmware": False, # And firmware flashing is disabled - } - ) - ] + "autoflash_firmware": False, + }, + ), + ) + assert start_addon.call_count == 1 + assert start_addon.call_args == call("core_openthread_border_router") # Finally, create the config entry create_result = await hass.config_entries.flow.async_configure( @@ -836,8 +807,13 @@ async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) - } -@pytest.mark.usefixtures("addon_store_info") -async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: +@pytest.mark.usefixtures("addon_not_installed") +async def test_options_flow_zigbee_to_thread( + hass: HomeAssistant, + install_addon: AsyncMock, + set_addon_options: AsyncMock, + start_addon: AsyncMock, +) -> None: """Test the options flow, migrating Zigbee to Thread.""" config_entry = MockConfigEntry( domain=TEST_DOMAIN, @@ -854,16 +830,16 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: assert await hass.config_entries.async_setup(config_entry.entry_id) with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, flash_app_type=ApplicationType.SPINEL, - ) 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 assert result["step_id"] == "pick_firmware" - assert result["description_placeholders"]["firmware_type"] == "ezsp" - assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME + description_placeholders = result["description_placeholders"] + assert description_placeholders is not None + assert description_placeholders["firmware_type"] == "ezsp" + assert description_placeholders["model"] == TEST_HARDWARE_NAME result = await hass.config_entries.options.async_configure( result["flow_id"], @@ -871,49 +847,47 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: ) assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["progress_action"] == "install_addon" - assert result["step_id"] == "install_otbr_addon" + assert result["step_id"] == "install_thread_firmware" + assert result["progress_action"] == "install_firmware" await hass.async_block_till_done(wait_background_tasks=True) - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={ - "device": "", - "baudrate": 460800, - "flow_control": True, - "autoflash_firmware": False, - }, - state=AddonState.NOT_RUNNING, - update_available=False, - version="1.2.3", - ) + result = await hass.config_entries.options.async_configure(result["flow_id"]) + + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_otbr_addon" + assert result["progress_action"] == "install_addon" + + await hass.async_block_till_done(wait_background_tasks=True) - # Progress the flow, it is now configuring the addon and running it result = await hass.config_entries.options.async_configure(result["flow_id"]) assert result["type"] is FlowResultType.SHOW_PROGRESS assert result["step_id"] == "start_otbr_addon" assert result["progress_action"] == "start_otbr_addon" - assert mock_otbr_manager.async_set_addon_options.mock_calls == [ - call( - { - "device": TEST_DEVICE, + await hass.async_block_till_done(wait_background_tasks=True) + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "confirm_otbr" + assert install_addon.call_count == 1 + assert install_addon.call_args == call("core_openthread_border_router") + assert set_addon_options.call_count == 1 + assert set_addon_options.call_args == call( + "core_openthread_border_router", + AddonsOptions( + config={ + "device": "/dev/SomeDevice123", "baudrate": 460800, "flow_control": True, "autoflash_firmware": False, - } - ) - ] - - await hass.async_block_till_done(wait_background_tasks=True) - - # The addon is now running - result = await hass.config_entries.options.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_otbr" + }, + ), + ) + assert start_addon.call_count == 1 + assert start_addon.call_args == call("core_openthread_border_router") # We are now done result = await hass.config_entries.options.async_configure( @@ -951,7 +925,6 @@ async def test_options_flow_thread_to_zigbee(hass: HomeAssistant) -> None: assert description_placeholders["model"] == TEST_HARDWARE_NAME with mock_firmware_info( - hass, probe_app_type=ApplicationType.SPINEL, ): pick_result = await hass.config_entries.options.async_configure( @@ -963,15 +936,24 @@ async def test_options_flow_thread_to_zigbee(hass: HomeAssistant) -> None: assert pick_result["step_id"] == "zigbee_installation_type" with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, ): # We are now done - create_result = await hass.config_entries.options.async_configure( + result = await hass.config_entries.options.async_configure( pick_result["flow_id"], user_input={"next_step_id": "zigbee_intent_recommended"}, ) + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_zigbee_firmware" + assert result["progress_action"] == "install_firmware" + + await hass.async_block_till_done(wait_background_tasks=True) + + create_result = await hass.config_entries.options.async_configure( + result["flow_id"] + ) + assert create_result["type"] is FlowResultType.CREATE_ENTRY # The firmware type has been updated @@ -1094,7 +1076,6 @@ async def test_config_flow_zigbee_migrate_handler(hass: HomeAssistant) -> None: ) with mock_firmware_info( - hass, probe_app_type=ApplicationType.SPINEL, flash_app_type=ApplicationType.EZSP, ): @@ -1109,7 +1090,7 @@ async def test_config_flow_zigbee_migrate_handler(hass: HomeAssistant) -> None: assert result["step_id"] == "zigbee_installation_type" -@pytest.mark.usefixtures("addon_store_info") +@pytest.mark.usefixtures("addon_installed") async def test_config_flow_thread_migrate_handler(hass: HomeAssistant) -> None: """Test that the Thread migrate handler works correctly.""" # Ensure Thread migrate option is available by adding an OTBR entry @@ -1125,17 +1106,16 @@ async def test_config_flow_thread_migrate_handler(hass: HomeAssistant) -> None: ) with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, flash_app_type=ApplicationType.SPINEL, - ) as (_, _): + ): # Test the migrate handler directly result = await hass.config_entries.flow.async_configure( init_result["flow_id"], user_input={"next_step_id": "pick_firmware_thread_migrate"}, ) - # Should proceed to OTBR addon installation (same as normal thread flow) + # Should proceed to firmware install (same as normal thread flow) assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["progress_action"] == "install_addon" - assert result["step_id"] == "install_otbr_addon" + assert result["progress_action"] == "install_firmware" + assert result["step_id"] == "install_thread_firmware" diff --git a/tests/components/homeassistant_hardware/test_config_flow_failures.py b/tests/components/homeassistant_hardware/test_config_flow_failures.py index e02faf97ced..217c331257e 100644 --- a/tests/components/homeassistant_hardware/test_config_flow_failures.py +++ b/tests/components/homeassistant_hardware/test_config_flow_failures.py @@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, patch from aiohttp import ClientError import pytest -from homeassistant.components.hassio import AddonError, AddonInfo, AddonState +from homeassistant.components.hassio import AddonError from homeassistant.components.homeassistant_hardware.firmware_config_flow import ( STEP_PICK_FIRMWARE_THREAD, STEP_PICK_FIRMWARE_ZIGBEE, @@ -13,6 +13,7 @@ from homeassistant.components.homeassistant_hardware.firmware_config_flow import from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, FirmwareInfo, + OwningAddon, OwningIntegration, ) from homeassistant.core import HomeAssistant @@ -44,7 +45,6 @@ async def test_config_flow_cannot_probe_firmware_zigbee(hass: HomeAssistant) -> """Test failure case when firmware cannot be probed for zigbee.""" with mock_firmware_info( - hass, probe_app_type=None, ): # Start the flow @@ -77,7 +77,7 @@ async def test_config_flow_cannot_probe_firmware_zigbee(hass: HomeAssistant) -> ["test_firmware_domain"], ) async def test_cannot_probe_after_install_zigbee(hass: HomeAssistant) -> None: - """Test unsupported firmware after install for Zigbee.""" + """Test unsupported firmware after firmware install for Zigbee.""" init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) @@ -86,7 +86,6 @@ async def test_cannot_probe_after_install_zigbee(hass: HomeAssistant) -> None: assert init_result["step_id"] == "pick_firmware" with mock_firmware_info( - hass, probe_app_type=ApplicationType.SPINEL, flash_app_type=ApplicationType.EZSP, ): @@ -109,7 +108,6 @@ async def test_cannot_probe_after_install_zigbee(hass: HomeAssistant) -> None: assert pick_result["step_id"] == "install_zigbee_firmware" with mock_firmware_info( - hass, probe_app_type=None, flash_app_type=ApplicationType.EZSP, ): @@ -132,7 +130,6 @@ async def test_config_flow_cannot_probe_firmware_thread(hass: HomeAssistant) -> """Test failure case when firmware cannot be probed for thread.""" with mock_firmware_info( - hass, probe_app_type=None, ): # Start the flow @@ -156,9 +153,9 @@ async def test_config_flow_cannot_probe_firmware_thread(hass: HomeAssistant) -> "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -@pytest.mark.usefixtures("addon_store_info") +@pytest.mark.usefixtures("addon_installed") async def test_cannot_probe_after_install_thread(hass: HomeAssistant) -> None: - """Test unsupported firmware after install for thread.""" + """Test unsupported firmware after firmware install for thread.""" init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) @@ -167,10 +164,9 @@ async def test_cannot_probe_after_install_thread(hass: HomeAssistant) -> None: assert init_result["step_id"] == "pick_firmware" with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, flash_app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, _): + ): # Pick the menu option pick_result = await hass.config_entries.flow.async_configure( init_result["flow_id"], @@ -178,31 +174,14 @@ async def test_cannot_probe_after_install_thread(hass: HomeAssistant) -> None: ) assert pick_result["type"] is FlowResultType.SHOW_PROGRESS - assert pick_result["progress_action"] == "install_addon" - assert pick_result["step_id"] == "install_otbr_addon" + assert pick_result["progress_action"] == "install_firmware" + assert pick_result["step_id"] == "install_thread_firmware" description_placeholders = pick_result["description_placeholders"] assert description_placeholders is not None assert description_placeholders["firmware_type"] == "ezsp" assert description_placeholders["model"] == TEST_HARDWARE_NAME - await hass.async_block_till_done(wait_background_tasks=True) - - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={ - "device": "", - "baudrate": 460800, - "flow_control": True, - "autoflash_firmware": False, - }, - state=AddonState.NOT_RUNNING, - update_available=False, - version="1.2.3", - ) - with mock_firmware_info( - hass, probe_app_type=None, flash_app_type=ApplicationType.SPINEL, ): @@ -232,15 +211,13 @@ async def test_config_flow_thread_not_hassio(hass: HomeAssistant) -> None: TEST_DOMAIN, context={"source": "hardware"} ) + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + with mock_firmware_info( - hass, is_hassio=False, probe_app_type=ApplicationType.EZSP, ): - 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}, @@ -253,20 +230,23 @@ async def test_config_flow_thread_not_hassio(hass: HomeAssistant) -> None: "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None: - """Test failure case when flasher addon cannot be installed.""" +async def test_config_flow_thread_addon_info_fails( + hass: HomeAssistant, + addon_store_info: AsyncMock, +) -> None: + """Test addon info fails before firmware install.""" + result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, - ) 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={} - ) + ): + addon_store_info.side_effect = AddonError() result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -277,73 +257,75 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None: assert result["reason"] == "addon_info_failed" +@pytest.mark.usefixtures("addon_not_installed") @pytest.mark.parametrize( "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> None: +async def test_config_flow_thread_addon_install_fails( + hass: HomeAssistant, + install_addon: AsyncMock, +) -> None: """Test failure case when flasher addon cannot be installed.""" result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - with mock_firmware_info( - hass, - probe_app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, _): - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=AddonError() - ) + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + + with mock_firmware_info( + probe_app_type=ApplicationType.EZSP, + ): + install_addon.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}, ) + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_thread_firmware" + assert result["progress_action"] == "install_firmware" + + result = await consume_progress_flow( + hass, + flow_id=result["flow_id"], + valid_step_ids=( + "install_otbr_addon", + "install_thread_firmware", + ), + ) + # Cannot install addon assert result["type"] == FlowResultType.ABORT assert result["reason"] == "addon_install_failed" +@pytest.mark.usefixtures("addon_installed") @pytest.mark.parametrize( "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> None: +async def test_config_flow_thread_addon_set_config_fails( + hass: HomeAssistant, + set_addon_options: AsyncMock, +) -> None: """Test failure case when flasher addon cannot be configured.""" init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) + assert init_result["type"] is FlowResultType.MENU + assert init_result["step_id"] == "pick_firmware" + with mock_firmware_info( - hass, probe_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()) - - confirm_result = await hass.config_entries.flow.async_configure( - init_result["flow_id"], user_input={} - ) + ): + set_addon_options.side_effect = AddonError() pick_thread_result = await hass.config_entries.flow.async_configure( - confirm_result["flow_id"], + init_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) @@ -361,36 +343,29 @@ async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> assert pick_thread_progress_result["reason"] == "addon_set_config_failed" +@pytest.mark.usefixtures("addon_installed") @pytest.mark.parametrize( "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None: +async def test_config_flow_thread_flasher_run_fails( + hass: HomeAssistant, + start_addon: AsyncMock, +) -> None: """Test failure case when flasher addon fails to run.""" + start_addon.side_effect = AddonError() init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) + assert init_result["type"] is FlowResultType.MENU + assert init_result["step_id"] == "pick_firmware" + with mock_firmware_info( - hass, probe_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() - ) - confirm_result = await hass.config_entries.flow.async_configure( - init_result["flow_id"], user_input={} - ) + ): pick_thread_result = await hass.config_entries.flow.async_configure( - confirm_result["flow_id"], + init_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) @@ -408,6 +383,7 @@ async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None assert pick_thread_progress_result["reason"] == "addon_start_failed" +@pytest.mark.usefixtures("addon_running") @pytest.mark.parametrize( "ignore_translations_for_mock_domains", ["test_firmware_domain"], @@ -418,24 +394,15 @@ async def test_config_flow_thread_confirmation_fails(hass: HomeAssistant) -> Non TEST_DOMAIN, context={"source": "hardware"} ) + assert init_result["type"] is FlowResultType.MENU + assert init_result["step_id"] == "pick_firmware" + with mock_firmware_info( - hass, probe_app_type=ApplicationType.EZSP, flash_app_type=None, - otbr_addon_info=AddonInfo( - available=True, - hostname=None, - options={"device": TEST_DEVICE}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ), ): - confirm_result = await hass.config_entries.flow.async_configure( - init_result["flow_id"], user_input={} - ) pick_thread_result = await hass.config_entries.flow.async_configure( - confirm_result["flow_id"], + init_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) @@ -467,13 +434,10 @@ async def test_config_flow_firmware_index_download_fails_and_required( assert init_result["type"] is FlowResultType.MENU assert init_result["step_id"] == "pick_firmware" - 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), - ): + with mock_firmware_info( + # 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( @@ -507,13 +471,10 @@ async def test_config_flow_firmware_download_fails_and_required( assert init_result["type"] is FlowResultType.MENU assert init_result["step_id"] == "pick_firmware" - 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), - ): + with mock_firmware_info( + # 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( @@ -585,7 +546,6 @@ async def test_options_flow_zigbee_to_thread_zha_configured( "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -@pytest.mark.usefixtures("addon_store_info") async def test_options_flow_thread_to_zigbee_otbr_configured( hass: HomeAssistant, ) -> None: @@ -607,21 +567,23 @@ async def test_options_flow_thread_to_zigbee_otbr_configured( # Confirm options flow result = await hass.config_entries.options.async_init(config_entry.entry_id) - with mock_firmware_info( - hass, - probe_app_type=ApplicationType.SPINEL, - otbr_addon_info=AddonInfo( - available=True, - hostname=None, - options={"device": TEST_DEVICE}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ), + # Pretend OTBR is using the stick + with patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners", + return_value=[ + FirmwareInfo( + device=TEST_DEVICE, + firmware_type=ApplicationType.EZSP, + firmware_version="1.2.3.4", + source="otbr", + owners=[OwningAddon(slug="openthread_border_router")], + ) + ], ): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, ) - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "otbr_still_using_stick" + + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "otbr_still_using_stick" diff --git a/tests/components/homeassistant_sky_connect/test_config_flow.py b/tests/components/homeassistant_sky_connect/test_config_flow.py index d9b98966f1d..2b863450d7d 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow.py @@ -1,6 +1,7 @@ """Test the Home Assistant SkyConnect config flow.""" -from unittest.mock import Mock, patch +from collections.abc import Generator +from unittest.mock import AsyncMock, Mock, call, patch import pytest @@ -29,6 +30,16 @@ from .common import USB_DATA_SKY, USB_DATA_ZBT1 from tests.common import MockConfigEntry +@pytest.fixture(name="supervisor") +def mock_supervisor_fixture() -> Generator[None]: + """Mock Supervisor.""" + with patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio", + return_value=True, + ): + yield + + @pytest.mark.parametrize( ("usb_data", "model"), [ @@ -70,16 +81,9 @@ async def test_config_flow_zigbee( step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._install_firmware_step", autospec=True, @@ -133,6 +137,7 @@ async def test_config_flow_zigbee( assert zha_flow["step_id"] == "confirm" +@pytest.mark.usefixtures("addon_installed", "supervisor") @pytest.mark.parametrize( ("usb_data", "model"), [ @@ -150,6 +155,7 @@ async def test_config_flow_thread( usb_data: UsbServiceInfo, model: str, hass: HomeAssistant, + start_addon: AsyncMock, ) -> None: """Test the config flow for SkyConnect with Thread.""" fw_type = ApplicationType.SPINEL @@ -174,16 +180,9 @@ async def test_config_flow_thread( step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._install_firmware_step", autospec=True, @@ -200,11 +199,23 @@ async def test_config_flow_thread( ), ), ): - confirm_result = await hass.config_entries.flow.async_configure( + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_otbr_addon" + + # Make sure the flow continues when the progress task is done. + await hass.async_block_till_done() + + confirm_result = await hass.config_entries.flow.async_configure( + result["flow_id"] + ) + + assert start_addon.call_count == 1 + assert start_addon.call_args == call("core_openthread_border_router") assert confirm_result["type"] is FlowResultType.FORM assert confirm_result["step_id"] == ("confirm_otbr") @@ -279,20 +290,13 @@ async def test_options_flow( step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners", return_value=[], ), - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow._install_firmware_step", autospec=True, diff --git a/tests/components/homeassistant_yellow/test_config_flow.py b/tests/components/homeassistant_yellow/test_config_flow.py index 815163ce206..518a1d3b4d1 100644 --- a/tests/components/homeassistant_yellow/test_config_flow.py +++ b/tests/components/homeassistant_yellow/test_config_flow.py @@ -1,7 +1,7 @@ """Test the Home Assistant Yellow config flow.""" from collections.abc import Generator -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import AsyncMock, Mock, call, patch import pytest @@ -352,20 +352,13 @@ async def test_firmware_options_flow_zigbee(hass: HomeAssistant) -> None: step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners", return_value=[], ), - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareInstallFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareInstallFlow._install_firmware_step", autospec=True, @@ -403,8 +396,10 @@ async def test_firmware_options_flow_zigbee(hass: HomeAssistant) -> None: } -@pytest.mark.usefixtures("addon_store_info") -async def test_firmware_options_flow_thread(hass: HomeAssistant) -> None: +@pytest.mark.usefixtures("addon_installed") +async def test_firmware_options_flow_thread( + hass: HomeAssistant, start_addon: AsyncMock +) -> None: """Test the firmware options flow for Yellow with Thread.""" fw_type = ApplicationType.SPINEL fw_version = "2.4.4.0" @@ -448,20 +443,13 @@ async def test_firmware_options_flow_thread(hass: HomeAssistant) -> None: step_id: str, next_step_id: str, ) -> ConfigFlowResult: - if next_step_id == "start_otbr_addon": - next_step_id = "pre_confirm_otbr" - - return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + return await getattr(self, f"async_step_{next_step_id}")() with ( patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners", return_value=[], ), - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareInstallFlow._ensure_thread_addon_setup", - return_value=None, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareInstallFlow._install_firmware_step", autospec=True, @@ -478,11 +466,23 @@ async def test_firmware_options_flow_thread(hass: HomeAssistant) -> None: ), ), ): - confirm_result = await hass.config_entries.options.async_configure( + result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_otbr_addon" + + # Make sure the flow continues when the progress task is done. + await hass.async_block_till_done() + + confirm_result = await hass.config_entries.options.async_configure( + result["flow_id"] + ) + + assert start_addon.call_count == 1 + assert start_addon.call_args == call("core_openthread_border_router") assert confirm_result["type"] is FlowResultType.FORM assert confirm_result["step_id"] == ("confirm_otbr")