evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Include extended address in response to WS otbr/info (#98440)

Browse source
This commit is contained in:
Erik Montnemery 2023-08-15 09:48:29 +02:00 • committed by GitHub
parent e6ea70fd00
commit 94ad4786c3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 9 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
homeassistant/components/otbr/websocket_api.py
View file

@ -24,7 +24,6 @@ def async_setup(hass: HomeAssistant) -> None:
"""Set up the OTBR Websocket API."""
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)
@ -50,8 +49,9 @@ async def websocket_info(
border_agent_id = await data.get_border_agent_id()
dataset = await data.get_active_dataset()
dataset_tlvs = await data.get_active_dataset_tlvs()
extended_address = await data.get_extended_address()
except HomeAssistantError as exc:
connection.send_error(msg["id"], "get_dataset_failed", str(exc))
connection.send_error(msg["id"], "otbr_info_failed", str(exc))
return
# The border agent ID is checked when the OTBR config entry is setup,
@ -60,10 +60,11 @@ async def websocket_info(
connection.send_result(
msg["id"],
{
"url": data.url,
"active_dataset_tlvs": dataset_tlvs.hex() if dataset_tlvs else None,
"border_agent_id": border_agent_id.hex(),
"channel": dataset.channel if dataset else None,
"extended_address": extended_address.hex(),
"url": data.url,
},
)
@ -192,32 +193,6 @@ async def websocket_set_network(
connection.send_result(msg["id"])
@websocket_api.websocket_command(
{
"type": "otbr/get_extended_address",
}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_get_extended_address(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
) -> None:
"""Get extended address (EUI-64)."""
if DOMAIN not in hass.data:
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
return
data: OTBRData = hass.data[DOMAIN]
try:
extended_address = await data.get_extended_address()
except HomeAssistantError as exc:
connection.send_error(msg["id"], "get_extended_address_failed", str(exc))
return
connection.send_result(msg["id"], {"extended_address": extended_address.hex()})
@websocket_api.websocket_command(
{
"type": "otbr/set_channel",

58
tests/components/otbr/test_websocket_api.py
View file

@ -35,6 +35,9 @@ async def test_get_info(
"python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=DATASET_CH16
), patch(
"python_otbr_api.OTBR.get_border_agent_id", return_value=TEST_BORDER_AGENT_ID
), patch(
"python_otbr_api.OTBR.get_extended_address",
return_value=bytes.fromhex("4EF6C4F3FF750626"),
):
await websocket_client.send_json_auto_id({"type": "otbr/info"})
msg = await websocket_client.receive_json()
@ -45,6 +48,7 @@ async def test_get_info(
"active_dataset_tlvs": DATASET_CH16.hex().lower(),
"channel": 16,
"border_agent_id": TEST_BORDER_AGENT_ID.hex(),
"extended_address": "4EF6C4F3FF750626".lower(),
}
@ -80,7 +84,7 @@ async def test_get_info_fetch_fails(
msg = await websocket_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "get_dataset_failed"
assert msg["error"]["code"] == "otbr_info_failed"
async def test_create_network(
@ -442,58 +446,6 @@ async def test_set_network_fails_3(
assert msg["error"]["code"] == "set_enabled_failed"
async def test_get_extended_address(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test get extended address."""
with patch(
"python_otbr_api.OTBR.get_extended_address",
return_value=bytes.fromhex("4EF6C4F3FF750626"),
):
await websocket_client.send_json_auto_id({"type": "otbr/get_extended_address"})
msg = await websocket_client.receive_json()
assert msg["success"]
assert msg["result"] == {"extended_address": "4EF6C4F3FF750626".lower()}
async def test_get_extended_address_no_entry(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test get extended address."""
await async_setup_component(hass, "otbr", {})
websocket_client = await hass_ws_client(hass)
await websocket_client.send_json_auto_id({"type": "otbr/get_extended_address"})
msg = await websocket_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "not_loaded"
async def test_get_extended_address_fetch_fails(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
otbr_config_entry_multipan,
websocket_client,
) -> None:
"""Test get extended address."""
with patch(
"python_otbr_api.OTBR.get_extended_address",
side_effect=python_otbr_api.OTBRError,
):
await websocket_client.send_json_auto_id({"type": "otbr/get_extended_address"})
msg = await websocket_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "get_extended_address_failed"
async def test_set_channel(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,