diff --git a/homeassistant/components/zha/api.py b/homeassistant/components/zha/api.py index 89d360577d4..40996be3248 100644 --- a/homeassistant/components/zha/api.py +++ b/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 diff --git a/homeassistant/components/zha/backup.py b/homeassistant/components/zha/backup.py new file mode 100644 index 00000000000..89d5294e1c4 --- /dev/null +++ b/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.""" diff --git a/tests/components/zha/conftest.py b/tests/components/zha/conftest.py index b1041a3e2a3..27155e16cc7 100644 --- a/tests/components/zha/conftest.py +++ b/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 diff --git a/tests/components/zha/test_api.py b/tests/components/zha/test_api.py index 08766bc74ac..b25bffebec7 100644 --- a/tests/components/zha/test_api.py +++ b/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 diff --git a/tests/components/zha/test_backup.py b/tests/components/zha/test_backup.py new file mode 100644 index 00000000000..aea50cf0923 --- /dev/null +++ b/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)