Improve OTBR factory reset (#98017)

Co-authored-by: Stefan Agner <stefan@agner.ch>

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."""

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:

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()

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"
+        "python_otbr_api.OTBR.factory_reset"
-    ) as delete_dataset_mock, patch(
+    ) 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(