mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 19:27:45 +00:00
Prepare otbr.silabs_multiprotocol
for multiple config entries (#124219)
* Prepare otbr.silabs_multiprotocol for multiple config entries * Simplify
This commit is contained in:
parent
e3ab30a2a5
commit
50f3c891fa
@ -2,7 +2,10 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from functools import wraps
|
||||
import logging
|
||||
from typing import Any, Concatenate
|
||||
|
||||
import aiohttp
|
||||
from python_otbr_api import tlv_parser
|
||||
@ -21,15 +24,53 @@ from .util import OTBRData
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_change_channel(hass: HomeAssistant, channel: int, delay: float) -> None:
|
||||
def async_get_otbr_data[**_P, _R, _R_Def](
|
||||
retval: _R_Def,
|
||||
) -> Callable[
|
||||
[Callable[Concatenate[HomeAssistant, OTBRData, _P], Coroutine[Any, Any, _R]]],
|
||||
Callable[Concatenate[HomeAssistant, _P], Coroutine[Any, Any, _R | _R_Def]],
|
||||
]:
|
||||
"""Decorate function to get OTBR data."""
|
||||
|
||||
def _async_get_otbr_data(
|
||||
orig_func: Callable[
|
||||
Concatenate[HomeAssistant, OTBRData, _P],
|
||||
Coroutine[Any, Any, _R],
|
||||
],
|
||||
) -> Callable[Concatenate[HomeAssistant, _P], Coroutine[Any, Any, _R | _R_Def]]:
|
||||
"""Decorate function to get OTBR data."""
|
||||
|
||||
@wraps(orig_func)
|
||||
async def async_get_otbr_data_wrapper(
|
||||
hass: HomeAssistant, *args: _P.args, **kwargs: _P.kwargs
|
||||
) -> _R | _R_Def:
|
||||
"""Fetch OTBR data and pass to orig_func."""
|
||||
if DOMAIN not in hass.data:
|
||||
return retval
|
||||
|
||||
data: OTBRData = hass.data[DOMAIN]
|
||||
|
||||
if not is_multiprotocol_url(data.url):
|
||||
return retval
|
||||
|
||||
return await orig_func(hass, data, *args, **kwargs)
|
||||
|
||||
return async_get_otbr_data_wrapper
|
||||
|
||||
return _async_get_otbr_data
|
||||
|
||||
|
||||
@async_get_otbr_data(None)
|
||||
async def async_change_channel(
|
||||
hass: HomeAssistant,
|
||||
data: OTBRData,
|
||||
channel: int,
|
||||
delay: float,
|
||||
) -> None:
|
||||
"""Set the channel to be used.
|
||||
|
||||
Does nothing if not configured.
|
||||
"""
|
||||
if DOMAIN not in hass.data:
|
||||
return
|
||||
|
||||
data: OTBRData = hass.data[DOMAIN]
|
||||
await data.set_channel(channel, delay)
|
||||
|
||||
# Import the new dataset
|
||||
@ -48,16 +89,12 @@ async def async_change_channel(hass: HomeAssistant, channel: int, delay: float)
|
||||
await async_add_dataset(hass, DOMAIN, dataset_tlvs_str)
|
||||
|
||||
|
||||
async def async_get_channel(hass: HomeAssistant) -> int | None:
|
||||
@async_get_otbr_data(None)
|
||||
async def async_get_channel(hass: HomeAssistant, data: OTBRData) -> int | None:
|
||||
"""Return the channel.
|
||||
|
||||
Returns None if not configured.
|
||||
"""
|
||||
if DOMAIN not in hass.data:
|
||||
return None
|
||||
|
||||
data: OTBRData = hass.data[DOMAIN]
|
||||
|
||||
try:
|
||||
dataset = await data.get_active_dataset()
|
||||
except (
|
||||
@ -74,13 +111,10 @@ async def async_get_channel(hass: HomeAssistant) -> int | None:
|
||||
return dataset.channel
|
||||
|
||||
|
||||
async def async_using_multipan(hass: HomeAssistant) -> bool:
|
||||
@async_get_otbr_data(False)
|
||||
async def async_using_multipan(hass: HomeAssistant, data: OTBRData) -> bool:
|
||||
"""Return if the multiprotocol device is used.
|
||||
|
||||
Returns False if not configured.
|
||||
"""
|
||||
if DOMAIN not in hass.data:
|
||||
return False
|
||||
|
||||
data: OTBRData = hass.data[DOMAIN]
|
||||
return is_multiprotocol_url(data.url)
|
||||
return True
|
||||
|
@ -126,6 +126,17 @@ async def test_async_change_channel_no_otbr(hass: HomeAssistant) -> None:
|
||||
mock_set_channel.assert_not_awaited()
|
||||
|
||||
|
||||
async def test_async_change_channel_non_matching_url(
|
||||
hass: HomeAssistant, otbr_config_entry_multipan
|
||||
) -> None:
|
||||
"""Test async_change_channel when otbr is not configured."""
|
||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
||||
data.url = OTBR_NON_MULTIPAN_URL
|
||||
with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel:
|
||||
await otbr_silabs_multiprotocol.async_change_channel(hass, 16, delay=0)
|
||||
mock_set_channel.assert_not_awaited()
|
||||
|
||||
|
||||
async def test_async_get_channel(
|
||||
hass: HomeAssistant, otbr_config_entry_multipan
|
||||
) -> None:
|
||||
@ -173,6 +184,17 @@ async def test_async_get_channel_no_otbr(hass: HomeAssistant) -> None:
|
||||
mock_get_active_dataset.assert_not_awaited()
|
||||
|
||||
|
||||
async def test_async_get_channel_non_matching_url(
|
||||
hass: HomeAssistant, otbr_config_entry_multipan
|
||||
) -> None:
|
||||
"""Test async_change_channel when otbr is not configured."""
|
||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
||||
data.url = OTBR_NON_MULTIPAN_URL
|
||||
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
|
||||
mock_get_active_dataset.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("url", "expected"),
|
||||
[(OTBR_MULTIPAN_URL, True), (OTBR_NON_MULTIPAN_URL, False)],
|
||||
@ -191,3 +213,12 @@ async def test_async_using_multipan_no_otbr(hass: HomeAssistant) -> None:
|
||||
"""Test async_change_channel when otbr is not configured."""
|
||||
|
||||
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is False
|
||||
|
||||
|
||||
async def test_async_using_multipan_non_matching_url(
|
||||
hass: HomeAssistant, otbr_config_entry_multipan
|
||||
) -> None:
|
||||
"""Test async_change_channel when otbr is not configured."""
|
||||
data: otbr.OTBRData = hass.data[otbr.DOMAIN]
|
||||
data.url = OTBR_NON_MULTIPAN_URL
|
||||
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is False
|
||||
|
Loading…
x
Reference in New Issue
Block a user