diff --git a/homeassistant/components/otbr/__init__.py b/homeassistant/components/otbr/__init__.py index ebe2ab00257..c2020402283 100644 --- a/homeassistant/components/otbr/__init__.py +++ b/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.""" diff --git a/homeassistant/components/otbr/websocket_api.py b/homeassistant/components/otbr/websocket_api.py index a07819793b1..d88581696c4 100644 --- a/homeassistant/components/otbr/websocket_api.py +++ b/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"]) diff --git a/tests/components/otbr/test_websocket_api.py b/tests/components/otbr/test_websocket_api.py index c071e760eb7..de01c6153e2 100644 --- a/tests/components/otbr/test_websocket_api.py +++ b/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"