Add WS command for connecting OTBR to a known Thread network (#89692)

* Add WS command for connecting OTBR to a known Thread network

* Add test

tests/components/otbr/test_websocket_api.py

@@ -4,11 +4,11 @@ from unittest.mock import patch
 import pytest
 import python_otbr_api
 
-from homeassistant.components import otbr
+from homeassistant.components import otbr, thread
 from homeassistant.core import HomeAssistant
 from homeassistant.setup import async_setup_component
 
-from . import BASE_URL, DATASET_CH16
+from . import BASE_URL, DATASET_CH15, DATASET_CH16
 
 from tests.test_util.aiohttp import AiohttpClientMocker
 from tests.typing import WebSocketGenerator
@@ -290,6 +290,190 @@ async def test_create_network_fails_5(
     assert msg["error"]["code"] == "get_active_dataset_tlvs_empty"
 
 
+async def test_set_network(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test set network."""
+
+    await thread.async_add_dataset(hass, "test", DATASET_CH15.hex())
+    dataset_store = await thread.dataset_store.async_get_store(hass)
+    dataset_id = list(dataset_store.datasets)[1]
+
+    with patch(
+        "python_otbr_api.OTBR.set_active_dataset_tlvs"
+    ) as set_active_dataset_tlvs_mock, patch(
+        "python_otbr_api.OTBR.set_enabled"
+    ) as set_enabled_mock:
+        await websocket_client.send_json_auto_id(
+            {
+                "type": "otbr/set_network",
+                "dataset_id": dataset_id,
+            }
+        )
+
+        msg = await websocket_client.receive_json()
+        assert msg["success"]
+        assert msg["result"] is None
+
+    set_active_dataset_tlvs_mock.assert_called_once_with(DATASET_CH15)
+    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_set_network_no_entry(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test set network."""
+    await async_setup_component(hass, "otbr", {})
+    websocket_client = await hass_ws_client(hass)
+    await websocket_client.send_json_auto_id(
+        {
+            "type": "otbr/set_network",
+            "dataset_id": "abc",
+        }
+    )
+
+    msg = await websocket_client.receive_json()
+    assert not msg["success"]
+    assert msg["error"]["code"] == "not_loaded"
+
+
+async def test_set_network_channel_conflict(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test set network."""
+
+    dataset_store = await thread.dataset_store.async_get_store(hass)
+    dataset_id = list(dataset_store.datasets)[0]
+
+    await websocket_client.send_json_auto_id(
+        {
+            "type": "otbr/set_network",
+            "dataset_id": dataset_id,
+        }
+    )
+
+    msg = await websocket_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "channel_conflict"
+
+
+async def test_set_network_unknown_dataset(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test set network."""
+
+    await websocket_client.send_json_auto_id(
+        {
+            "type": "otbr/set_network",
+            "dataset_id": "abc",
+        }
+    )
+
+    msg = await websocket_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "unknown_dataset"
+
+
+async def test_set_network_fails_1(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test set network."""
+    await thread.async_add_dataset(hass, "test", DATASET_CH15.hex())
+    dataset_store = await thread.dataset_store.async_get_store(hass)
+    dataset_id = list(dataset_store.datasets)[1]
+
+    with patch(
+        "python_otbr_api.OTBR.set_enabled",
+        side_effect=python_otbr_api.OTBRError,
+    ):
+        await websocket_client.send_json_auto_id(
+            {
+                "type": "otbr/set_network",
+                "dataset_id": dataset_id,
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "set_enabled_failed"
+
+
+async def test_set_network_fails_2(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test set network."""
+    await thread.async_add_dataset(hass, "test", DATASET_CH15.hex())
+    dataset_store = await thread.dataset_store.async_get_store(hass)
+    dataset_id = list(dataset_store.datasets)[1]
+
+    with patch(
+        "python_otbr_api.OTBR.set_enabled",
+    ), patch(
+        "python_otbr_api.OTBR.set_active_dataset_tlvs",
+        side_effect=python_otbr_api.OTBRError,
+    ):
+        await websocket_client.send_json_auto_id(
+            {
+                "type": "otbr/set_network",
+                "dataset_id": dataset_id,
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "set_active_dataset_tlvs_failed"
+
+
+async def test_set_network_fails_3(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test set network."""
+    await thread.async_add_dataset(hass, "test", DATASET_CH15.hex())
+    dataset_store = await thread.dataset_store.async_get_store(hass)
+    dataset_id = list(dataset_store.datasets)[1]
+
+    with patch(
+        "python_otbr_api.OTBR.set_enabled",
+        side_effect=[None, python_otbr_api.OTBRError],
+    ), patch(
+        "python_otbr_api.OTBR.set_active_dataset_tlvs",
+    ):
+        await websocket_client.send_json_auto_id(
+            {
+                "type": "otbr/set_network",
+                "dataset_id": dataset_id,
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "set_enabled_failed"
+
+
 async def test_get_extended_address(
     hass: HomeAssistant,
     aioclient_mock: AiohttpClientMocker,