ZHA network backup and restore API (#75791)

* Implement WS API endpoints for zigpy backups

* Implement backup restoration

* Display error messages caused by invalid backup JSON

* Indicate to the frontend when a backup is incomplete

* Perform a coordinator backup before HA performs a backup

* Fix `backup.async_post_backup` docstring

* Rename `data` to `backup` in restore command

* Add unit tests for new websocket APIs

* Unit test backup platform

* Move code to overwrite EZSP EUI64 into ZHA

* Include the radio type in the network settings API response

homeassistant/components/zha/api.py

@@ -6,6 +6,8 @@ import logging
 from typing import TYPE_CHECKING, Any, NamedTuple
 
 import voluptuous as vol
+import zigpy.backups
+from zigpy.backups import NetworkBackup
 from zigpy.config.validators import cv_boolean
 from zigpy.types.named import EUI64
 from zigpy.zcl.clusters.security import IasAce
@@ -43,6 +45,7 @@ from .core.const import (
     CLUSTER_COMMANDS_SERVER,
     CLUSTER_TYPE_IN,
     CLUSTER_TYPE_OUT,
+    CONF_RADIO_TYPE,
     CUSTOM_CONFIGURATION,
     DATA_ZHA,
     DATA_ZHA_GATEWAY,
@@ -229,6 +232,15 @@ def _cv_cluster_binding(value: dict[str, Any]) -> ClusterBinding:
     )
 
 
+def _cv_zigpy_network_backup(value: dict[str, Any]) -> zigpy.backups.NetworkBackup:
+    """Transform a zigpy network backup."""
+
+    try:
+        return zigpy.backups.NetworkBackup.from_dict(value)
+    except ValueError as err:
+        raise vol.Invalid(str(err)) from err
+
+
 GROUP_MEMBER_SCHEMA = vol.All(
     vol.Schema(
         {
@@ -302,7 +314,7 @@ async def websocket_permit_devices(
         )
     else:
         await zha_gateway.application_controller.permit(time_s=duration, node=ieee)
-    connection.send_result(msg["id"])
+    connection.send_result(msg[ID])
 
 
 @websocket_api.require_admin
@@ -989,7 +1001,7 @@ async def websocket_get_configuration(
     hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
 ) -> None:
     """Get ZHA configuration."""
-    zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
     import voluptuous_serialize  # pylint: disable=import-outside-toplevel
 
     def custom_serializer(schema: Any) -> Any:
@@ -1047,6 +1059,99 @@ async def websocket_update_zha_configuration(
     connection.send_result(msg[ID], status)
 
 
+@websocket_api.require_admin
+@websocket_api.websocket_command({vol.Required(TYPE): "zha/network/settings"})
+@websocket_api.async_response
+async def websocket_get_network_settings(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Get ZHA network settings."""
+    zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    application_controller = zha_gateway.application_controller
+
+    # Serialize the current network settings
+    backup = NetworkBackup(
+        node_info=application_controller.state.node_info,
+        network_info=application_controller.state.network_info,
+    )
+
+    connection.send_result(
+        msg[ID],
+        {
+            "radio_type": zha_gateway.config_entry.data[CONF_RADIO_TYPE],
+            "settings": backup.as_dict(),
+        },
+    )
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command({vol.Required(TYPE): "zha/network/backups/list"})
+@websocket_api.async_response
+async def websocket_list_network_backups(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Get ZHA network settings."""
+    zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    application_controller = zha_gateway.application_controller
+
+    # Serialize known backups
+    connection.send_result(
+        msg[ID], [backup.as_dict() for backup in application_controller.backups]
+    )
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command({vol.Required(TYPE): "zha/network/backups/create"})
+@websocket_api.async_response
+async def websocket_create_network_backup(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Create a ZHA network backup."""
+    zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    application_controller = zha_gateway.application_controller
+
+    # This can take 5-30s
+    backup = await application_controller.backups.create_backup(load_devices=True)
+    connection.send_result(
+        msg[ID],
+        {
+            "backup": backup.as_dict(),
+            "is_complete": backup.is_complete(),
+        },
+    )
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "zha/network/backups/restore",
+        vol.Required("backup"): _cv_zigpy_network_backup,
+        vol.Optional("ezsp_force_write_eui64", default=False): cv.boolean,
+    }
+)
+@websocket_api.async_response
+async def websocket_restore_network_backup(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Restore a ZHA network backup."""
+    zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    application_controller = zha_gateway.application_controller
+    backup = msg["backup"]
+
+    if msg["ezsp_force_write_eui64"]:
+        backup.network_info.stack_specific.setdefault("ezsp", {})[
+            "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it"
+        ] = True
+
+    # This can take 30-40s
+    try:
+        await application_controller.backups.restore_backup(backup)
+    except ValueError as err:
+        connection.send_error(msg[ID], websocket_api.const.ERR_INVALID_FORMAT, str(err))
+    else:
+        connection.send_result(msg[ID])
+
+
 @callback
 def async_load_api(hass: HomeAssistant) -> None:
     """Set up the web socket API."""
@@ -1356,6 +1461,10 @@ def async_load_api(hass: HomeAssistant) -> None:
     websocket_api.async_register_command(hass, websocket_update_topology)
     websocket_api.async_register_command(hass, websocket_get_configuration)
     websocket_api.async_register_command(hass, websocket_update_zha_configuration)
+    websocket_api.async_register_command(hass, websocket_get_network_settings)
+    websocket_api.async_register_command(hass, websocket_list_network_backups)
+    websocket_api.async_register_command(hass, websocket_create_network_backup)
+    websocket_api.async_register_command(hass, websocket_restore_network_backup)
 
 
 @callback

homeassistant/components/zha/backup.py

@@ -0,0 +1,21 @@
+"""Backup platform for the ZHA integration."""
+import logging
+
+from homeassistant.core import HomeAssistant
+
+from .core import ZHAGateway
+from .core.const import DATA_ZHA, DATA_ZHA_GATEWAY
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_pre_backup(hass: HomeAssistant) -> None:
+    """Perform operations before a backup starts."""
+    _LOGGER.debug("Performing coordinator backup")
+
+    zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    await zha_gateway.application_controller.backups.create_backup(load_devices=True)
+
+
+async def async_post_backup(hass: HomeAssistant) -> None:
+    """Perform operations after a backup finishes."""

tests/components/zha/conftest.py

@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
 import pytest
 import zigpy
 from zigpy.application import ControllerApplication
+import zigpy.backups
 import zigpy.config
 from zigpy.const import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
 import zigpy.device
@@ -54,7 +55,16 @@ def zigpy_app_controller():
     app.ieee.return_value = zigpy.types.EUI64.convert("00:15:8d:00:02:32:4f:32")
     type(app).nwk = PropertyMock(return_value=zigpy.types.NWK(0x0000))
     type(app).devices = PropertyMock(return_value={})
-    type(app).state = PropertyMock(return_value=State())
+    type(app).backups = zigpy.backups.BackupManager(app)
+
+    state = State()
+    state.node_info.ieee = app.ieee.return_value
+    state.network_info.extended_pan_id = app.ieee.return_value
+    state.network_info.pan_id = 0x1234
+    state.network_info.channel = 15
+    state.network_info.network_key.key = zigpy.types.KeyData(range(16))
+    type(app).state = PropertyMock(return_value=state)
+
     return app
 
 

tests/components/zha/test_api.py

@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, patch
 
 import pytest
 import voluptuous as vol
+import zigpy.backups
 import zigpy.profiles.zha
 import zigpy.types
 import zigpy.zcl.clusters.general as general
@@ -620,3 +621,125 @@ async def test_ws_permit_ha12(app_controller, zha_client, params, duration, node
     assert app_controller.permit.await_args[1]["time_s"] == duration
     assert app_controller.permit.await_args[1]["node"] == node
     assert app_controller.permit_with_key.call_count == 0
+
+
+async def test_get_network_settings(app_controller, zha_client):
+    """Test current network settings are returned."""
+
+    await app_controller.backups.create_backup()
+
+    await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/settings"})
+    msg = await zha_client.receive_json()
+
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert msg["success"]
+    assert "radio_type" in msg["result"]
+    assert "network_info" in msg["result"]["settings"]
+
+
+async def test_list_network_backups(app_controller, zha_client):
+    """Test backups are serialized."""
+
+    await app_controller.backups.create_backup()
+
+    await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/backups/list"})
+    msg = await zha_client.receive_json()
+
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert msg["success"]
+    assert "network_info" in msg["result"][0]
+
+
+async def test_create_network_backup(app_controller, zha_client):
+    """Test creating backup."""
+
+    assert not app_controller.backups.backups
+    await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/backups/create"})
+    msg = await zha_client.receive_json()
+    assert len(app_controller.backups.backups) == 1
+
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert msg["success"]
+    assert "backup" in msg["result"] and "is_complete" in msg["result"]
+
+
+async def test_restore_network_backup_success(app_controller, zha_client):
+    """Test successfully restoring a backup."""
+
+    backup = zigpy.backups.NetworkBackup()
+
+    with patch.object(app_controller.backups, "restore_backup", new=AsyncMock()) as p:
+        await zha_client.send_json(
+            {
+                ID: 6,
+                TYPE: f"{DOMAIN}/network/backups/restore",
+                "backup": backup.as_dict(),
+            }
+        )
+        msg = await zha_client.receive_json()
+
+    p.assert_called_once_with(backup)
+    assert "ezsp" not in backup.network_info.stack_specific
+
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert msg["success"]
+
+
+async def test_restore_network_backup_force_write_eui64(app_controller, zha_client):
+    """Test successfully restoring a backup."""
+
+    backup = zigpy.backups.NetworkBackup()
+
+    with patch.object(app_controller.backups, "restore_backup", new=AsyncMock()) as p:
+        await zha_client.send_json(
+            {
+                ID: 6,
+                TYPE: f"{DOMAIN}/network/backups/restore",
+                "backup": backup.as_dict(),
+                "ezsp_force_write_eui64": True,
+            }
+        )
+        msg = await zha_client.receive_json()
+
+    # EUI64 will be overwritten
+    p.assert_called_once_with(
+        backup.replace(
+            network_info=backup.network_info.replace(
+                stack_specific={
+                    "ezsp": {
+                        "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it": True
+                    }
+                }
+            )
+        )
+    )
+
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert msg["success"]
+
+
+@patch("zigpy.backups.NetworkBackup.from_dict", new=lambda v: v)
+async def test_restore_network_backup_failure(app_controller, zha_client):
+    """Test successfully restoring a backup."""
+
+    with patch.object(
+        app_controller.backups,
+        "restore_backup",
+        new=AsyncMock(side_effect=ValueError("Restore failed")),
+    ) as p:
+        await zha_client.send_json(
+            {ID: 6, TYPE: f"{DOMAIN}/network/backups/restore", "backup": "a backup"}
+        )
+        msg = await zha_client.receive_json()
+
+    p.assert_called_once_with("a backup")
+
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert not msg["success"]
+    assert msg["error"]["code"] == const.ERR_INVALID_FORMAT

tests/components/zha/test_backup.py

@@ -0,0 +1,20 @@
+"""Unit tests for ZHA backup platform."""
+
+from unittest.mock import AsyncMock, patch
+
+from homeassistant.components.zha.backup import async_post_backup, async_pre_backup
+
+
+async def test_pre_backup(hass, setup_zha):
+    """Test backup creation when `async_pre_backup` is called."""
+    with patch("zigpy.backups.BackupManager.create_backup", AsyncMock()) as backup_mock:
+        await setup_zha()
+        await async_pre_backup(hass)
+
+    backup_mock.assert_called_once_with(load_devices=True)
+
+
+async def test_post_backup(hass, setup_zha):
+    """Test no-op `async_post_backup`."""
+    await setup_zha()
+    await async_post_backup(hass)