From 55c8ef1c7b5f5a2e2adf8210c3762c0ccac21f40 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Wed, 22 May 2024 14:09:30 -0400 Subject: [PATCH] Simplify SkyConnect setup flow (#117868) * Delay firmware probing until after the user picks the firmware type * Remove confirmation step * Fix unit tests * Simplify unit test patching logic Further simplify unit tests * Bump Zigbee firmware up to the first choice * Reuse `async_step_pick_firmware` during options flow * Proactively validate all ZHA entries, not just the first There can only be one (for now) so this changes nothing functionally * Add unit test for bad firmware when configuring Thread --- .../homeassistant_sky_connect/config_flow.py | 75 ++- .../homeassistant_sky_connect/strings.json | 14 +- .../test_config_flow.py | 371 ++++-------- .../test_config_flow_failures.py | 534 +++++------------- 4 files changed, 294 insertions(+), 700 deletions(-) diff --git a/homeassistant/components/homeassistant_sky_connect/config_flow.py b/homeassistant/components/homeassistant_sky_connect/config_flow.py index a65aefe96f2..8eeb703248a 100644 --- a/homeassistant/components/homeassistant_sky_connect/config_flow.py +++ b/homeassistant/components/homeassistant_sky_connect/config_flow.py @@ -121,6 +121,17 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Pick Thread or Zigbee firmware.""" + return self.async_show_menu( + step_id="pick_firmware", + menu_options=[ + STEP_PICK_FIRMWARE_ZIGBEE, + STEP_PICK_FIRMWARE_THREAD, + ], + description_placeholders=self._get_translation_placeholders(), + ) + + async def _probe_firmware_type(self) -> bool: + """Probe the firmware currently on the device.""" assert self._usb_info is not None self._probed_firmware_type = await probe_silabs_firmware_type( @@ -134,29 +145,22 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): ), ) - if self._probed_firmware_type not in ( + return self._probed_firmware_type in ( ApplicationType.EZSP, ApplicationType.SPINEL, ApplicationType.CPC, - ): - return self.async_abort( - reason="unsupported_firmware", - description_placeholders=self._get_translation_placeholders(), - ) - - return self.async_show_menu( - step_id="pick_firmware", - menu_options=[ - STEP_PICK_FIRMWARE_THREAD, - STEP_PICK_FIRMWARE_ZIGBEE, - ], - description_placeholders=self._get_translation_placeholders(), ) async def async_step_pick_firmware_zigbee( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Pick Zigbee firmware.""" + if not await self._probe_firmware_type(): + return self.async_abort( + reason="unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + # Allow the stick to be used with ZHA without flashing if self._probed_firmware_type == ApplicationType.EZSP: return await self.async_step_confirm_zigbee() @@ -372,6 +376,12 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Pick Thread firmware.""" + if not await self._probe_firmware_type(): + return self.async_abort( + reason="unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + # 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( @@ -528,17 +538,7 @@ class HomeAssistantSkyConnectConfigFlow( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Confirm a discovery.""" - self._set_confirm_only() - - # Without confirmation, discovery can automatically progress into parts of the - # config flow logic that interacts with hardware. - if user_input is not None: - return await self.async_step_pick_firmware() - - return self.async_show_form( - step_id="confirm", - description_placeholders=self._get_translation_placeholders(), - ) + return await self.async_step_pick_firmware() def _async_flow_finished(self) -> ConfigFlowResult: """Create the config entry.""" @@ -641,15 +641,7 @@ class HomeAssistantSkyConnectOptionsFlowHandler( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Manage the options flow.""" - # Don't probe the running firmware, we load it from the config entry - return self.async_show_menu( - step_id="pick_firmware", - menu_options=[ - STEP_PICK_FIRMWARE_THREAD, - STEP_PICK_FIRMWARE_ZIGBEE, - ], - description_placeholders=self._get_translation_placeholders(), - ) + return await self.async_step_pick_firmware() async def async_step_pick_firmware_zigbee( self, user_input: dict[str, Any] | None = None @@ -678,17 +670,16 @@ class HomeAssistantSkyConnectOptionsFlowHandler( """Pick Thread firmware.""" assert self._usb_info is not None - zha_entries = self.hass.config_entries.async_entries( + for zha_entry in self.hass.config_entries.async_entries( ZHA_DOMAIN, include_ignore=False, include_disabled=True, - ) - - if zha_entries and get_zha_device_path(zha_entries[0]) == self._usb_info.device: - raise AbortFlow( - "zha_still_using_stick", - description_placeholders=self._get_translation_placeholders(), - ) + ): + if get_zha_device_path(zha_entry) == self._usb_info.device: + raise AbortFlow( + "zha_still_using_stick", + description_placeholders=self._get_translation_placeholders(), + ) return await super().async_step_pick_firmware_thread(user_input) diff --git a/homeassistant/components/homeassistant_sky_connect/strings.json b/homeassistant/components/homeassistant_sky_connect/strings.json index 792406dcb02..59bcb6e606a 100644 --- a/homeassistant/components/homeassistant_sky_connect/strings.json +++ b/homeassistant/components/homeassistant_sky_connect/strings.json @@ -58,10 +58,6 @@ "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]", "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]" }, - "confirm": { - "title": "[%key:component::homeassistant_sky_connect::config::step::confirm::title%]", - "description": "[%key:component::homeassistant_sky_connect::config::step::confirm::description%]" - }, "pick_firmware": { "title": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::title%]", "description": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::description%]", @@ -131,16 +127,12 @@ "config": { "flow_title": "{model}", "step": { - "confirm": { - "title": "Set up the {model}", - "description": "The {model} can be used as either a Thread border router or a Zigbee coordinator. In the next step, you will choose which firmware will be configured." - }, "pick_firmware": { "title": "Pick your firmware", - "description": "The {model} can be used as a Thread border router or a Zigbee coordinator.", + "description": "Let's get started with setting up your {model}. Do you want to use it to set up a Zigbee or Thread network?", "menu_options": { - "pick_firmware_thread": "Use as a Thread border router", - "pick_firmware_zigbee": "Use as a Zigbee coordinator" + "pick_firmware_zigbee": "Zigbee", + "pick_firmware_thread": "Thread" } }, "install_zigbee_flasher_addon": { diff --git a/tests/components/homeassistant_sky_connect/test_config_flow.py b/tests/components/homeassistant_sky_connect/test_config_flow.py index 611dda4a917..a4b7b4fb81d 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow.py @@ -2,6 +2,7 @@ import asyncio from collections.abc import Awaitable, Callable +import contextlib from typing import Any from unittest.mock import AsyncMock, Mock, call, patch @@ -57,6 +58,77 @@ def delayed_side_effect() -> Callable[..., Awaitable[None]]: return side_effect +@contextlib.contextmanager +def mock_addon_info( + hass: HomeAssistant, + *, + is_hassio: bool = True, + app_type: ApplicationType = ApplicationType.EZSP, + otbr_addon_info: AddonInfo = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ), + flasher_addon_info: AddonInfo = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ), +): + """Mock the main addon states for the config flow.""" + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_get_addon_info.return_value = flasher_addon_info + + 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 + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=is_hassio, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=app_type, + ), + ): + yield mock_otbr_manager, mock_flasher_manager + + @pytest.mark.parametrize( ("usb_data", "model"), [ @@ -72,57 +144,13 @@ async def test_config_flow_zigbee( DOMAIN, context={"source": "usb"}, data=usb_data ) - # First step is confirmation, we haven't probed the firmware yet - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm" - assert result["description_placeholders"]["firmware_type"] == "unknown" - assert result["description_placeholders"]["model"] == model - - # Next, we probe the firmware - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, # Ensure we re-install it - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - assert result["type"] is FlowResultType.MENU assert result["step_id"] == "pick_firmware" - assert result["description_placeholders"]["firmware_type"] == "spinel" - - # Set up Zigbee firmware - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + ) as (mock_otbr_manager, mock_flasher_manager): # Pick the menu option: we are now installing the addon result = await hass.config_entries.flow.async_configure( result["flow_id"], @@ -131,6 +159,7 @@ async def test_config_flow_zigbee( assert result["type"] is FlowResultType.SHOW_PROGRESS assert result["progress_action"] == "install_addon" assert result["step_id"] == "install_zigbee_flasher_addon" + assert result["description_placeholders"]["firmware_type"] == "spinel" await hass.async_block_till_done(wait_background_tasks=True) @@ -208,46 +237,13 @@ async def test_config_flow_zigbee_skip_step_if_installed( DOMAIN, context={"source": "usb"}, data=usb_data ) - # First step is confirmation, we haven't probed the firmware yet - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm" - assert result["description_placeholders"]["firmware_type"] == "unknown" - assert result["description_placeholders"]["model"] == model - - # Next, we probe the firmware - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, # Ensure we re-install it - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - assert result["type"] is FlowResultType.MENU assert result["step_id"] == "pick_firmware" - assert result["description_placeholders"]["firmware_type"] == "spinel" - # Set up Zigbee firmware - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + flasher_addon_info=AddonInfo( available=True, hostname=None, options={ @@ -259,16 +255,18 @@ async def test_config_flow_zigbee_skip_step_if_installed( state=AddonState.NOT_RUNNING, update_available=False, version="1.2.3", - ) - + ), + ) as (mock_otbr_manager, mock_flasher_manager): # Pick the menu option: we skip installation, instead we directly run it result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, ) + assert result["type"] is FlowResultType.SHOW_PROGRESS assert result["step_id"] == "run_zigbee_flasher_addon" assert result["progress_action"] == "run_zigbee_flasher_addon" + assert result["description_placeholders"]["firmware_type"] == "spinel" assert mock_flasher_manager.async_set_addon_options.mock_calls == [ call( { @@ -306,54 +304,13 @@ async def test_config_flow_thread( DOMAIN, context={"source": "usb"}, data=usb_data ) - # First step is confirmation, we haven't probed the firmware yet - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm" - assert result["description_placeholders"]["firmware_type"] == "unknown" - assert result["description_placeholders"]["model"] == model - - # Next, we probe the firmware - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - assert result["type"] is FlowResultType.MENU assert result["step_id"] == "pick_firmware" - assert result["description_placeholders"]["firmware_type"] == "ezsp" - - # Set up Thread firmware - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): # Pick the menu option result = await hass.config_entries.flow.async_configure( result["flow_id"], @@ -363,6 +320,8 @@ async def test_config_flow_thread( assert result["type"] is FlowResultType.SHOW_PROGRESS assert result["progress_action"] == "install_addon" assert result["step_id"] == "install_otbr_addon" + assert result["description_placeholders"]["firmware_type"] == "ezsp" + assert result["description_placeholders"]["model"] == model await hass.async_block_till_done(wait_background_tasks=True) @@ -438,41 +397,18 @@ async def test_config_flow_thread_addon_already_installed( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - - 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_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_RUNNING, - update_available=False, - version=None, - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + otbr_addon_info=AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_RUNNING, + update_available=False, + version=None, ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): + ) as (mock_otbr_manager, mock_flasher_manager): # Pick the menu option result = await hass.config_entries.flow.async_configure( result["flow_id"], @@ -520,20 +456,11 @@ async def test_config_flow_zigbee_not_hassio( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=False, - ), - ): + with mock_addon_info( + hass, + is_hassio=False, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -604,35 +531,10 @@ async def test_options_flow_zigbee_to_thread( assert result["description_placeholders"]["firmware_type"] == "ezsp" assert result["description_placeholders"]["model"] == model - # Pick Thread - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -730,53 +632,10 @@ async def test_options_flow_thread_to_zigbee( assert result["description_placeholders"]["firmware_type"] == "spinel" assert result["description_placeholders"]["model"] == model - # Set up Zigbee firmware - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - - # OTBR is not installed - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + ) as (mock_otbr_manager, mock_flasher_manager): # Pick the menu option: we are now installing the addon result = await hass.config_entries.options.async_configure( result["flow_id"], diff --git a/tests/components/homeassistant_sky_connect/test_config_flow_failures.py b/tests/components/homeassistant_sky_connect/test_config_flow_failures.py index 128c812272f..b29f8d808ae 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow_failures.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow_failures.py @@ -1,6 +1,6 @@ """Test the Home Assistant SkyConnect config flow failure cases.""" -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import AsyncMock import pytest from universal_silabs_flasher.const import ApplicationType @@ -16,41 +16,38 @@ from homeassistant.components.homeassistant_sky_connect.config_flow import ( STEP_PICK_FIRMWARE_ZIGBEE, ) from homeassistant.components.homeassistant_sky_connect.const import DOMAIN -from homeassistant.components.homeassistant_sky_connect.util import ( - get_otbr_addon_manager, - get_zigbee_flasher_addon_manager, -) from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType -from .test_config_flow import USB_DATA_ZBT1, delayed_side_effect +from .test_config_flow import USB_DATA_ZBT1, delayed_side_effect, mock_addon_info from tests.common import MockConfigEntry @pytest.mark.parametrize( - ("usb_data", "model"), + ("usb_data", "model", "next_step"), [ - (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1", STEP_PICK_FIRMWARE_ZIGBEE), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1", STEP_PICK_FIRMWARE_THREAD), ], ) async def test_config_flow_cannot_probe_firmware( - usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant + usb_data: usb.UsbServiceInfo, model: str, next_step: str, hass: HomeAssistant ) -> None: """Test failure case when firmware cannot be probed.""" - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=None, - ): + with mock_addon_info( + hass, + app_type=None, + ) as (mock_otbr_manager, mock_flasher_manager): # Start the flow result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": "usb"}, data=usb_data ) - # Probing fails result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + result["flow_id"], + user_input={"next_step_id": next_step}, ) assert result["type"] == FlowResultType.ABORT @@ -71,20 +68,15 @@ async def test_config_flow_zigbee_not_hassio_wrong_firmware( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + is_hassio=False, + ) as (mock_otbr_manager, mock_flasher_manager): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=False, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -107,35 +99,22 @@ async def test_config_flow_zigbee_flasher_addon_already_running( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + flasher_addon_info=AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ), + ) as (mock_otbr_manager, mock_flasher_manager): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -160,28 +139,23 @@ async def test_config_flow_zigbee_flasher_addon_info_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + flasher_addon_info=AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ), + ) as (mock_otbr_manager, mock_flasher_manager): + mock_flasher_manager.async_get_addon_info.side_effect = AddonError() + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.side_effect = AddonError() - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -206,38 +180,18 @@ async def test_config_flow_zigbee_flasher_addon_install_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -262,39 +216,20 @@ async def test_config_flow_zigbee_flasher_addon_set_config_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_set_addon_options = AsyncMock( + side_effect=AddonError() + ) + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_set_addon_options = AsyncMock(side_effect=AddonError()) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -321,39 +256,17 @@ async def test_config_flow_zigbee_flasher_run_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_flasher_manager.async_start_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_start_addon_waiting = AsyncMock(side_effect=AddonError()) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -380,44 +293,16 @@ async def test_config_flow_zigbee_flasher_uninstall_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.SPINEL, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=AddonError() + ) result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, @@ -448,20 +333,15 @@ async def test_config_flow_thread_not_hassio( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + is_hassio=False, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=False, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -484,28 +364,14 @@ async def test_config_flow_thread_addon_info_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_otbr_manager.async_get_addon_info.side_effect = AddonError() result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.side_effect = AddonError() - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -530,36 +396,25 @@ async def test_config_flow_thread_addon_already_running( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + otbr_addon_info=AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ), + ) as (mock_otbr_manager, mock_flasher_manager): + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ) - mock_otbr_manager.async_install_addon_waiting = AsyncMock(side_effect=AddonError()) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -584,36 +439,17 @@ async def test_config_flow_thread_addon_install_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_otbr_manager.async_install_addon_waiting = AsyncMock(side_effect=AddonError()) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -638,39 +474,15 @@ async def test_config_flow_thread_addon_set_config_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_manager): + mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError()) + result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) - - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError()) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -697,39 +509,16 @@ async def test_config_flow_thread_flasher_run_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_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={} ) - - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_otbr_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_otbr_manager.async_start_addon_waiting = AsyncMock(side_effect=AddonError()) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -756,44 +545,17 @@ async def test_config_flow_thread_flasher_uninstall_fails( DOMAIN, context={"source": "usb"}, data=usb_data ) - with patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", - return_value=ApplicationType.EZSP, - ): + with mock_addon_info( + hass, + app_type=ApplicationType.EZSP, + ) as (mock_otbr_manager, mock_flasher_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={} ) - - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ) - mock_otbr_manager.async_install_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_uninstall_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, - ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, @@ -890,28 +652,18 @@ 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) - # Pick Zigbee - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) - mock_otbr_manager.addon_name = "OpenThread Border Router" - mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={"device": usb_data.device}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ) - - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", - return_value=mock_otbr_manager, + with mock_addon_info( + hass, + app_type=ApplicationType.SPINEL, + otbr_addon_info=AddonInfo( + available=True, + hostname=None, + options={"device": usb_data.device}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", ), - patch( - "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", - return_value=True, - ), - ): + ) as (mock_otbr_manager, mock_flasher_manager): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},