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
This commit is contained in:
puddly 2024-05-22 14:09:30 -04:00 committed by GitHub
parent 6113b58e9c
commit 55c8ef1c7b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 294 additions and 700 deletions

View file

@ -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)

View file

@ -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": {

View file

@ -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"],

View file

@ -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},