Prepare otbr.silabs_multiprotocol for multiple config entries

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 @@
 _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

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