Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add WS command for changing thread channels (#94525)

pull/96896/head
Erik Montnemery 2023-07-19 10:48:32 +02:00 committed by GitHub
parent 80a7447030
commit b53eae2846
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 200 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
homeassistant/components/otbr/websocket_api.py
View File

@ -3,11 +3,14 @@
from typing import cast
import python_otbr_api
from python_otbr_api import tlv_parser
from python_otbr_api import PENDING_DATASET_DELAY_TIMER, tlv_parser
from python_otbr_api.tlv_parser import MeshcopTLVType
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
is_multiprotocol_url,
)
from homeassistant.components.thread import async_add_dataset, async_get_dataset
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
@ -22,6 +25,7 @@ def async_setup(hass: HomeAssistant) -> None:
websocket_api.async_register_command(hass, websocket_info)
websocket_api.async_register_command(hass, websocket_create_network)
websocket_api.async_register_command(hass, websocket_get_extended_address)
websocket_api.async_register_command(hass, websocket_set_channel)
websocket_api.async_register_command(hass, websocket_set_network)
@ -43,7 +47,8 @@ async def websocket_info(
data: OTBRData = hass.data[DOMAIN]
try:
dataset = await data.get_active_dataset_tlvs()
dataset = await data.get_active_dataset()
dataset_tlvs = await data.get_active_dataset_tlvs()
except HomeAssistantError as exc:
connection.send_error(msg["id"], "get_dataset_failed", str(exc))
return
@ -52,7 +57,8 @@ async def websocket_info(
msg["id"],
{
"url": data.url,
"active_dataset_tlvs": dataset.hex() if dataset else None,
"active_dataset_tlvs": dataset_tlvs.hex() if dataset_tlvs else None,
"channel": dataset.channel if dataset else None,
},
)
@ -205,3 +211,41 @@ async def websocket_get_extended_address(
return
connection.send_result(msg["id"], {"extended_address": extended_address.hex()})
@websocket_api.websocket_command(
{
"type": "otbr/set_channel",
vol.Required("channel"): int,
}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_set_channel(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
) -> None:
"""Set current channel."""
if DOMAIN not in hass.data:
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
return
data: OTBRData = hass.data[DOMAIN]
if is_multiprotocol_url(data.url):
connection.send_error(
msg["id"],
"multiprotocol_enabled",
"Channel change not allowed when in multiprotocol mode",
)
return
channel: int = msg["channel"]
delay: float = PENDING_DATASET_DELAY_TIMER / 1000
try:
await data.set_channel(channel)
except HomeAssistantError as exc:
connection.send_error(msg["id"], "set_channel_failed", str(exc))
return
connection.send_result(msg["id"], {"delay": delay})

4
tests/components/otbr/__init__.py
View File

@ -1,7 +1,7 @@
"""Tests for the Open Thread Border Router integration."""
BASE_URL = "http://core-silabs-multiprotocol:8081"
CONFIG_ENTRY_DATA = {"url": "http://core-silabs-multiprotocol:8081"}
CONFIG_ENTRY_DATA_2 = {"url": "http://core-silabs-multiprotocol_2:8081"}
CONFIG_ENTRY_DATA_MULTIPAN = {"url": "http://core-silabs-multiprotocol:8081"}
CONFIG_ENTRY_DATA_THREAD = {"url": "/dev/ttyAMA1"}
DATASET_CH15 = bytes.fromhex(
"0E080000000000010000000300000F35060004001FFFE00208F642646DA209B1D00708FDF57B5A"

26
tests/components/otbr/conftest.py
View File

@ -6,16 +6,34 @@ import pytest
from homeassistant.components import otbr
from homeassistant.core import HomeAssistant
from . import CONFIG_ENTRY_DATA, DATASET_CH16
from . import CONFIG_ENTRY_DATA_MULTIPAN, CONFIG_ENTRY_DATA_THREAD, DATASET_CH16
from tests.common import MockConfigEntry
@pytest.fixture(name="otbr_config_entry")
async def otbr_config_entry_fixture(hass):
@pytest.fixture(name="otbr_config_entry_multipan")
async def otbr_config_entry_multipan_fixture(hass):
"""Mock Open Thread Border Router config entry."""
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
)
config_entry.add_to_hass(hass)
with patch(
"python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=DATASET_CH16
), patch(
"homeassistant.components.otbr.util.compute_pskc"
): # Patch to speed up tests
assert await hass.config_entries.async_setup(config_entry.entry_id)
@pytest.fixture(name="otbr_config_entry_thread")
async def otbr_config_entry_thread_fixture(hass):
"""Mock Open Thread Border Router config entry."""
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA_THREAD,
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",

36
tests/components/otbr/test_init.py
View File

@ -15,8 +15,8 @@ from homeassistant.setup import async_setup_component
from . import (
BASE_URL,
CONFIG_ENTRY_DATA,
CONFIG_ENTRY_DATA_2,
CONFIG_ENTRY_DATA_MULTIPAN,
CONFIG_ENTRY_DATA_THREAD,
DATASET_CH15,
DATASET_CH16,
DATASET_INSECURE_NW_KEY,
@ -38,7 +38,7 @@ async def test_import_dataset(hass: HomeAssistant) -> None:
issue_registry = ir.async_get(hass)
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="My OTBR",
@ -74,7 +74,7 @@ async def test_import_share_radio_channel_collision(
multiprotocol_addon_manager_mock.async_get_channel.return_value = 15
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="My OTBR",
@ -107,7 +107,7 @@ async def test_import_share_radio_no_channel_collision(
multiprotocol_addon_manager_mock.async_get_channel.return_value = 15
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="My OTBR",
@ -138,7 +138,7 @@ async def test_import_insecure_dataset(hass: HomeAssistant, dataset: bytes) -> N
issue_registry = ir.async_get(hass)
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="My OTBR",
@ -169,7 +169,7 @@ async def test_config_entry_not_ready(hass: HomeAssistant, error) -> None:
"""Test raising ConfigEntryNotReady ."""
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="My OTBR",
@ -182,7 +182,7 @@ async def test_config_entry_not_ready(hass: HomeAssistant, error) -> None:
async def test_config_entry_update(hass: HomeAssistant) -> None:
"""Test update config entry settings."""
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="My OTBR",
@ -193,10 +193,10 @@ async def test_config_entry_update(hass: HomeAssistant) -> None:
with patch("python_otbr_api.OTBR", return_value=mock_api) as mock_otrb_api:
assert await hass.config_entries.async_setup(config_entry.entry_id)
mock_otrb_api.assert_called_once_with(CONFIG_ENTRY_DATA["url"], ANY, ANY)
mock_otrb_api.assert_called_once_with(CONFIG_ENTRY_DATA_MULTIPAN["url"], ANY, ANY)
new_config_entry_data = {"url": "http://core-silabs-multiprotocol:8082"}
assert CONFIG_ENTRY_DATA["url"] != new_config_entry_data["url"]
assert CONFIG_ENTRY_DATA_MULTIPAN["url"] != new_config_entry_data["url"]
with patch("python_otbr_api.OTBR", return_value=mock_api) as mock_otrb_api:
hass.config_entries.async_update_entry(config_entry, data=new_config_entry_data)
await hass.async_block_till_done()
@ -205,7 +205,7 @@ async def test_config_entry_update(hass: HomeAssistant) -> None:
async def test_remove_entry(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry_multipan
) -> None:
"""Test async_get_active_dataset_tlvs after removing the config entry."""
@ -221,7 +221,7 @@ async def test_remove_entry(
async def test_get_active_dataset_tlvs(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry_multipan
) -> None:
"""Test async_get_active_dataset_tlvs."""
@ -239,7 +239,7 @@ async def test_get_active_dataset_tlvs(
async def test_get_active_dataset_tlvs_empty(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry_multipan
) -> None:
"""Test async_get_active_dataset_tlvs."""
@ -255,7 +255,7 @@ async def test_get_active_dataset_tlvs_addon_not_installed(hass: HomeAssistant)
async def test_get_active_dataset_tlvs_404(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry_multipan
) -> None:
"""Test async_get_active_dataset_tlvs with error."""
@ -265,7 +265,7 @@ async def test_get_active_dataset_tlvs_404(
async def test_get_active_dataset_tlvs_201(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry_multipan
) -> None:
"""Test async_get_active_dataset_tlvs with error."""
@ -275,7 +275,7 @@ async def test_get_active_dataset_tlvs_201(
async def test_get_active_dataset_tlvs_invalid(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_config_entry_multipan
) -> None:
"""Test async_get_active_dataset_tlvs with error."""
@ -290,13 +290,13 @@ async def test_remove_extra_entries(
"""Test we remove additional config entries."""
config_entry1 = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
)
config_entry2 = MockConfigEntry(
data=CONFIG_ENTRY_DATA_2,
data=CONFIG_ENTRY_DATA_THREAD,
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",

20
tests/components/otbr/test_silabs_multiprotocol.py
View File

@ -31,7 +31,9 @@ DATASET_CH16_PENDING = (
)
async def test_async_change_channel(hass: HomeAssistant, otbr_config_entry) -> None:
async def test_async_change_channel(
hass: HomeAssistant, otbr_config_entry_multipan
) -> None:
"""Test test_async_change_channel."""
store = await dataset_store.async_get_store(hass)
@ -55,7 +57,7 @@ async def test_async_change_channel(hass: HomeAssistant, otbr_config_entry) -> N
async def test_async_change_channel_no_pending(
hass: HomeAssistant, otbr_config_entry
hass: HomeAssistant, otbr_config_entry_multipan
) -> None:
"""Test test_async_change_channel when the pending dataset already expired."""
@ -83,7 +85,7 @@ async def test_async_change_channel_no_pending(
async def test_async_change_channel_no_update(
hass: HomeAssistant, otbr_config_entry
hass: HomeAssistant, otbr_config_entry_multipan
) -> None:
"""Test test_async_change_channel when we didn't get a dataset from the OTBR."""
@ -112,7 +114,9 @@ async def test_async_change_channel_no_otbr(hass: HomeAssistant) -> None:
mock_set_channel.assert_not_awaited()
async def test_async_get_channel(hass: HomeAssistant, otbr_config_entry) -> None:
async def test_async_get_channel(
hass: HomeAssistant, otbr_config_entry_multipan
) -> None:
"""Test test_async_get_channel."""
with patch(
@ -124,7 +128,7 @@ async def test_async_get_channel(hass: HomeAssistant, otbr_config_entry) -> None
async def test_async_get_channel_no_dataset(
hass: HomeAssistant, otbr_config_entry
hass: HomeAssistant, otbr_config_entry_multipan
) -> None:
"""Test test_async_get_channel."""
@ -136,7 +140,9 @@ async def test_async_get_channel_no_dataset(
mock_get_active_dataset.assert_awaited_once_with()
async def test_async_get_channel_error(hass: HomeAssistant, otbr_config_entry) -> None:
async def test_async_get_channel_error(
hass: HomeAssistant, otbr_config_entry_multipan
) -> None:
"""Test test_async_get_channel."""
with patch(
@ -160,7 +166,7 @@ async def test_async_get_channel_no_otbr(hass: HomeAssistant) -> None:
[(OTBR_MULTIPAN_URL, True), (OTBR_NON_MULTIPAN_URL, False)],
)
async def test_async_using_multipan(
hass: HomeAssistant, otbr_config_entry, url: str, expected: bool
hass: HomeAssistant, otbr_config_entry_multipan, url: str, expected: bool
) -> None:
"""Test async_change_channel when otbr is not configured."""
data: otbr.OTBRData = hass.data[otbr.DOMAIN]

118
tests/components/otbr/test_websocket_api.py
View File

@ -23,20 +23,23 @@ async def websocket_client(hass, hass_ws_client):
async def test_get_info(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test async_get_info."""
aioclient_mock.get(f"{BASE_URL}/node/dataset/active", text=DATASET_CH16.hex())
with patch(
"python_otbr_api.OTBR.get_active_dataset",
return_value=python_otbr_api.ActiveDataSet(channel=16),
), patch("python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=DATASET_CH16):
await websocket_client.send_json_auto_id({"type": "otbr/info"})
msg = await websocket_client.receive_json()
assert msg["success"]
assert msg["result"] == {
"url": BASE_URL,
"active_dataset_tlvs": DATASET_CH16.hex().lower(),
"channel": 16,
}
@ -58,12 +61,12 @@ async def test_get_info_no_entry(
async def test_get_info_fetch_fails(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test async_get_info."""
with patch(
"python_otbr_api.OTBR.get_active_dataset_tlvs",
"python_otbr_api.OTBR.get_active_dataset",
side_effect=python_otbr_api.OTBRError,
):
await websocket_client.send_json_auto_id({"type": "otbr/info"})
@ -76,7 +79,7 @@ async def test_get_info_fetch_fails(
async def test_create_network(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -127,7 +130,7 @@ async def test_create_network_no_entry(
async def test_create_network_fails_1(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -145,7 +148,7 @@ async def test_create_network_fails_1(
async def test_create_network_fails_2(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -165,7 +168,7 @@ async def test_create_network_fails_2(
async def test_create_network_fails_3(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -187,7 +190,7 @@ async def test_create_network_fails_3(
async def test_create_network_fails_4(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -209,7 +212,7 @@ async def test_create_network_fails_4(
async def test_create_network_fails_5(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -228,7 +231,7 @@ async def test_create_network_fails_5(
async def test_create_network_fails_6(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test create network."""
@ -248,7 +251,7 @@ async def test_create_network_fails_6(
async def test_set_network(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set network."""
@ -303,7 +306,7 @@ async def test_set_network_channel_conflict(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
multiprotocol_addon_manager_mock,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set network."""
@ -329,7 +332,7 @@ async def test_set_network_channel_conflict(
async def test_set_network_unknown_dataset(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set network."""
@ -350,7 +353,7 @@ async def test_set_network_unknown_dataset(
async def test_set_network_fails_1(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set network."""
@ -377,7 +380,7 @@ async def test_set_network_fails_1(
async def test_set_network_fails_2(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set network."""
@ -406,7 +409,7 @@ async def test_set_network_fails_2(
async def test_set_network_fails_3(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set network."""
@ -435,7 +438,7 @@ async def test_set_network_fails_3(
async def test_get_extended_address(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test get extended address."""
@ -469,7 +472,7 @@ async def test_get_extended_address_no_entry(
async def test_get_extended_address_fetch_fails(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test get extended address."""
@ -482,3 +485,76 @@ async def test_get_extended_address_fetch_fails(
assert not msg["success"]
assert msg["error"]["code"] == "get_extended_address_failed"
async def test_set_channel(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry_thread,
websocket_client,
) -> None:
"""Test set channel."""
with patch("python_otbr_api.OTBR.set_channel"):
await websocket_client.send_json_auto_id(
{"type": "otbr/set_channel", "channel": 12}
)
msg = await websocket_client.receive_json()
assert msg["success"]
assert msg["result"] == {"delay": 300.0}
async def test_set_channel_multiprotocol(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test set channel."""
with patch("python_otbr_api.OTBR.set_channel"):
await websocket_client.send_json_auto_id(
{"type": "otbr/set_channel", "channel": 12}
)
msg = await websocket_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "multiprotocol_enabled"
async def test_set_channel_no_entry(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test set channel."""
await async_setup_component(hass, "otbr", {})
websocket_client = await hass_ws_client(hass)
await websocket_client.send_json_auto_id(
{"type": "otbr/set_channel", "channel": 12}
)
msg = await websocket_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "not_loaded"
async def test_set_channel_fails(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry_thread,
websocket_client,
) -> None:
"""Test set channel."""
with patch(
"python_otbr_api.OTBR.set_channel",
side_effect=python_otbr_api.OTBRError,
):
await websocket_client.send_json_auto_id(
{"type": "otbr/set_channel", "channel": 12}
)
msg = await websocket_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "set_channel_failed"