Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 08:06:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

Move hardware thread add-on install after firmware install (#152800)

Browse Source
This commit is contained in:
Martin Hjelmare
2025-09-23 10:47:30 +01:00
committed by GitHub
parent 22709506c6
commit dd7f7be6ad
9 changed files with 345 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/homeassistant_connect_zbt2/config_flow.py
View File

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

56
homeassistant/components/homeassistant_hardware/firmware_config_flow.py
View File

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

2
homeassistant/components/homeassistant_sky_connect/config_flow.py
View File

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

2
homeassistant/components/homeassistant_yellow/config_flow.py
View File

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

56
tests/components/homeassistant_connect_zbt2/test_config_flow.py
View File

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

288
tests/components/homeassistant_hardware/test_config_flow.py
View File

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

232
tests/components/homeassistant_hardware/test_config_flow_failures.py
View File

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

56
tests/components/homeassistant_sky_connect/test_config_flow.py
View File

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

40
tests/components/homeassistant_yellow/test_config_flow.py
View File

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