starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-30 18:14:50 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add hardware Zigbee flow strategy (#153190)

Browse Source
This commit is contained in:
Martin Hjelmare
2025-09-29 20:08:43 +02:00
committed by GitHub
parent 81917425dc
commit cfab789823
5 changed files with 428 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/homeassistant_hardware/firmware_config_flow.py
+11
View File
@@ -61,6 +61,13 @@ class PickedFirmwareType(StrEnum):
ZIGBEE = "zigbee"
class ZigbeeFlowStrategy(StrEnum):
"""Zigbee setup strategies that can be picked."""
ADVANCED = "advanced"
RECOMMENDED = "recommended"
class ZigbeeIntegration(StrEnum):
"""Zigbee integrations that can be picked."""
@@ -73,6 +80,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
ZIGBEE_BAUDRATE = 115200 # Default, subclasses may override
_picked_firmware_type: PickedFirmwareType
_zigbee_flow_strategy: ZigbeeFlowStrategy = ZigbeeFlowStrategy.RECOMMENDED
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate base flow."""
@@ -395,12 +403,14 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
) -> ConfigFlowResult:
"""Select recommended installation type."""
self._zigbee_integration = ZigbeeIntegration.ZHA
self._zigbee_flow_strategy = ZigbeeFlowStrategy.RECOMMENDED
return await self._async_continue_picked_firmware()
async def async_step_zigbee_intent_custom(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Select custom installation type."""
self._zigbee_flow_strategy = ZigbeeFlowStrategy.ADVANCED
return await self.async_step_zigbee_integration()
async def async_step_zigbee_integration(
@@ -521,6 +531,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
"flow_control": "hardware",
},
"radio_type": "ezsp",
"flow_strategy": self._zigbee_flow_strategy,
},
)
return self._continue_zha_flow(result)
homeassistant/components/zha/config_flow.py
+17
View File
@@ -20,6 +20,9 @@ from homeassistant.components import onboarding, usb
from homeassistant.components.file_upload import process_uploaded_file
from homeassistant.components.hassio import AddonError, AddonState
from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
ZigbeeFlowStrategy,
)
from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware
from homeassistant.config_entries import (
SOURCE_IGNORE,
@@ -163,6 +166,7 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]:
class BaseZhaFlow(ConfigEntryBaseFlow):
"""Mixin for common ZHA flow steps and forms."""
_flow_strategy: ZigbeeFlowStrategy | None = None
_hass: HomeAssistant
_title: str
@@ -373,6 +377,12 @@ class BaseZhaFlow(ConfigEntryBaseFlow):
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Choose how to set up the integration from scratch."""
if self._flow_strategy == ZigbeeFlowStrategy.RECOMMENDED:
# Fast path: automatically form a new network
return await self.async_step_setup_strategy_recommended()
if self._flow_strategy == ZigbeeFlowStrategy.ADVANCED:
# Advanced path: let the user choose
return await self.async_step_setup_strategy_advanced()
# Allow onboarding for new users to just create a new network automatically
if (
@@ -406,6 +416,12 @@ class BaseZhaFlow(ConfigEntryBaseFlow):
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Choose how to deal with the current radio's settings during migration."""
if self._flow_strategy == ZigbeeFlowStrategy.RECOMMENDED:
# Fast path: automatically migrate everything
return await self.async_step_migration_strategy_recommended()
if self._flow_strategy == ZigbeeFlowStrategy.ADVANCED:
# Advanced path: let the user choose
return await self.async_step_migration_strategy_advanced()
return self.async_show_menu(
step_id="choose_migration_strategy",
menu_options=[
@@ -867,6 +883,7 @@ class ZhaConfigFlowHandler(BaseZhaFlow, ConfigFlow, domain=DOMAIN):
radio_type = self._radio_mgr.parse_radio_type(discovery_data["radio_type"])
device_settings = discovery_data["port"]
device_path = device_settings[CONF_DEVICE_PATH]
self._flow_strategy = discovery_data.get("flow_strategy")
await self._set_unique_id_and_update_ignored_flow(
unique_id=f"{name}_{radio_type.name}_{device_path}",
homeassistant/components/zha/radio_manager.py
+4
View File
@@ -28,6 +28,9 @@ from zigpy.exceptions import NetworkNotFormed
from homeassistant import config_entries
from homeassistant.components import usb
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
ZigbeeFlowStrategy,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.service_info.usb import UsbServiceInfo
@@ -74,6 +77,7 @@ HARDWARE_DISCOVERY_SCHEMA = vol.Schema(
vol.Required("name"): str,
vol.Required("port"): DEVICE_SCHEMA,
vol.Required("radio_type"): str,
vol.Optional("flow_strategy"): vol.All(str, vol.Coerce(ZigbeeFlowStrategy)),
}
)
tests/components/homeassistant_hardware/test_config_flow.py
+116 -34
View File
@@ -364,8 +364,8 @@ async def consume_progress_flow(
return result
async def test_config_flow_recommended(hass: HomeAssistant) -> None:
"""Test the config flow with recommended installation type for Zigbee."""
async def test_config_flow_zigbee_recommended(hass: HomeAssistant) -> None:
"""Test flow with recommended Zigbee installation type."""
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
@@ -418,37 +418,28 @@ async def test_config_flow_recommended(hass: HomeAssistant) -> None:
assert zha_flow["context"]["source"] == "hardware"
assert zha_flow["step_id"] == "confirm"
progress_zha_flows = hass.config_entries.flow._async_progress_by_handler(
handler="zha",
match_context=None,
)
@pytest.mark.parametrize(
("zigbee_integration", "zha_flows"),
[
(
"zigbee_integration_zha",
[
{
"context": {
"confirm_only": True,
"source": "hardware",
"title_placeholders": {
"name": "Some Hardware Name",
},
"unique_id": "Some Hardware Name_ezsp_/dev/SomeDevice123",
},
"flow_id": ANY,
"handler": "zha",
"step_id": "confirm",
}
],
),
("zigbee_integration_other", []),
],
)
async def test_config_flow_zigbee_custom(
hass: HomeAssistant,
zigbee_integration: str,
zha_flows: list[ConfigFlowResult],
) -> None:
"""Test the config flow with custom installation type selected for Zigbee."""
assert len(progress_zha_flows) == 1
progress_zha_flow = progress_zha_flows[0]
assert progress_zha_flow.init_data == {
"name": "Some Hardware Name",
"port": {
"path": "/dev/SomeDevice123",
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "ezsp",
"flow_strategy": "recommended",
}
async def test_config_flow_zigbee_custom_zha(hass: HomeAssistant) -> None:
"""Test flow with custom Zigbee installation type and ZHA selected."""
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
@@ -479,7 +470,7 @@ async def test_config_flow_zigbee_custom(
pick_result = await hass.config_entries.flow.async_configure(
pick_result["flow_id"],
user_input={"next_step_id": zigbee_integration},
user_input={"next_step_id": "zigbee_integration_zha"},
)
assert pick_result["type"] is FlowResultType.SHOW_PROGRESS
@@ -503,7 +494,98 @@ async def test_config_flow_zigbee_custom(
# Ensure a ZHA discovery flow has been created
flows = hass.config_entries.flow.async_progress()
assert flows == zha_flows
assert flows == [
{
"context": {
"confirm_only": True,
"source": "hardware",
"title_placeholders": {
"name": "Some Hardware Name",
},
"unique_id": "Some Hardware Name_ezsp_/dev/SomeDevice123",
},
"flow_id": ANY,
"handler": "zha",
"step_id": "confirm",
}
]
progress_zha_flows = hass.config_entries.flow._async_progress_by_handler(
handler="zha",
match_context=None,
)
assert len(progress_zha_flows) == 1
progress_zha_flow = progress_zha_flows[0]
assert progress_zha_flow.init_data == {
"name": "Some Hardware Name",
"port": {
"path": "/dev/SomeDevice123",
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "ezsp",
"flow_strategy": "advanced",
}
async def test_config_flow_zigbee_custom_other(hass: HomeAssistant) -> None:
"""Test flow with custom Zigbee installation type and Other selected."""
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(
probe_app_type=ApplicationType.SPINEL,
flash_app_type=ApplicationType.EZSP,
):
# Pick the menu option: we are flashing the firmware
pick_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert pick_result["type"] is FlowResultType.MENU
assert pick_result["step_id"] == "zigbee_installation_type"
pick_result = await hass.config_entries.flow.async_configure(
pick_result["flow_id"],
user_input={"next_step_id": "zigbee_intent_custom"},
)
assert pick_result["type"] is FlowResultType.MENU
assert pick_result["step_id"] == "zigbee_integration"
pick_result = await hass.config_entries.flow.async_configure(
pick_result["flow_id"],
user_input={"next_step_id": "zigbee_integration_other"},
)
assert pick_result["type"] is FlowResultType.SHOW_PROGRESS
assert pick_result["progress_action"] == "install_firmware"
assert pick_result["step_id"] == "install_zigbee_firmware"
create_result = await consume_progress_flow(
hass,
flow_id=pick_result["flow_id"],
valid_step_ids=("install_zigbee_firmware",),
)
assert create_result["type"] is FlowResultType.CREATE_ENTRY
config_entry = create_result["result"]
assert config_entry.data == {
"firmware": "ezsp",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
}
flows = hass.config_entries.flow.async_progress()
assert flows == []
async def test_config_flow_firmware_index_download_fails_but_not_required(
tests/components/zha/test_config_flow.py
+280 -25
View File
@@ -1180,9 +1180,8 @@ async def test_user_port_config(probe_mock, hass: HomeAssistant) -> None:
assert probe_mock.await_count == 1
@pytest.mark.parametrize("onboarded", [True, False])
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware(onboarded, hass: HomeAssistant) -> None:
async def test_hardware_not_onboarded(hass: HomeAssistant) -> None:
"""Test hardware flow."""
data = {
"name": "Yellow",
@@ -1194,33 +1193,12 @@ async def test_hardware(onboarded, hass: HomeAssistant) -> None:
},
}
with patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=onboarded
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
):
result1 = await hass.config_entries.flow.async_init(
result_create = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_HARDWARE}, data=data
)
if onboarded:
# Confirm discovery
assert result1["type"] is FlowResultType.FORM
assert result1["step_id"] == "confirm"
result2 = await hass.config_entries.flow.async_configure(
result1["flow_id"],
user_input={},
)
assert result2["type"] is FlowResultType.MENU
assert result2["step_id"] == "choose_setup_strategy"
result_create = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={"next_step_id": config_flow.SETUP_STRATEGY_RECOMMENDED},
)
await hass.async_block_till_done()
else:
# No need to confirm
result_create = result1
assert result_create["title"] == "Yellow"
assert result_create["data"] == {
@@ -1233,6 +1211,283 @@ async def test_hardware(onboarded, hass: HomeAssistant) -> None:
}
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware_no_flow_strategy(hass: HomeAssistant) -> None:
"""Test hardware flow."""
data = {
"name": "Yellow",
"radio_type": "efr32",
"port": {
"path": "/dev/ttyAMA1",
"baudrate": 115200,
"flow_control": "hardware",
},
}
with patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=True
):
result1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_HARDWARE}, data=data
)
# Confirm discovery
assert result1["type"] is FlowResultType.FORM
assert result1["step_id"] == "confirm"
result2 = await hass.config_entries.flow.async_configure(
result1["flow_id"],
user_input={},
)
assert result2["type"] is FlowResultType.MENU
assert result2["step_id"] == "choose_setup_strategy"
result_create = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={"next_step_id": config_flow.SETUP_STRATEGY_RECOMMENDED},
)
await hass.async_block_till_done()
assert result_create["title"] == "Yellow"
assert result_create["data"] == {
CONF_DEVICE: {
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: "hardware",
CONF_DEVICE_PATH: "/dev/ttyAMA1",
},
CONF_RADIO_TYPE: "ezsp",
}
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware_flow_strategy_advanced(hass: HomeAssistant) -> None:
"""Test advanced flow strategy for hardware flow."""
data = {
"name": "Yellow",
"radio_type": "efr32",
"port": {
"path": "/dev/ttyAMA1",
"baudrate": 115200,
"flow_control": "hardware",
},
"flow_strategy": "advanced",
}
with patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=True
):
result_hardware = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_HARDWARE}, data=data
)
assert result_hardware["type"] is FlowResultType.FORM
assert result_hardware["step_id"] == "confirm"
confirm_result = await hass.config_entries.flow.async_configure(
result_hardware["flow_id"],
user_input={},
)
assert confirm_result["type"] is FlowResultType.MENU
assert confirm_result["step_id"] == "choose_formation_strategy"
result_create = await hass.config_entries.flow.async_configure(
confirm_result["flow_id"],
user_input={"next_step_id": "form_new_network"},
)
await hass.async_block_till_done()
assert result_create["type"] is FlowResultType.CREATE_ENTRY
assert result_create["title"] == "Yellow"
assert result_create["data"] == {
CONF_DEVICE: {
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: "hardware",
CONF_DEVICE_PATH: "/dev/ttyAMA1",
},
CONF_RADIO_TYPE: "ezsp",
}
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware_flow_strategy_recommended(hass: HomeAssistant) -> None:
"""Test recommended flow strategy for hardware flow."""
data = {
"name": "Yellow",
"radio_type": "efr32",
"port": {
"path": "/dev/ttyAMA1",
"baudrate": 115200,
"flow_control": "hardware",
},
"flow_strategy": "recommended",
}
with patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=True
):
result_hardware = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_HARDWARE}, data=data
)
assert result_hardware["type"] is FlowResultType.FORM
assert result_hardware["step_id"] == "confirm"
result_create = await hass.config_entries.flow.async_configure(
result_hardware["flow_id"],
user_input={},
)
await hass.async_block_till_done()
assert result_create["type"] is FlowResultType.CREATE_ENTRY
assert result_create["title"] == "Yellow"
assert result_create["data"] == {
CONF_DEVICE: {
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: "hardware",
CONF_DEVICE_PATH: "/dev/ttyAMA1",
},
CONF_RADIO_TYPE: "ezsp",
}
@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True))
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware_migration_flow_strategy_advanced(
hass: HomeAssistant,
backup: zigpy.backups.NetworkBackup,
mock_app: AsyncMock,
) -> None:
"""Test advanced flow strategy for hardware migration flow."""
entry = MockConfigEntry(
version=config_flow.ZhaConfigFlowHandler.VERSION,
domain=DOMAIN,
data={
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/ttyUSB0",
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
)
entry.add_to_hass(hass)
data = {
"name": "Yellow",
"radio_type": "efr32",
"port": {
"path": "/dev/ttyAMA1",
"baudrate": 115200,
"flow_control": "hardware",
},
"flow_strategy": "advanced",
}
with (
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=True
),
patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._async_read_backups_from_database",
return_value=[backup],
),
patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager.restore_backup",
) as mock_restore_backup,
patch(
"homeassistant.config_entries.ConfigEntries.async_unload",
return_value=True,
) as mock_async_unload,
):
result_hardware = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_HARDWARE}, data=data
)
assert result_hardware["type"] is FlowResultType.FORM
assert result_hardware["step_id"] == "confirm"
result_confirm = await hass.config_entries.flow.async_configure(
result_hardware["flow_id"], user_input={}
)
assert result_confirm["type"] is FlowResultType.MENU
assert result_confirm["step_id"] == "choose_formation_strategy"
result_formation_strategy = await hass.config_entries.flow.async_configure(
result_confirm["flow_id"],
user_input={"next_step_id": "form_new_network"},
)
await hass.async_block_till_done()
assert result_formation_strategy["type"] is FlowResultType.ABORT
assert result_formation_strategy["reason"] == "reconfigure_successful"
assert mock_async_unload.call_count == 0
assert mock_restore_backup.call_count == 0
@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True))
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware_migration_flow_strategy_recommended(
hass: HomeAssistant,
backup: zigpy.backups.NetworkBackup,
mock_app: AsyncMock,
) -> None:
"""Test recommended flow strategy for hardware migration flow."""
entry = MockConfigEntry(
version=config_flow.ZhaConfigFlowHandler.VERSION,
domain=DOMAIN,
data={
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/ttyUSB0",
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
)
entry.add_to_hass(hass)
data = {
"name": "Yellow",
"radio_type": "efr32",
"port": {
"path": "/dev/ttyAMA1",
"baudrate": 115200,
"flow_control": "hardware",
},
"flow_strategy": "recommended",
}
with (
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=True
),
patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._async_read_backups_from_database",
return_value=[backup],
),
patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager.restore_backup",
) as mock_restore_backup,
patch(
"homeassistant.config_entries.ConfigEntries.async_unload",
return_value=True,
) as mock_async_unload,
):
result_hardware = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_HARDWARE}, data=data
)
assert result_hardware["type"] is FlowResultType.FORM
assert result_hardware["step_id"] == "confirm"
result_confirm = await hass.config_entries.flow.async_configure(
result_hardware["flow_id"], user_input={}
)
assert result_confirm["type"] is FlowResultType.ABORT
assert result_confirm["reason"] == "reconfigure_successful"
assert mock_async_unload.mock_calls == [call(entry.entry_id)]
assert mock_restore_backup.call_count == 1
@pytest.mark.parametrize(
"data", [None, {}, {"radio_type": "best_radio"}, {"radio_type": "efr32"}]
)