mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Use HassKey in otbr (#124240)
This commit is contained in:
parent
d99f1631ca
commit
b31c6012ae
@ -14,7 +14,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|||||||
from homeassistant.helpers.typing import ConfigType
|
from homeassistant.helpers.typing import ConfigType
|
||||||
|
|
||||||
from . import websocket_api
|
from . import websocket_api
|
||||||
from .const import DOMAIN
|
from .const import DATA_OTBR, DOMAIN
|
||||||
from .util import OTBRData, update_issues
|
from .util import OTBRData, update_issues
|
||||||
|
|
||||||
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
||||||
@ -67,14 +67,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||||||
|
|
||||||
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
|
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
|
||||||
|
|
||||||
hass.data[DOMAIN] = otbrdata
|
hass.data[DATA_OTBR] = otbrdata
|
||||||
|
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
"""Unload a config entry."""
|
"""Unload a config entry."""
|
||||||
hass.data.pop(DOMAIN)
|
hass.data.pop(DATA_OTBR)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
@ -1,5 +1,15 @@
|
|||||||
"""Constants for the Open Thread Border Router integration."""
|
"""Constants for the Open Thread Border Router integration."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from homeassistant.util.hass_dict import HassKey
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from .util import OTBRData
|
||||||
|
|
||||||
DOMAIN = "otbr"
|
DOMAIN = "otbr"
|
||||||
|
DATA_OTBR: HassKey[OTBRData] = HassKey(DOMAIN)
|
||||||
|
|
||||||
DEFAULT_CHANNEL = 15
|
DEFAULT_CHANNEL = 15
|
||||||
|
@ -18,7 +18,7 @@ from homeassistant.components.thread import async_add_dataset
|
|||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.exceptions import HomeAssistantError
|
from homeassistant.exceptions import HomeAssistantError
|
||||||
|
|
||||||
from . import DOMAIN
|
from .const import DATA_OTBR, DOMAIN
|
||||||
from .util import OTBRData
|
from .util import OTBRData
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
@ -45,10 +45,10 @@ def async_get_otbr_data[**_P, _R, _R_Def](
|
|||||||
hass: HomeAssistant, *args: _P.args, **kwargs: _P.kwargs
|
hass: HomeAssistant, *args: _P.args, **kwargs: _P.kwargs
|
||||||
) -> _R | _R_Def:
|
) -> _R | _R_Def:
|
||||||
"""Fetch OTBR data and pass to orig_func."""
|
"""Fetch OTBR data and pass to orig_func."""
|
||||||
if DOMAIN not in hass.data:
|
if DATA_OTBR not in hass.data:
|
||||||
return retval
|
return retval
|
||||||
|
|
||||||
data: OTBRData = hass.data[DOMAIN]
|
data = hass.data[DATA_OTBR]
|
||||||
|
|
||||||
if not is_multiprotocol_url(data.url):
|
if not is_multiprotocol_url(data.url):
|
||||||
return retval
|
return retval
|
||||||
|
@ -17,7 +17,7 @@ from homeassistant.components.thread import async_add_dataset, async_get_dataset
|
|||||||
from homeassistant.core import HomeAssistant, callback
|
from homeassistant.core import HomeAssistant, callback
|
||||||
from homeassistant.exceptions import HomeAssistantError
|
from homeassistant.exceptions import HomeAssistantError
|
||||||
|
|
||||||
from .const import DEFAULT_CHANNEL, DOMAIN
|
from .const import DATA_OTBR, DEFAULT_CHANNEL, DOMAIN
|
||||||
from .util import (
|
from .util import (
|
||||||
OTBRData,
|
OTBRData,
|
||||||
compose_default_network_name,
|
compose_default_network_name,
|
||||||
@ -47,11 +47,11 @@ async def websocket_info(
|
|||||||
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
|
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Get OTBR info."""
|
"""Get OTBR info."""
|
||||||
if DOMAIN not in hass.data:
|
if DATA_OTBR not in hass.data:
|
||||||
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
|
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
|
||||||
return
|
return
|
||||||
|
|
||||||
data: OTBRData = hass.data[DOMAIN]
|
data = hass.data[DATA_OTBR]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
border_agent_id = await data.get_border_agent_id()
|
border_agent_id = await data.get_border_agent_id()
|
||||||
@ -99,11 +99,11 @@ def async_get_otbr_data(
|
|||||||
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
|
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Fetch OTBR data and pass to orig_func."""
|
"""Fetch OTBR data and pass to orig_func."""
|
||||||
if DOMAIN not in hass.data:
|
if DATA_OTBR not in hass.data:
|
||||||
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
|
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
|
||||||
return
|
return
|
||||||
|
|
||||||
data: OTBRData = hass.data[DOMAIN]
|
data = hass.data[DATA_OTBR]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
extended_address = await data.get_extended_address()
|
extended_address = await data.get_extended_address()
|
||||||
|
@ -130,8 +130,7 @@ async def test_async_change_channel_non_matching_url(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan
|
hass: HomeAssistant, otbr_config_entry_multipan
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test async_change_channel when otbr is not configured."""
|
"""Test async_change_channel when otbr is not configured."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
hass.data[otbr.DATA_OTBR].url = OTBR_NON_MULTIPAN_URL
|
||||||
data.url = OTBR_NON_MULTIPAN_URL
|
|
||||||
with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel:
|
with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel:
|
||||||
await otbr_silabs_multiprotocol.async_change_channel(hass, 16, delay=0)
|
await otbr_silabs_multiprotocol.async_change_channel(hass, 16, delay=0)
|
||||||
mock_set_channel.assert_not_awaited()
|
mock_set_channel.assert_not_awaited()
|
||||||
@ -188,8 +187,7 @@ async def test_async_get_channel_non_matching_url(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan
|
hass: HomeAssistant, otbr_config_entry_multipan
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test async_change_channel when otbr is not configured."""
|
"""Test async_change_channel when otbr is not configured."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
hass.data[otbr.DATA_OTBR].url = OTBR_NON_MULTIPAN_URL
|
||||||
data.url = OTBR_NON_MULTIPAN_URL
|
|
||||||
with patch("python_otbr_api.OTBR.get_active_dataset") as mock_get_active_dataset:
|
with patch("python_otbr_api.OTBR.get_active_dataset") as mock_get_active_dataset:
|
||||||
assert await otbr_silabs_multiprotocol.async_get_channel(hass) is None
|
assert await otbr_silabs_multiprotocol.async_get_channel(hass) is None
|
||||||
mock_get_active_dataset.assert_not_awaited()
|
mock_get_active_dataset.assert_not_awaited()
|
||||||
@ -203,8 +201,7 @@ async def test_async_using_multipan(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan, url: str, expected: bool
|
hass: HomeAssistant, otbr_config_entry_multipan, url: str, expected: bool
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test async_change_channel when otbr is not configured."""
|
"""Test async_change_channel when otbr is not configured."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
hass.data[otbr.DATA_OTBR].url = url
|
||||||
data.url = url
|
|
||||||
|
|
||||||
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is expected
|
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is expected
|
||||||
|
|
||||||
@ -219,6 +216,5 @@ async def test_async_using_multipan_non_matching_url(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan
|
hass: HomeAssistant, otbr_config_entry_multipan
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test async_change_channel when otbr is not configured."""
|
"""Test async_change_channel when otbr is not configured."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
hass.data[otbr.DATA_OTBR].url = OTBR_NON_MULTIPAN_URL
|
||||||
data.url = OTBR_NON_MULTIPAN_URL
|
|
||||||
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is False
|
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is False
|
||||||
|
@ -33,15 +33,13 @@ async def test_get_allowed_channel(
|
|||||||
|
|
||||||
async def test_factory_reset(hass: HomeAssistant, otbr_config_entry_multipan) -> None:
|
async def test_factory_reset(hass: HomeAssistant, otbr_config_entry_multipan) -> None:
|
||||||
"""Test factory_reset."""
|
"""Test factory_reset."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch("python_otbr_api.OTBR.factory_reset") as factory_reset_mock,
|
patch("python_otbr_api.OTBR.factory_reset") as factory_reset_mock,
|
||||||
patch(
|
patch(
|
||||||
"python_otbr_api.OTBR.delete_active_dataset"
|
"python_otbr_api.OTBR.delete_active_dataset"
|
||||||
) as delete_active_dataset_mock,
|
) as delete_active_dataset_mock,
|
||||||
):
|
):
|
||||||
await data.factory_reset()
|
await hass.data[otbr.DATA_OTBR].factory_reset()
|
||||||
|
|
||||||
delete_active_dataset_mock.assert_not_called()
|
delete_active_dataset_mock.assert_not_called()
|
||||||
factory_reset_mock.assert_called_once_with()
|
factory_reset_mock.assert_called_once_with()
|
||||||
@ -51,8 +49,6 @@ async def test_factory_reset_not_supported(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan
|
hass: HomeAssistant, otbr_config_entry_multipan
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test factory_reset."""
|
"""Test factory_reset."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch(
|
patch(
|
||||||
"python_otbr_api.OTBR.factory_reset",
|
"python_otbr_api.OTBR.factory_reset",
|
||||||
@ -62,7 +58,7 @@ async def test_factory_reset_not_supported(
|
|||||||
"python_otbr_api.OTBR.delete_active_dataset"
|
"python_otbr_api.OTBR.delete_active_dataset"
|
||||||
) as delete_active_dataset_mock,
|
) as delete_active_dataset_mock,
|
||||||
):
|
):
|
||||||
await data.factory_reset()
|
await hass.data[otbr.DATA_OTBR].factory_reset()
|
||||||
|
|
||||||
delete_active_dataset_mock.assert_called_once_with()
|
delete_active_dataset_mock.assert_called_once_with()
|
||||||
factory_reset_mock.assert_called_once_with()
|
factory_reset_mock.assert_called_once_with()
|
||||||
@ -72,8 +68,6 @@ async def test_factory_reset_error_1(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan
|
hass: HomeAssistant, otbr_config_entry_multipan
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test factory_reset."""
|
"""Test factory_reset."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch(
|
patch(
|
||||||
"python_otbr_api.OTBR.factory_reset",
|
"python_otbr_api.OTBR.factory_reset",
|
||||||
@ -86,7 +80,7 @@ async def test_factory_reset_error_1(
|
|||||||
HomeAssistantError,
|
HomeAssistantError,
|
||||||
),
|
),
|
||||||
):
|
):
|
||||||
await data.factory_reset()
|
await hass.data[otbr.DATA_OTBR].factory_reset()
|
||||||
|
|
||||||
delete_active_dataset_mock.assert_not_called()
|
delete_active_dataset_mock.assert_not_called()
|
||||||
factory_reset_mock.assert_called_once_with()
|
factory_reset_mock.assert_called_once_with()
|
||||||
@ -96,8 +90,6 @@ async def test_factory_reset_error_2(
|
|||||||
hass: HomeAssistant, otbr_config_entry_multipan
|
hass: HomeAssistant, otbr_config_entry_multipan
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test factory_reset."""
|
"""Test factory_reset."""
|
||||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch(
|
patch(
|
||||||
"python_otbr_api.OTBR.factory_reset",
|
"python_otbr_api.OTBR.factory_reset",
|
||||||
@ -111,7 +103,7 @@ async def test_factory_reset_error_2(
|
|||||||
HomeAssistantError,
|
HomeAssistantError,
|
||||||
),
|
),
|
||||||
):
|
):
|
||||||
await data.factory_reset()
|
await hass.data[otbr.DATA_OTBR].factory_reset()
|
||||||
|
|
||||||
delete_active_dataset_mock.assert_called_once_with()
|
delete_active_dataset_mock.assert_called_once_with()
|
||||||
factory_reset_mock.assert_called_once_with()
|
factory_reset_mock.assert_called_once_with()
|
||||||
|
Loading…
x
Reference in New Issue
Block a user