jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-15 17:27:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

ZHA radio migration: reset the old adapter (#79663)

Browse Source
This commit is contained in:
puddly 2022-10-06 14:02:24 -04:00 committed by GitHub
parent 0a59d37e62
commit 2dab9073fe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 219 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
homeassistant/components/zha/config_flow.py
View File

@ -1,6 +1,7 @@
"""Config flow for ZHA."""
from __future__ import annotations
import asyncio
import collections
import contextlib
import copy
@ -65,8 +66,16 @@ FORMATION_UPLOAD_MANUAL_BACKUP = "upload_manual_backup"
CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup"
OVERWRITE_COORDINATOR_IEEE = "overwrite_coordinator_ieee"
OPTIONS_INTENT_MIGRATE = "intent_migrate"
OPTIONS_INTENT_RECONFIGURE = "intent_reconfigure"
UPLOADED_BACKUP_FILE = "uploaded_backup_file"
DEFAULT_ZHA_ZEROCONF_PORT = 6638
ESPHOME_API_PORT = 6053
CONNECT_DELAY_S = 1.0
_LOGGER = logging.getLogger(__name__)
@ -159,6 +168,7 @@ class BaseZhaFlow(FlowHandler):
yield app
finally:
await app.disconnect()
await asyncio.sleep(CONNECT_DELAY_S)
async def _restore_backup(
self, backup: zigpy.backups.NetworkBackup, **kwargs: Any
@ -628,14 +638,21 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN
# Hostname is format: livingroom.local.
local_name = discovery_info.hostname[:-1]
radio_type = discovery_info.properties.get("radio_type") or local_name
port = discovery_info.port or DEFAULT_ZHA_ZEROCONF_PORT
# Fix incorrect port for older TubesZB devices
if "tube" in local_name and port == ESPHOME_API_PORT:
port = DEFAULT_ZHA_ZEROCONF_PORT
if "radio_type" in discovery_info.properties:
self._radio_type = RadioType[discovery_info.properties["radio_type"]]
elif "efr32" in local_name:
self._radio_type = RadioType.ezsp
else:
self._radio_type = RadioType.znp
node_name = local_name[: -len(".local")]
host = discovery_info.host
port = discovery_info.port
if local_name.startswith("tube") or "efr32" in local_name:
# This is hard coded to work with legacy devices
port = 6638
device_path = f"socket://{host}:{port}"
device_path = f"socket://{discovery_info.host}:{port}"
if current_entry := await self.async_set_unique_id(node_name):
self._abort_if_unique_id_configured(
@ -651,13 +668,6 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN
self._title = device_path
self._device_path = device_path
if "efr32" in radio_type:
self._radio_type = RadioType.ezsp
elif "zigate" in radio_type:
self._radio_type = RadioType.zigate
else:
self._radio_type = RadioType.znp
return await self.async_step_confirm()
async def async_step_hardware(
@ -720,10 +730,54 @@ class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow):
# ZHA is not running
pass
return await self.async_step_choose_serial_port()
return await self.async_step_prompt_migrate_or_reconfigure()
return self.async_show_form(step_id="init")
async def async_step_prompt_migrate_or_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm if we are migrating adapters or just re-configuring."""
return self.async_show_menu(
step_id="prompt_migrate_or_reconfigure",
menu_options=[
OPTIONS_INTENT_RECONFIGURE,
OPTIONS_INTENT_MIGRATE,
],
)
async def async_step_intent_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Virtual step for when the user is reconfiguring the integration."""
return await self.async_step_choose_serial_port()
async def async_step_intent_migrate(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm the user wants to reset their current radio."""
if user_input is not None:
# Reset the current adapter
async with self._connect_zigpy_app() as app:
await app.reset_network_info()
return await self.async_step_instruct_unplug()
return self.async_show_form(step_id="intent_migrate")
async def async_step_instruct_unplug(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Instruct the user to unplug the current radio, if possible."""
if user_input is not None:
# Now that the old radio is gone, we can scan for serial ports again
return await self.async_step_choose_serial_port()
return self.async_show_form(step_id="instruct_unplug")
async def _async_create_radio_entity(self):
"""Re-implementation of the base flow's final step to update the config."""
device_settings = self._device_settings.copy()

16
homeassistant/components/zha/strings.json
View File

@ -76,6 +76,22 @@
"title": "Reconfigure ZHA",
"description": "ZHA will be stopped. Do you wish to continue?"
},
"prompt_migrate_or_reconfigure": {
"title": "Migrate or re-configure",
"description": "Are you migrating to a new radio or re-configuring the current radio?",
"menu_options": {
"intent_migrate": "Migrate to a new radio",
"intent_reconfigure": "Re-configure the current radio"
}
},
"intent_migrate": {
"title": "Migrate to a new radio",
"description": "Your old radio will be factory reset. If you are using a combined Z-Wave and Zigbee adapter like the HUSBZB-1, this will only reset the Zigbee portion.\n\nDo you wish to continue?"
},
"instruct_unplug": {
"title": "Unplug your old radio",
"description": "Your old radio has been reset. If the hardware is no longer needed, you can now unplug it."
},
"choose_serial_port": {
"title": "[%key:component::zha::config::step::choose_serial_port::title%]",
"data": {

39
homeassistant/components/zha/translations/en.json
View File

@ -64,35 +64,12 @@
"description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.",
"title": "Overwrite Radio IEEE Address"
},
"pick_radio": {
"data": {
"radio_type": "Radio Type"
},
"description": "Pick a type of your Zigbee radio",
"title": "Radio Type"
},
"port_config": {
"data": {
"baudrate": "port speed",
"flow_control": "data flow control",
"path": "Serial device path"
},
"description": "Enter port specific settings",
"title": "Settings"
},
"upload_manual_backup": {
"data": {
"uploaded_backup_file": "Upload a file"
},
"description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.",
"title": "Upload a Manual Backup"
},
"user": {
"data": {
"path": "Serial Device Path"
},
"description": "Select serial port for Zigbee radio",
"title": "ZHA"
}
}
},
@ -212,6 +189,14 @@
"description": "ZHA will be stopped. Do you wish to continue?",
"title": "Reconfigure ZHA"
},
"instruct_unplug": {
"description": "Your old radio has been reset. If the hardware is no longer needed, you can now unplug it.",
"title": "Unplug your old radio"
},
"intent_migrate": {
"description": "Your old radio will be factory reset. If you are using a combined Z-Wave and Zigbee adapter like the HUSBZB-1, this will only reset the Zigbee portion.\n\nDo you wish to continue?",
"title": "Migrate to a new radio"
},
"manual_pick_radio_type": {
"data": {
"radio_type": "Radio Type"
@ -235,6 +220,14 @@
"description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.",
"title": "Overwrite Radio IEEE Address"
},
"prompt_migrate_or_reconfigure": {
"description": "Are you migrating to a new radio or re-configuring the current radio?",
"menu_options": {
"intent_migrate": "Migrate to a new radio",
"intent_reconfigure": "Re-configure the current radio"
},
"title": "Migrate or re-configure"
},
"upload_manual_backup": {
"data": {
"uploaded_backup_file": "Upload a file"

148
tests/components/zha/test_config_flow.py
View File

@ -46,6 +46,13 @@ def disable_platform_only():
yield
@pytest.fixture(autouse=True)
def reduce_reconnect_timeout():
"""Reduces reconnect timeout to speed up tests."""
with patch("homeassistant.components.zha.config_flow.CONNECT_DELAY_S", 0.01):
yield
@pytest.fixture(autouse=True)
def mock_app():
"""Mock zigpy app interface."""
@ -230,10 +237,10 @@ async def test_efr32_via_zeroconf(hass):
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "socket://192.168.1.200:6638"
assert result3["title"] == "socket://192.168.1.200:1234"
assert result3["data"] == {
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://192.168.1.200:6638",
CONF_DEVICE_PATH: "socket://192.168.1.200:1234",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: "software",
},
@ -1476,21 +1483,28 @@ async def test_options_flow_defaults(async_setup_entry, async_unload_effect, has
# Unload it ourselves
entry.state = config_entries.ConfigEntryState.NOT_LOADED
# Reconfigure ZHA
assert result1["step_id"] == "prompt_migrate_or_reconfigure"
result2 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={"next_step_id": config_flow.OPTIONS_INTENT_RECONFIGURE},
)
# Current path is the default
assert result1["step_id"] == "choose_serial_port"
assert "/dev/ttyUSB0" in result1["data_schema"]({})[CONF_DEVICE_PATH]
assert result2["step_id"] == "choose_serial_port"
assert "/dev/ttyUSB0" in result2["data_schema"]({})[CONF_DEVICE_PATH]
# Autoprobing fails, we have to manually choose the radio type
result2 = await hass.config_entries.options.async_configure(
result3 = await hass.config_entries.options.async_configure(
flow["flow_id"], user_input={}
)
# Current radio type is the default
assert result2["step_id"] == "manual_pick_radio_type"
assert result2["data_schema"]({})[CONF_RADIO_TYPE] == RadioType.znp.description
assert result3["step_id"] == "manual_pick_radio_type"
assert result3["data_schema"]({})[CONF_RADIO_TYPE] == RadioType.znp.description
# Continue on to port settings
result3 = await hass.config_entries.options.async_configure(
result4 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={
CONF_RADIO_TYPE: RadioType.znp.description,
@ -1498,12 +1512,12 @@ async def test_options_flow_defaults(async_setup_entry, async_unload_effect, has
)
# The defaults match our current settings
assert result3["step_id"] == "manual_port_config"
assert result3["data_schema"]({}) == entry.data[CONF_DEVICE]
assert result4["step_id"] == "manual_port_config"
assert result4["data_schema"]({}) == entry.data[CONF_DEVICE]
with patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)):
# Change the serial port path
result4 = await hass.config_entries.options.async_configure(
result5 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={
# Change everything
@ -1514,18 +1528,18 @@ async def test_options_flow_defaults(async_setup_entry, async_unload_effect, has
)
# The radio has been detected, we can move on to creating the config entry
assert result4["step_id"] == "choose_formation_strategy"
assert result5["step_id"] == "choose_formation_strategy"
async_setup_entry.assert_not_called()
result5 = await hass.config_entries.options.async_configure(
result6 = await hass.config_entries.options.async_configure(
result1["flow_id"],
user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS},
)
await hass.async_block_till_done()
assert result5["type"] == FlowResultType.CREATE_ENTRY
assert result5["data"] == {}
assert result6["type"] == FlowResultType.CREATE_ENTRY
assert result6["data"] == {}
# The updated entry contains correct settings
assert entry.data == {
@ -1581,33 +1595,39 @@ async def test_options_flow_defaults_socket(hass):
flow["flow_id"], user_input={}
)
# Radio path must be manually entered
assert result1["step_id"] == "choose_serial_port"
assert result1["data_schema"]({})[CONF_DEVICE_PATH] == config_flow.CONF_MANUAL_PATH
assert result1["step_id"] == "prompt_migrate_or_reconfigure"
result2 = await hass.config_entries.options.async_configure(
flow["flow_id"], user_input={}
flow["flow_id"],
user_input={"next_step_id": config_flow.OPTIONS_INTENT_RECONFIGURE},
)
# Current radio type is the default
assert result2["step_id"] == "manual_pick_radio_type"
assert result2["data_schema"]({})[CONF_RADIO_TYPE] == RadioType.znp.description
# Radio path must be manually entered
assert result2["step_id"] == "choose_serial_port"
assert result2["data_schema"]({})[CONF_DEVICE_PATH] == config_flow.CONF_MANUAL_PATH
# Continue on to port settings
result3 = await hass.config_entries.options.async_configure(
flow["flow_id"], user_input={}
)
# Current radio type is the default
assert result3["step_id"] == "manual_pick_radio_type"
assert result3["data_schema"]({})[CONF_RADIO_TYPE] == RadioType.znp.description
# Continue on to port settings
result4 = await hass.config_entries.options.async_configure(
flow["flow_id"], user_input={}
)
# The defaults match our current settings
assert result3["step_id"] == "manual_port_config"
assert result3["data_schema"]({}) == entry.data[CONF_DEVICE]
assert result4["step_id"] == "manual_port_config"
assert result4["data_schema"]({}) == entry.data[CONF_DEVICE]
with patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)):
result4 = await hass.config_entries.options.async_configure(
result5 = await hass.config_entries.options.async_configure(
flow["flow_id"], user_input={}
)
assert result4["step_id"] == "choose_formation_strategy"
assert result5["step_id"] == "choose_formation_strategy"
@patch("homeassistant.components.zha.async_setup_entry", return_value=True)
@ -1643,14 +1663,82 @@ async def test_options_flow_restarts_running_zha_if_cancelled(async_setup_entry,
entry.state = config_entries.ConfigEntryState.NOT_LOADED
assert result1["step_id"] == "prompt_migrate_or_reconfigure"
result2 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={"next_step_id": config_flow.OPTIONS_INTENT_RECONFIGURE},
)
# Radio path must be manually entered
assert result1["step_id"] == "choose_serial_port"
assert result2["step_id"] == "choose_serial_port"
async_setup_entry.reset_mock()
# Abort the flow
hass.config_entries.options.async_abort(result1["flow_id"])
hass.config_entries.options.async_abort(result2["flow_id"])
await hass.async_block_till_done()
# ZHA was set up once more
async_setup_entry.assert_called_once_with(hass, entry)
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_options_flow_migration_reset_old_adapter(hass, mock_app):
"""Test options flow for migrating from an old radio."""
entry = MockConfigEntry(
version=config_flow.ZhaConfigFlowHandler.VERSION,
domain=DOMAIN,
data={
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/serial/by-id/old_radio",
CONF_BAUDRATE: 12345,
CONF_FLOWCONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
flow = await hass.config_entries.options.async_init(entry.entry_id)
# ZHA gets unloaded
with patch(
"homeassistant.config_entries.ConfigEntries.async_unload", return_value=True
):
result1 = await hass.config_entries.options.async_configure(
flow["flow_id"], user_input={}
)
entry.state = config_entries.ConfigEntryState.NOT_LOADED
assert result1["step_id"] == "prompt_migrate_or_reconfigure"
result2 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={"next_step_id": config_flow.OPTIONS_INTENT_MIGRATE},
)
# User must explicitly approve radio reset
assert result2["step_id"] == "intent_migrate"
mock_app.reset_network_info = AsyncMock()
result3 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={},
)
mock_app.reset_network_info.assert_awaited_once()
# Now we can unplug the old radio
assert result3["step_id"] == "instruct_unplug"
# And move on to choosing the new radio
result4 = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={},
)
assert result4["step_id"] == "choose_serial_port"