Add WS API for creating a Thread network (#88830)

* Add WS API for creating a Thread network

* Add tests

homeassistant/components/otbr/__init__.py

@@ -46,11 +46,23 @@ class OTBRData:
     url: str
     api: python_otbr_api.OTBR
 
+    @_handle_otbr_error
+    async def set_enabled(self, enabled: bool) -> None:
+        """Enable or disable the router."""
+        return await self.api.set_enabled(enabled)
+
     @_handle_otbr_error
     async def get_active_dataset_tlvs(self) -> bytes | None:
         """Get current active operational dataset in TLVS format, or None."""
         return await self.api.get_active_dataset_tlvs()
 
+    @_handle_otbr_error
+    async def create_active_dataset(
+        self, dataset: python_otbr_api.OperationalDataSet
+    ) -> None:
+        """Create an active operational dataset."""
+        return await self.api.create_active_dataset(dataset)
+
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the Open Thread Border Router component."""

homeassistant/components/otbr/websocket_api.py

@@ -1,6 +1,8 @@
 """Websocket API for OTBR."""
 from typing import TYPE_CHECKING
 
+import python_otbr_api
+
 from homeassistant.components.websocket_api import (
     ActiveConnection,
     async_register_command,
@@ -20,6 +22,7 @@ if TYPE_CHECKING:
 def async_setup(hass: HomeAssistant) -> None:
     """Set up the OTBR Websocket API."""
     async_register_command(hass, websocket_info)
+    async_register_command(hass, websocket_create_network)
 
 
 @websocket_command(
@@ -51,3 +54,42 @@ async def websocket_info(
             "active_dataset_tlvs": dataset.hex() if dataset else None,
         },
     )
+
+
+@websocket_command(
+    {
+        "type": "otbr/create_network",
+    }
+)
+@async_response
+async def websocket_create_network(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict
+) -> None:
+    """Create a new Thread network."""
+    if DOMAIN not in hass.data:
+        connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
+        return
+
+    data: OTBRData = hass.data[DOMAIN]
+
+    try:
+        await data.set_enabled(False)
+    except HomeAssistantError as exc:
+        connection.send_error(msg["id"], "set_enabled_failed", str(exc))
+        return
+
+    try:
+        await data.create_active_dataset(
+            python_otbr_api.OperationalDataSet(network_name="home-assistant")
+        )
+    except HomeAssistantError as exc:
+        connection.send_error(msg["id"], "create_active_dataset_failed", str(exc))
+        return
+
+    try:
+        await data.set_enabled(True)
+    except HomeAssistantError as exc:
+        connection.send_error(msg["id"], "set_enabled_failed", str(exc))
+        return
+
+    connection.send_result(msg["id"])

tests/components/otbr/test_websocket_api.py

@@ -96,3 +96,139 @@ async def test_get_info_fetch_fails(
     assert msg["id"] == 5
     assert not msg["success"]
     assert msg["error"]["code"] == "get_dataset_failed"
+
+
+async def test_create_network(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test create network."""
+
+    with patch(
+        "python_otbr_api.OTBR.create_active_dataset"
+    ) as create_dataset_mock, patch(
+        "python_otbr_api.OTBR.set_enabled"
+    ) as set_enabled_mock:
+        await websocket_client.send_json(
+            {
+                "id": 5,
+                "type": "otbr/create_network",
+            }
+        )
+
+        msg = await websocket_client.receive_json()
+        assert msg["id"] == 5
+        assert msg["success"]
+        assert msg["result"] is None
+
+    create_dataset_mock.assert_called_once_with(
+        python_otbr_api.models.OperationalDataSet(network_name="home-assistant")
+    )
+    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
+
+
+async def test_create_network_no_entry(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test create network."""
+    await async_setup_component(hass, "otbr", {})
+    websocket_client = await hass_ws_client(hass)
+    await websocket_client.send_json(
+        {
+            "id": 5,
+            "type": "otbr/create_network",
+        }
+    )
+
+    msg = await websocket_client.receive_json()
+    assert msg["id"] == 5
+    assert not msg["success"]
+    assert msg["error"]["code"] == "not_loaded"
+
+
+async def test_get_info_fetch_fails_1(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test create network."""
+    await async_setup_component(hass, "otbr", {})
+
+    with patch(
+        "python_otbr_api.OTBR.set_enabled",
+        side_effect=python_otbr_api.OTBRError,
+    ):
+        await websocket_client.send_json(
+            {
+                "id": 5,
+                "type": "otbr/create_network",
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert msg["id"] == 5
+    assert not msg["success"]
+    assert msg["error"]["code"] == "set_enabled_failed"
+
+
+async def test_get_info_fetch_fails_2(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test create network."""
+    await async_setup_component(hass, "otbr", {})
+
+    with patch(
+        "python_otbr_api.OTBR.set_enabled",
+    ), patch(
+        "python_otbr_api.OTBR.create_active_dataset",
+        side_effect=python_otbr_api.OTBRError,
+    ):
+        await websocket_client.send_json(
+            {
+                "id": 5,
+                "type": "otbr/create_network",
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert msg["id"] == 5
+    assert not msg["success"]
+    assert msg["error"]["code"] == "create_active_dataset_failed"
+
+
+async def test_get_info_fetch_fails_3(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test create network."""
+    await async_setup_component(hass, "otbr", {})
+
+    with patch(
+        "python_otbr_api.OTBR.set_enabled",
+        side_effect=[None, python_otbr_api.OTBRError],
+    ), patch(
+        "python_otbr_api.OTBR.create_active_dataset",
+    ):
+        await websocket_client.send_json(
+            {
+                "id": 5,
+                "type": "otbr/create_network",
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert msg["id"] == 5
+    assert not msg["success"]
+    assert msg["error"]["code"] == "set_enabled_failed"