diff --git a/homeassistant/components/otbr/silabs_multiprotocol.py b/homeassistant/components/otbr/silabs_multiprotocol.py index bd7eb9975580df6ca24c487f988363dbae893d31..411da0a9361657c51f97ab0e15d3845a2dd4db9e 100644 --- a/homeassistant/components/otbr/silabs_multiprotocol.py +++ b/homeassistant/components/otbr/silabs_multiprotocol.py @@ -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 diff --git a/tests/components/otbr/test_silabs_multiprotocol.py b/tests/components/otbr/test_silabs_multiprotocol.py index dcb0c10b9e4b44b648d9cf0c615d5b5281f00319..ad3d6540b47d4c9b35498b7eac1e360f4e175995 100644 --- a/tests/components/otbr/test_silabs_multiprotocol.py +++ b/tests/components/otbr/test_silabs_multiprotocol.py @@ -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