Add OTBR WebSocket API (#86107)

* Add OTBR WebSocket API

* Not always active dataset

* Move logic to data class

* Remove retry until we need it

* Test all the things
This commit is contained in:
Paulus Schoutsen 2023-01-18 11:33:15 -05:00 committed by GitHub
parent e43802eb07
commit f2b348dbdf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 204 additions and 28 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import dataclasses
from functools import wraps
import python_otbr_api
@ -9,22 +10,48 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from . import websocket_api
from .const import DOMAIN
def _handle_otbr_error(func):
"""Handle OTBR errors."""
@wraps(func)
async def _func(self, *args, **kwargs):
try:
return await func(self, *args, **kwargs)
except python_otbr_api.OTBRError as exc:
raise HomeAssistantError("Failed to call OTBR API") from exc
return _func
@dataclasses.dataclass
class OTBRData:
"""Container for OTBR data."""
url: str
api: python_otbr_api.OTBR
@_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()
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Open Thread Border Router component."""
websocket_api.async_setup(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up an Open Thread Border Router config entry."""
hass.data[DOMAIN] = OTBRData(entry.data["url"])
api = python_otbr_api.OTBR(entry.data["url"], async_get_clientsession(hass), 10)
hass.data[DOMAIN] = OTBRData(entry.data["url"], api)
return True
@ -34,26 +61,14 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
def _async_get_thread_rest_service_url(hass) -> str:
"""Return Thread REST API URL."""
otbr_data: OTBRData | None = hass.data.get(DOMAIN)
if not otbr_data:
raise HomeAssistantError("otbr not setup")
return otbr_data.url
async def async_get_active_dataset_tlvs(hass: HomeAssistant) -> bytes | None:
"""Get current active operational dataset in TLVS format, or None.
Returns None if there is no active operational dataset.
Raises if the http status is 400 or higher or if the response is invalid.
"""
if DOMAIN not in hass.data:
raise HomeAssistantError("OTBR API not available")
api = python_otbr_api.OTBR(
_async_get_thread_rest_service_url(hass), async_get_clientsession(hass), 10
)
try:
return await api.get_active_dataset_tlvs()
except python_otbr_api.OTBRError as exc:
raise HomeAssistantError("Failed to call OTBR API") from exc
data: OTBRData = hass.data[DOMAIN]
return await data.get_active_dataset_tlvs()

View file

@ -1,11 +1,11 @@
{
"codeowners": ["@home-assistant/core"],
"after_dependencies": ["hassio"],
"domain": "otbr",
"iot_class": "local_polling",
"name": "Thread",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/otbr",
"integration_type": "system",
"name": "Thread",
"requirements": ["python-otbr-api==1.0.1"]
"requirements": ["python-otbr-api==1.0.1"],
"after_dependencies": ["hassio"],
"codeowners": ["@home-assistant/core"],
"iot_class": "local_polling",
"integration_type": "service"
}

View file

@ -0,0 +1,56 @@
"""Websocket API for OTBR."""
from typing import TYPE_CHECKING
from homeassistant.components.websocket_api import (
ActiveConnection,
async_register_command,
async_response,
websocket_command,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
if TYPE_CHECKING:
from . import OTBRData
@callback
def async_setup(hass) -> None:
"""Set up the OTBR Websocket API."""
async_register_command(hass, websocket_info)
@websocket_command(
{
"type": "otbr/info",
}
)
@async_response
async def websocket_info(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Get OTBR info."""
if DOMAIN not in hass.data:
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
return
data: OTBRData = hass.data[DOMAIN]
try:
dataset = await data.get_active_dataset_tlvs()
except HomeAssistantError as exc:
connection.send_error(msg["id"], "get_dataset_failed", str(exc))
return
if dataset:
dataset = dataset.hex()
connection.send_result(
msg["id"],
{
"url": data.url,
"active_dataset_tlvs": dataset,
},
)

View file

@ -3965,6 +3965,12 @@
"config_flow": false,
"iot_class": "local_polling"
},
"otbr": {
"name": "Thread",
"integration_type": "service",
"config_flow": true,
"iot_class": "local_polling"
},
"otp": {
"name": "One-Time Password (OTP)",
"integration_type": "hub",

View file

@ -1 +1,2 @@
"""Tests for the Thread integration."""
BASE_URL = "http://core-silabs-multiprotocol:8081"

View file

@ -1,4 +1,5 @@
"""Test fixtures for the Home Assistant SkyConnect integration."""
from unittest.mock import patch
import pytest
@ -19,4 +20,5 @@ async def thread_config_entry_fixture(hass):
title="Thread",
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
with patch("python_otbr_api.OTBR.get_active_dataset_tlvs"):
assert await hass.config_entries.async_setup(config_entry.entry_id)

View file

@ -8,9 +8,9 @@ from homeassistant.components import otbr
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from tests.test_util.aiohttp import AiohttpClientMocker
from . import BASE_URL
BASE_URL = "http://core-silabs-multiprotocol:8081"
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_remove_entry(

View file

@ -0,0 +1,96 @@
"""Test OTBR Websocket API."""
from unittest.mock import patch
import pytest
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from . import BASE_URL
from tests.test_util.aiohttp import AiohttpClientMocker
@pytest.fixture
async def websocket_client(hass, hass_ws_client):
"""Create a websocket client."""
return await hass_ws_client(hass)
async def test_get_info(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
thread_config_entry,
websocket_client,
):
"""Test async_get_info."""
mock_response = (
"0E080000000000010000000300001035060004001FFFE00208F642646DA209B1C00708FDF57B5A"
"0FE2AAF60510DE98B5BA1A528FEE049D4B4B01835375030D4F70656E5468726561642048410102"
"25A40410F5DD18371BFD29E1A601EF6FFAD94C030C0402A0F7F8"
)
aioclient_mock.get(f"{BASE_URL}/node/dataset/active", text=mock_response)
await websocket_client.send_json(
{
"id": 5,
"type": "otbr/info",
}
)
msg = await websocket_client.receive_json()
assert msg["id"] == 5
assert msg["success"]
assert msg["result"] == {
"url": BASE_URL,
"active_dataset_tlvs": mock_response.lower(),
}
async def test_get_info_no_entry(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
websocket_client,
):
"""Test async_get_info."""
await async_setup_component(hass, "otbr", {})
await websocket_client.send_json(
{
"id": 5,
"type": "otbr/info",
}
)
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(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
thread_config_entry,
websocket_client,
):
"""Test async_get_info."""
await async_setup_component(hass, "otbr", {})
with patch(
"homeassistant.components.otbr.OTBRData.get_active_dataset_tlvs",
side_effect=HomeAssistantError,
):
await websocket_client.send_json(
{
"id": 5,
"type": "otbr/info",
}
)
msg = await websocket_client.receive_json()
assert msg["id"] == 5
assert not msg["success"]
assert msg["error"]["code"] == "get_dataset_failed"