From 1ee0c907b08958c222ced4a257e9d1913a2ff856 Mon Sep 17 00:00:00 2001 From: Erik Montnemery Date: Tue, 8 Aug 2023 11:38:01 +0200 Subject: [PATCH] Improve OTBR factory reset (#98017) Co-authored-by: Stefan Agner --- homeassistant/components/otbr/util.py | 14 ++++ .../components/otbr/websocket_api.py | 4 +- tests/components/otbr/test_util.py | 77 +++++++++++++++++++ tests/components/otbr/test_websocket_api.py | 18 ++--- 4 files changed, 102 insertions(+), 11 deletions(-) diff --git a/homeassistant/components/otbr/util.py b/homeassistant/components/otbr/util.py index 2d6217ea585..4d6efb9a9f0 100644 --- a/homeassistant/components/otbr/util.py +++ b/homeassistant/components/otbr/util.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import dataclasses from functools import wraps +import logging from typing import Any, Concatenate, ParamSpec, TypeVar, cast import python_otbr_api @@ -27,6 +28,8 @@ from .const import DOMAIN _R = TypeVar("_R") _P = ParamSpec("_P") +_LOGGER = logging.getLogger(__name__) + INFO_URL_SKY_CONNECT = ( "https://skyconnect.home-assistant.io/multiprotocol-channel-missmatch" ) @@ -68,6 +71,17 @@ class OTBRData: api: python_otbr_api.OTBR entry_id: str + @_handle_otbr_error + async def factory_reset(self) -> None: + """Reset the router.""" + try: + await self.api.factory_reset() + except python_otbr_api.FactoryResetNotSupportedError: + _LOGGER.warning( + "OTBR does not support factory reset, attempting to delete dataset" + ) + await self.delete_active_dataset() + @_handle_otbr_error async def set_enabled(self, enabled: bool) -> None: """Enable or disable the router.""" diff --git a/homeassistant/components/otbr/websocket_api.py b/homeassistant/components/otbr/websocket_api.py index 3b631057529..9b57cd8ebd1 100644 --- a/homeassistant/components/otbr/websocket_api.py +++ b/homeassistant/components/otbr/websocket_api.py @@ -88,9 +88,9 @@ async def websocket_create_network( return try: - await data.delete_active_dataset() + await data.factory_reset() except HomeAssistantError as exc: - connection.send_error(msg["id"], "delete_active_dataset_failed", str(exc)) + connection.send_error(msg["id"], "factory_reset_failed", str(exc)) return try: diff --git a/tests/components/otbr/test_util.py b/tests/components/otbr/test_util.py index f8ed79b91ee..171a607d200 100644 --- a/tests/components/otbr/test_util.py +++ b/tests/components/otbr/test_util.py @@ -1,7 +1,12 @@ """Test OTBR Utility functions.""" +from unittest.mock import patch + +import pytest +import python_otbr_api from homeassistant.components import otbr from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError OTBR_MULTIPAN_URL = "http://core-silabs-multiprotocol:8081" OTBR_NON_MULTIPAN_URL = "/dev/ttyAMA1" @@ -23,3 +28,75 @@ async def test_get_allowed_channel( # OTBR no multipan + multipan using channel 15 -> no restriction multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 assert await otbr.util.get_allowed_channel(hass, OTBR_NON_MULTIPAN_URL) is None + + +async def test_factory_reset(hass: HomeAssistant, otbr_config_entry_multipan) -> None: + """Test factory_reset.""" + data: otbr.OTBRData = hass.data[otbr.DOMAIN] + + with patch("python_otbr_api.OTBR.factory_reset") as factory_reset_mock, patch( + "python_otbr_api.OTBR.delete_active_dataset" + ) as delete_active_dataset_mock: + await data.factory_reset() + + delete_active_dataset_mock.assert_not_called() + factory_reset_mock.assert_called_once_with() + + +async def test_factory_reset_not_supported( + hass: HomeAssistant, otbr_config_entry_multipan +) -> None: + """Test factory_reset.""" + data: otbr.OTBRData = hass.data[otbr.DOMAIN] + + with patch( + "python_otbr_api.OTBR.factory_reset", + side_effect=python_otbr_api.FactoryResetNotSupportedError, + ) as factory_reset_mock, patch( + "python_otbr_api.OTBR.delete_active_dataset" + ) as delete_active_dataset_mock: + await data.factory_reset() + + delete_active_dataset_mock.assert_called_once_with() + factory_reset_mock.assert_called_once_with() + + +async def test_factory_reset_error_1( + hass: HomeAssistant, otbr_config_entry_multipan +) -> None: + """Test factory_reset.""" + data: otbr.OTBRData = hass.data[otbr.DOMAIN] + + with patch( + "python_otbr_api.OTBR.factory_reset", + side_effect=python_otbr_api.OTBRError, + ) as factory_reset_mock, patch( + "python_otbr_api.OTBR.delete_active_dataset" + ) as delete_active_dataset_mock, pytest.raises( + HomeAssistantError + ): + await data.factory_reset() + + delete_active_dataset_mock.assert_not_called() + factory_reset_mock.assert_called_once_with() + + +async def test_factory_reset_error_2( + hass: HomeAssistant, otbr_config_entry_multipan +) -> None: + """Test factory_reset.""" + data: otbr.OTBRData = hass.data[otbr.DOMAIN] + + with patch( + "python_otbr_api.OTBR.factory_reset", + side_effect=python_otbr_api.FactoryResetNotSupportedError, + ) as factory_reset_mock, patch( + "python_otbr_api.OTBR.delete_active_dataset", + side_effect=python_otbr_api.OTBRError, + ) as delete_active_dataset_mock, pytest.raises( + HomeAssistantError + ): + await data.factory_reset() + + delete_active_dataset_mock.assert_called_once_with() + factory_reset_mock.assert_called_once_with() diff --git a/tests/components/otbr/test_websocket_api.py b/tests/components/otbr/test_websocket_api.py index b5dd7aa62c4..d62213ce78b 100644 --- a/tests/components/otbr/test_websocket_api.py +++ b/tests/components/otbr/test_websocket_api.py @@ -87,8 +87,8 @@ async def test_create_network( with patch( "python_otbr_api.OTBR.create_active_dataset" ) as create_dataset_mock, patch( - "python_otbr_api.OTBR.delete_active_dataset" - ) as delete_dataset_mock, patch( + "python_otbr_api.OTBR.factory_reset" + ) as factory_reset_mock, patch( "python_otbr_api.OTBR.set_enabled" ) as set_enabled_mock, patch( "python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=DATASET_CH16 @@ -104,7 +104,7 @@ async def test_create_network( create_dataset_mock.assert_called_once_with( python_otbr_api.models.ActiveDataSet(channel=15, network_name="home-assistant") ) - delete_dataset_mock.assert_called_once_with() + factory_reset_mock.assert_called_once_with() assert len(set_enabled_mock.mock_calls) == 2 assert set_enabled_mock.mock_calls[0][1][0] is False assert set_enabled_mock.mock_calls[1][1][0] is True @@ -157,7 +157,7 @@ async def test_create_network_fails_2( ), patch( "python_otbr_api.OTBR.create_active_dataset", side_effect=python_otbr_api.OTBRError, - ), patch("python_otbr_api.OTBR.delete_active_dataset"): + ), patch("python_otbr_api.OTBR.factory_reset"): await websocket_client.send_json_auto_id({"type": "otbr/create_network"}) msg = await websocket_client.receive_json() @@ -178,7 +178,7 @@ async def test_create_network_fails_3( ), patch( "python_otbr_api.OTBR.create_active_dataset", ), patch( - "python_otbr_api.OTBR.delete_active_dataset" + "python_otbr_api.OTBR.factory_reset" ): await websocket_client.send_json_auto_id({"type": "otbr/create_network"}) msg = await websocket_client.receive_json() @@ -200,7 +200,7 @@ async def test_create_network_fails_4( "python_otbr_api.OTBR.get_active_dataset_tlvs", side_effect=python_otbr_api.OTBRError, ), patch( - "python_otbr_api.OTBR.delete_active_dataset" + "python_otbr_api.OTBR.factory_reset" ): await websocket_client.send_json_auto_id({"type": "otbr/create_network"}) msg = await websocket_client.receive_json() @@ -219,7 +219,7 @@ async def test_create_network_fails_5( with patch("python_otbr_api.OTBR.set_enabled"), patch( "python_otbr_api.OTBR.create_active_dataset" ), patch("python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=None), patch( - "python_otbr_api.OTBR.delete_active_dataset" + "python_otbr_api.OTBR.factory_reset" ): await websocket_client.send_json_auto_id({"type": "otbr/create_network"}) msg = await websocket_client.receive_json() @@ -238,14 +238,14 @@ async def test_create_network_fails_6( with patch("python_otbr_api.OTBR.set_enabled"), patch( "python_otbr_api.OTBR.create_active_dataset" ), patch("python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=None), patch( - "python_otbr_api.OTBR.delete_active_dataset", + "python_otbr_api.OTBR.factory_reset", side_effect=python_otbr_api.OTBRError, ): await websocket_client.send_json_auto_id({"type": "otbr/create_network"}) msg = await websocket_client.receive_json() assert not msg["success"] - assert msg["error"]["code"] == "delete_active_dataset_failed" + assert msg["error"]["code"] == "factory_reset_failed" async def test_set_network(