diff --git a/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py b/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py index 8c502f080f6..34ab9a3cedb 100644 --- a/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py +++ b/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py @@ -5,7 +5,7 @@ from abc import ABC, abstractmethod import asyncio import dataclasses import logging -from typing import Any +from typing import Any, Protocol import voluptuous as vol import yarl @@ -19,12 +19,19 @@ from homeassistant.components.hassio import ( hostname_from_addon_slug, is_hassio, ) -from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN -from homeassistant.components.zha.radio_manager import ZhaMultiPANMigrationHelper from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import AbortFlow, FlowResult from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.integration_platform import ( + async_process_integration_platforms, +) +from homeassistant.helpers.selector import ( + SelectSelector, + SelectSelectorConfig, + SelectSelectorMode, +) from homeassistant.helpers.singleton import singleton +from homeassistant.helpers.storage import Store from .const import LOGGER, SILABS_MULTIPROTOCOL_ADDON_SLUG @@ -39,17 +46,144 @@ CONF_ADDON_AUTOFLASH_FW = "autoflash_firmware" CONF_ADDON_DEVICE = "device" CONF_ENABLE_MULTI_PAN = "enable_multi_pan" +DEFAULT_CHANNEL = 15 +DEFAULT_CHANNEL_CHANGE_DELAY = 5 * 60 # Thread recommendation + +STORAGE_KEY = "homeassistant_hardware.silabs" +STORAGE_VERSION_MAJOR = 1 +STORAGE_VERSION_MINOR = 1 +SAVE_DELAY = 10 + @singleton(DATA_ADDON_MANAGER) -@callback -def get_addon_manager(hass: HomeAssistant) -> AddonManager: +async def get_addon_manager(hass: HomeAssistant) -> MultiprotocolAddonManager: """Get the add-on manager.""" - return AddonManager( - hass, - LOGGER, - "Silicon Labs Multiprotocol", - SILABS_MULTIPROTOCOL_ADDON_SLUG, - ) + manager = MultiprotocolAddonManager(hass) + await manager.async_setup() + return manager + + +class MultiprotocolAddonManager(AddonManager): + """Silicon Labs Multiprotocol add-on manager.""" + + def __init__(self, hass: HomeAssistant) -> None: + """Initialize the manager.""" + super().__init__( + hass, + LOGGER, + "Silicon Labs Multiprotocol", + SILABS_MULTIPROTOCOL_ADDON_SLUG, + ) + self._channel: int | None = None + self._platforms: dict[str, MultipanProtocol] = {} + self._store: Store[dict[str, Any]] = Store( + hass, + STORAGE_VERSION_MAJOR, + STORAGE_KEY, + atomic_writes=True, + minor_version=STORAGE_VERSION_MINOR, + ) + + async def async_setup(self) -> None: + """Set up the manager.""" + await async_process_integration_platforms( + self._hass, "silabs_multiprotocol", self._register_multipan_platform + ) + await self.async_load() + + async def _register_multipan_platform( + self, hass: HomeAssistant, integration_domain: str, platform: MultipanProtocol + ) -> None: + """Register a multipan platform.""" + self._platforms[integration_domain] = platform + if self._channel is not None or not await platform.async_using_multipan(hass): + return + + new_channel = await platform.async_get_channel(hass) + if new_channel is None: + return + + _LOGGER.info( + "Setting multipan channel to %s (source: '%s')", + new_channel, + integration_domain, + ) + self.async_set_channel(new_channel) + + async def async_change_channel( + self, channel: int, delay: float + ) -> list[asyncio.Task]: + """Change the channel and notify platforms.""" + self.async_set_channel(channel) + + tasks = [] + + for platform in self._platforms.values(): + if not await platform.async_using_multipan(self._hass): + continue + task = await platform.async_change_channel(self._hass, channel, delay) + if not task: + continue + tasks.append(task) + + return tasks + + @callback + def async_get_channel(self) -> int | None: + """Get the channel.""" + return self._channel + + @callback + def async_set_channel(self, channel: int) -> None: + """Set the channel without notifying platforms. + + This must only be called when first initializing the manager. + """ + self._channel = channel + self.async_schedule_save() + + async def async_load(self) -> None: + """Load the store.""" + data = await self._store.async_load() + + if data is not None: + self._channel = data["channel"] + + @callback + def async_schedule_save(self) -> None: + """Schedule saving the store.""" + self._store.async_delay_save(self._data_to_save, SAVE_DELAY) + + @callback + def _data_to_save(self) -> dict[str, list[dict[str, str | None]]]: + """Return data to store in a file.""" + data: dict[str, Any] = {} + data["channel"] = self._channel + return data + + +class MultipanProtocol(Protocol): + """Define the format of multipan platforms.""" + + async def async_change_channel( + self, hass: HomeAssistant, channel: int, delay: float + ) -> asyncio.Task | None: + """Set the channel to be used. + + Does nothing if not configured or the multiprotocol add-on is not used. + """ + + async def async_get_channel(self, hass: HomeAssistant) -> int | None: + """Return the channel. + + Returns None if not configured or the multiprotocol add-on is not used. + """ + + async def async_using_multipan(self, hass: HomeAssistant) -> bool: + """Return if the multiprotocol device is used. + + Returns False if not configured. + """ @dataclasses.dataclass @@ -82,6 +216,11 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): def __init__(self, config_entry: config_entries.ConfigEntry) -> None: """Set up the options flow.""" + # pylint: disable-next=import-outside-toplevel + from homeassistant.components.zha.radio_manager import ( + ZhaMultiPANMigrationHelper, + ) + # If we install the add-on we should uninstall it on entry remove. self.install_task: asyncio.Task | None = None self.start_task: asyncio.Task | None = None @@ -117,7 +256,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): async def _async_get_addon_info(self) -> AddonInfo: """Return and cache Silicon Labs Multiprotocol add-on info.""" - addon_manager: AddonManager = get_addon_manager(self.hass) + addon_manager: AddonManager = await get_addon_manager(self.hass) try: addon_info: AddonInfo = await addon_manager.async_get_addon_info() except AddonError as err: @@ -128,7 +267,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): async def _async_set_addon_config(self, config: dict) -> None: """Set Silicon Labs Multiprotocol add-on config.""" - addon_manager: AddonManager = get_addon_manager(self.hass) + addon_manager: AddonManager = await get_addon_manager(self.hass) try: await addon_manager.async_set_addon_options(config) except AddonError as err: @@ -137,7 +276,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): async def _async_install_addon(self) -> None: """Install the Silicon Labs Multiprotocol add-on.""" - addon_manager: AddonManager = get_addon_manager(self.hass) + addon_manager: AddonManager = await get_addon_manager(self.hass) try: await addon_manager.async_schedule_install_addon() finally: @@ -213,6 +352,19 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Configure the Silicon Labs Multiprotocol add-on.""" + # pylint: disable-next=import-outside-toplevel + from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN + + # pylint: disable-next=import-outside-toplevel + from homeassistant.components.zha.radio_manager import ( + ZhaMultiPANMigrationHelper, + ) + + # pylint: disable-next=import-outside-toplevel + from homeassistant.components.zha.silabs_multiprotocol import ( + async_get_channel as async_get_zha_channel, + ) + addon_info = await self._async_get_addon_info() addon_config = addon_info.options @@ -224,6 +376,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): **dataclasses.asdict(serial_port_settings), } + multipan_channel = DEFAULT_CHANNEL + # Initiate ZHA migration zha_entries = self.hass.config_entries.async_entries(ZHA_DOMAIN) @@ -247,6 +401,13 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): _LOGGER.exception("Unexpected exception during ZHA migration") raise AbortFlow("zha_migration_failed") from err + if (zha_channel := await async_get_zha_channel(self.hass)) is not None: + multipan_channel = zha_channel + + # Initialize the shared channel + multipan_manager = await get_addon_manager(self.hass) + multipan_manager.async_set_channel(multipan_channel) + if new_addon_config != addon_config: # Copy the add-on config to keep the objects separate. self.original_addon_config = dict(addon_config) @@ -283,7 +444,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): async def _async_start_addon(self) -> None: """Start Silicon Labs Multiprotocol add-on.""" - addon_manager: AddonManager = get_addon_manager(self.hass) + addon_manager: AddonManager = await get_addon_manager(self.hass) try: await addon_manager.async_schedule_start_addon() finally: @@ -319,9 +480,73 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): serial_device = (await self._async_serial_port_settings()).device if addon_info.options.get(CONF_ADDON_DEVICE) == serial_device: - return await self.async_step_show_revert_guide() + return await self.async_step_show_addon_menu() return await self.async_step_addon_installed_other_device() + async def async_step_show_addon_menu( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Show menu options for the addon.""" + return self.async_show_menu( + step_id="addon_menu", + menu_options=[ + "reconfigure_addon", + "uninstall_addon", + ], + ) + + async def async_step_reconfigure_addon( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Reconfigure the addon.""" + multipan_manager = await get_addon_manager(self.hass) + + if user_input is None: + channels = [str(x) for x in range(11, 27)] + suggested_channel = DEFAULT_CHANNEL + if (channel := multipan_manager.async_get_channel()) is not None: + suggested_channel = channel + data_schema = vol.Schema( + { + vol.Required( + "channel", + description={"suggested_value": str(suggested_channel)}, + ): SelectSelector( + SelectSelectorConfig( + options=channels, mode=SelectSelectorMode.DROPDOWN + ) + ) + } + ) + return self.async_show_form( + step_id="reconfigure_addon", data_schema=data_schema + ) + + # Change the shared channel + await multipan_manager.async_change_channel( + int(user_input["channel"]), DEFAULT_CHANNEL_CHANGE_DELAY + ) + return await self.async_step_notify_channel_change() + + async def async_step_notify_channel_change( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Notify that the channel change will take about five minutes.""" + if user_input is None: + return self.async_show_form( + step_id="notify_channel_change", + description_placeholders={ + "delay_minutes": str(DEFAULT_CHANNEL_CHANGE_DELAY // 60) + }, + ) + return self.async_create_entry(title="", data={}) + + async def async_step_uninstall_addon( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Uninstall the addon (not implemented).""" + return await self.async_step_show_revert_guide() + async def async_step_show_revert_guide( self, user_input: dict[str, Any] | None = None ) -> FlowResult: @@ -348,7 +573,7 @@ async def check_multi_pan_addon(hass: HomeAssistant) -> None: if not is_hassio(hass): return - addon_manager: AddonManager = get_addon_manager(hass) + addon_manager: AddonManager = await get_addon_manager(hass) try: addon_info: AddonInfo = await addon_manager.async_get_addon_info() except AddonError as err: @@ -375,7 +600,7 @@ async def multi_pan_addon_using_device(hass: HomeAssistant, device_path: str) -> if not is_hassio(hass): return False - addon_manager: AddonManager = get_addon_manager(hass) + addon_manager: AddonManager = await get_addon_manager(hass) addon_info: AddonInfo = await addon_manager.async_get_addon_info() if addon_info.state != AddonState.RUNNING: diff --git a/homeassistant/components/homeassistant_hardware/strings.json b/homeassistant/components/homeassistant_hardware/strings.json index 47549794fc8..60501397557 100644 --- a/homeassistant/components/homeassistant_hardware/strings.json +++ b/homeassistant/components/homeassistant_hardware/strings.json @@ -12,15 +12,34 @@ "addon_installed_other_device": { "title": "Multiprotocol support is already enabled for another device" }, + "addon_menu": { + "menu_options": { + "reconfigure_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]", + "uninstall_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" + } + }, "install_addon": { "title": "The Silicon Labs Multiprotocol add-on installation has started" }, + "notify_channel_change": { + "title": "Channel change initiated", + "description": "A Zigbee and Thread channel change has been initiated and will finish in {delay_minutes} minutes." + }, + "reconfigure_addon": { + "title": "Reconfigure IEEE 802.15.4 radio multiprotocol support", + "data": { + "channel": "Channel" + } + }, "show_revert_guide": { "title": "Multiprotocol support is enabled for this device", "description": "If you want to change to Zigbee only firmware, please complete the following manual steps:\n\n * Remove the Silicon Labs Multiprotocol addon\n\n * Flash the Zigbee only firmware, follow the guide at https://github.com/NabuCasa/silabs-firmware/wiki/Flash-Silicon-Labs-radio-firmware-manually.\n\n * Reconfigure ZHA to migrate settings to the reflashed radio" }, "start_addon": { "title": "The Silicon Labs Multiprotocol add-on is starting." + }, + "uninstall_addon": { + "title": "Remove IEEE 802.15.4 radio multiprotocol support." } }, "error": { diff --git a/homeassistant/components/homeassistant_sky_connect/strings.json b/homeassistant/components/homeassistant_sky_connect/strings.json index 970f9d97a4c..415df2092a1 100644 --- a/homeassistant/components/homeassistant_sky_connect/strings.json +++ b/homeassistant/components/homeassistant_sky_connect/strings.json @@ -11,15 +11,34 @@ "addon_installed_other_device": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::addon_installed_other_device::title%]" }, + "addon_menu": { + "menu_options": { + "reconfigure_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::addon_menu::menu_options::reconfigure_addon%]", + "uninstall_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::addon_menu::menu_options::uninstall_addon%]" + } + }, "install_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_addon::title%]" }, + "notify_channel_change": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::notify_channel_change::title%]", + "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::notify_channel_change::description%]" + }, + "reconfigure_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]", + "data": { + "channel": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::data::channel%]" + } + }, "show_revert_guide": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::title%]", "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::description%]" }, "start_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]" + }, + "uninstall_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" } }, "error": { diff --git a/homeassistant/components/homeassistant_yellow/strings.json b/homeassistant/components/homeassistant_yellow/strings.json index d97b01c7c84..c1069a7e755 100644 --- a/homeassistant/components/homeassistant_yellow/strings.json +++ b/homeassistant/components/homeassistant_yellow/strings.json @@ -11,6 +11,12 @@ "addon_installed_other_device": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::addon_installed_other_device::title%]" }, + "addon_menu": { + "menu_options": { + "reconfigure_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::addon_menu::menu_options::reconfigure_addon%]", + "uninstall_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::addon_menu::menu_options::uninstall_addon%]" + } + }, "hardware_settings": { "title": "Configure hardware settings", "data": { @@ -22,6 +28,10 @@ "install_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_addon::title%]" }, + "notify_channel_change": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::notify_channel_change::title%]", + "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::notify_channel_change::description%]" + }, "main_menu": { "menu_options": { "hardware_settings": "[%key:component::homeassistant_yellow::options::step::hardware_settings::title%]", @@ -36,12 +46,21 @@ "reboot_now": "Reboot now" } }, + "reconfigure_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]", + "data": { + "channel": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::data::channel%]" + } + }, "show_revert_guide": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::title%]", "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::description%]" }, "start_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]" + }, + "uninstall_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" } }, "error": { diff --git a/homeassistant/components/otbr/silabs_multiprotocol.py b/homeassistant/components/otbr/silabs_multiprotocol.py new file mode 100644 index 00000000000..9a462c4610b --- /dev/null +++ b/homeassistant/components/otbr/silabs_multiprotocol.py @@ -0,0 +1,87 @@ +"""Silicon Labs Multiprotocol support.""" + +from __future__ import annotations + +import asyncio +import logging + +import aiohttp +from python_otbr_api import tlv_parser +from python_otbr_api.tlv_parser import MeshcopTLVType + +from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( + is_multiprotocol_url, +) +from homeassistant.components.thread import async_add_dataset +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError + +from . import DOMAIN +from .util import OTBRData + +_LOGGER = logging.getLogger(__name__) + + +async def async_change_channel(hass: HomeAssistant, 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 + dataset_tlvs = await data.get_pending_dataset_tlvs() + if dataset_tlvs is None: + # The activation timer may have expired already + dataset_tlvs = await data.get_active_dataset_tlvs() + if dataset_tlvs is None: + # Don't try to import a None dataset + return + + dataset = tlv_parser.parse_tlv(dataset_tlvs.hex()) + dataset.pop(MeshcopTLVType.DELAYTIMER, None) + dataset.pop(MeshcopTLVType.PENDINGTIMESTAMP, None) + dataset_tlvs_str = tlv_parser.encode_tlv(dataset) + await async_add_dataset(hass, DOMAIN, dataset_tlvs_str) + + +async def async_get_channel(hass: HomeAssistant) -> 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 ( + HomeAssistantError, + aiohttp.ClientError, + asyncio.TimeoutError, + ) as err: + _LOGGER.warning("Failed to communicate with OTBR %s", err) + return None + + if dataset is None: + return None + + return dataset.channel + + +async def async_using_multipan(hass: HomeAssistant) -> 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) diff --git a/homeassistant/components/otbr/util.py b/homeassistant/components/otbr/util.py index 5541ecb6874..5caebba5eb5 100644 --- a/homeassistant/components/otbr/util.py +++ b/homeassistant/components/otbr/util.py @@ -2,22 +2,22 @@ from __future__ import annotations from collections.abc import Callable, Coroutine -import contextlib import dataclasses from functools import wraps from typing import Any, Concatenate, ParamSpec, TypeVar, cast import python_otbr_api -from python_otbr_api import tlv_parser +from python_otbr_api import PENDING_DATASET_DELAY_TIMER, tlv_parser from python_otbr_api.pskc import compute_pskc from python_otbr_api.tlv_parser import MeshcopTLVType from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( + MultiprotocolAddonManager, + get_addon_manager, is_multiprotocol_url, multi_pan_addon_using_device, ) from homeassistant.components.homeassistant_yellow import RADIO_DEVICE as YELLOW_RADIO -from homeassistant.components.zha import api as zha_api from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import issue_registry as ir @@ -73,11 +73,21 @@ class OTBRData: """Enable or disable the router.""" return await self.api.set_enabled(enabled) + @_handle_otbr_error + async def get_active_dataset(self) -> python_otbr_api.ActiveDataSet | None: + """Get current active operational dataset, or None.""" + return await self.api.get_active_dataset() + @_handle_otbr_error async def get_active_dataset_tlvs(self) -> bytes | None: """Get current active operational dataset in TLVS format, or None.""" return await self.api.get_active_dataset_tlvs() + @_handle_otbr_error + async def get_pending_dataset_tlvs(self) -> bytes | None: + """Get current pending operational dataset in TLVS format, or None.""" + return await self.api.get_pending_dataset_tlvs() + @_handle_otbr_error async def create_active_dataset( self, dataset: python_otbr_api.ActiveDataSet @@ -90,43 +100,27 @@ class OTBRData: """Set current active operational dataset in TLVS format.""" await self.api.set_active_dataset_tlvs(dataset) + @_handle_otbr_error + async def set_channel( + self, channel: int, delay: float = PENDING_DATASET_DELAY_TIMER / 1000 + ) -> None: + """Set current channel.""" + await self.api.set_channel(channel, delay=int(delay * 1000)) + @_handle_otbr_error async def get_extended_address(self) -> bytes: """Get extended address (EUI-64).""" return await self.api.get_extended_address() -def _get_zha_url(hass: HomeAssistant) -> str | None: - """Get ZHA radio path, or None if there's no ZHA config entry.""" - with contextlib.suppress(ValueError): - return zha_api.async_get_radio_path(hass) - return None - - -async def _get_zha_channel(hass: HomeAssistant) -> int | None: - """Get ZHA channel, or None if there's no ZHA config entry.""" - zha_network_settings: zha_api.NetworkBackup | None - with contextlib.suppress(ValueError): - zha_network_settings = await zha_api.async_get_network_settings(hass) - if not zha_network_settings: - return None - channel: int = zha_network_settings.network_info.channel - # ZHA uses channel 0 when no channel is set - return channel or None - - async def get_allowed_channel(hass: HomeAssistant, otbr_url: str) -> int | None: """Return the allowed channel, or None if there's no restriction.""" if not is_multiprotocol_url(otbr_url): # The OTBR is not sharing the radio, no restriction return None - zha_url = _get_zha_url(hass) - if not zha_url or not is_multiprotocol_url(zha_url): - # ZHA is not configured or not sharing the radio with this OTBR, no restriction - return None - - return await _get_zha_channel(hass) + addon_manager: MultiprotocolAddonManager = await get_addon_manager(hass) + return addon_manager.async_get_channel() async def _warn_on_channel_collision( diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index d0f124a0838..91bc2ac42a2 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -96,7 +96,7 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]: yellow_radio.manufacturer = "Nabu Casa" # Present the multi-PAN addon as a setup option, if it's available - addon_manager = silabs_multiprotocol_addon.get_addon_manager(hass) + addon_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) try: addon_info = await addon_manager.async_get_addon_info() diff --git a/homeassistant/components/zha/silabs_multiprotocol.py b/homeassistant/components/zha/silabs_multiprotocol.py new file mode 100644 index 00000000000..aec52b4ac75 --- /dev/null +++ b/homeassistant/components/zha/silabs_multiprotocol.py @@ -0,0 +1,81 @@ +"""Silicon Labs Multiprotocol support.""" + +from __future__ import annotations + +import asyncio +import contextlib + +from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( + is_multiprotocol_url, +) +from homeassistant.core import HomeAssistant + +from . import api + +# The approximate time it takes ZHA to change channels on SiLabs coordinators +ZHA_CHANNEL_CHANGE_TIME_S = 10.27 + + +def _get_zha_url(hass: HomeAssistant) -> str | None: + """Return the ZHA radio path, or None if there's no ZHA config entry.""" + with contextlib.suppress(ValueError): + return api.async_get_radio_path(hass) + return None + + +async def _get_zha_channel(hass: HomeAssistant) -> int | None: + """Get ZHA channel, or None if there's no ZHA config entry.""" + zha_network_settings: api.NetworkBackup | None + with contextlib.suppress(ValueError): + zha_network_settings = await api.async_get_network_settings(hass) + if not zha_network_settings: + return None + channel: int = zha_network_settings.network_info.channel + # ZHA uses channel 0 when no channel is set + return channel or None + + +async def async_change_channel( + hass: HomeAssistant, channel: int, delay: float = 0 +) -> asyncio.Task | None: + """Set the channel to be used. + + Does nothing if not configured. + """ + zha_url = _get_zha_url(hass) + if not zha_url: + # ZHA is not configured + return None + + async def finish_migration() -> None: + """Finish the channel migration.""" + await asyncio.sleep(max(0, delay - ZHA_CHANNEL_CHANGE_TIME_S)) + return await api.async_change_channel(hass, channel) + + return hass.async_create_task(finish_migration()) + + +async def async_get_channel(hass: HomeAssistant) -> int | None: + """Return the channel. + + Returns None if not configured. + """ + zha_url = _get_zha_url(hass) + if not zha_url: + # ZHA is not configured + return None + + return await _get_zha_channel(hass) + + +async def async_using_multipan(hass: HomeAssistant) -> bool: + """Return if the multiprotocol device is used. + + Returns False if not configured. + """ + zha_url = _get_zha_url(hass) + if not zha_url: + # ZHA is not configured + return False + + return is_multiprotocol_url(zha_url) diff --git a/tests/components/homeassistant_hardware/conftest.py b/tests/components/homeassistant_hardware/conftest.py index 4add48781a9..60c766c7204 100644 --- a/tests/components/homeassistant_hardware/conftest.py +++ b/tests/components/homeassistant_hardware/conftest.py @@ -1,7 +1,7 @@ """Test fixtures for the Home Assistant Hardware integration.""" from collections.abc import Generator from typing import Any -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -32,6 +32,17 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]: yield +@pytest.fixture(autouse=True) +def mock_zha_get_last_network_settings() -> Generator[None, None, None]: + """Mock zha.api.async_get_last_network_settings.""" + + with patch( + "homeassistant.components.zha.api.async_get_last_network_settings", + AsyncMock(return_value=None), + ): + yield + + @pytest.fixture(name="addon_running") def mock_addon_running(addon_store_info, addon_info): """Mock add-on already running.""" diff --git a/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py b/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py index a195899136d..83702adcc3a 100644 --- a/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py +++ b/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py @@ -11,12 +11,16 @@ from homeassistant.components.hassio.handler import HassioAPIError from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon from homeassistant.components.zha.core.const import DOMAIN as ZHA_DOMAIN from homeassistant.config_entries import ConfigEntry, ConfigFlow +from homeassistant.const import EVENT_COMPONENT_LOADED from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResult, FlowResultType +from homeassistant.setup import ATTR_COMPONENT from tests.common import ( MockConfigEntry, MockModule, + MockPlatform, + flush_store, mock_config_flow, mock_integration, mock_platform, @@ -96,6 +100,54 @@ def config_flow_handler( yield +class MockMultiprotocolPlatform(MockPlatform): + """A mock multiprotocol platform.""" + + channel = 15 + using_multipan = True + + def __init__(self, **kwargs: Any) -> None: + """Initialize.""" + super().__init__(**kwargs) + self.change_channel_calls = [] + + async def async_change_channel( + self, hass: HomeAssistant, channel: int, delay: float + ) -> None: + """Set the channel to be used.""" + self.change_channel_calls.append((channel, delay)) + + async def async_get_channel(self, hass: HomeAssistant) -> int | None: + """Return the channel.""" + return self.channel + + async def async_using_multipan(self, hass: HomeAssistant) -> bool: + """Return if the multiprotocol device is used.""" + return self.using_multipan + + +@pytest.fixture +def mock_multiprotocol_platform( + hass: HomeAssistant, +) -> Generator[FakeConfigFlow, None, None]: + """Fixture for a test silabs multiprotocol platform.""" + hass.config.components.add(TEST_DOMAIN) + platform = MockMultiprotocolPlatform() + mock_platform(hass, f"{TEST_DOMAIN}.silabs_multiprotocol", platform) + return platform + + +def get_suggested(schema, key): + """Get suggested value for key in voluptuous schema.""" + for k in schema: + if k == key: + if k.description is None or "suggested_value" not in k.description: + return None + return k.description["suggested_value"] + # Wanted key absent from schema + raise Exception + + async def test_option_flow_install_multi_pan_addon( hass: HomeAssistant, addon_store_info, @@ -215,7 +267,13 @@ async def test_option_flow_install_multi_pan_addon_zha( assert result["step_id"] == "configure_addon" install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") - result = await hass.config_entries.options.async_configure(result["flow_id"]) + multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + assert multipan_manager._channel is None + with patch( + "homeassistant.components.zha.silabs_multiprotocol.async_get_channel", + return_value=11, + ): + result = await hass.config_entries.options.async_configure(result["flow_id"]) assert result["type"] == FlowResultType.SHOW_PROGRESS assert result["step_id"] == "start_addon" set_addon_options.assert_called_once_with( @@ -230,6 +288,8 @@ async def test_option_flow_install_multi_pan_addon_zha( } }, ) + # Check the channel is initialized from ZHA + assert multipan_manager._channel == 11 # Check the ZHA config entry data is updated assert zha_config_entry.data == { "device": { @@ -393,7 +453,64 @@ async def test_option_flow_addon_installed_other_device( assert result["type"] == FlowResultType.CREATE_ENTRY -async def test_option_flow_addon_installed_same_device( +@pytest.mark.parametrize( + ("configured_channel", "suggested_channel"), [(None, "15"), (11, "11")] +) +async def test_option_flow_addon_installed_same_device_reconfigure( + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + mock_multiprotocol_platform: MockMultiprotocolPlatform, + configured_channel: int | None, + suggested_channel: int, +) -> None: + """Test installing the multi pan addon.""" + mock_integration(hass, MockModule("hassio")) + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager._channel = configured_channel + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + side_effect=Mock(return_value=True), + ): + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "reconfigure_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "reconfigure_addon" + assert get_suggested(result["data_schema"].schema, "channel") == suggested_channel + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {"channel": "14"} + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "notify_channel_change" + assert result["description_placeholders"] == {"delay_minutes": "5"} + + result = await hass.config_entries.options.async_configure(result["flow_id"], {}) + assert result["type"] == FlowResultType.CREATE_ENTRY + + assert mock_multiprotocol_platform.change_channel_calls == [(14, 300)] + + +async def test_option_flow_addon_installed_same_device_uninstall( hass: HomeAssistant, addon_info, addon_store_info, @@ -417,8 +534,15 @@ async def test_option_flow_addon_installed_same_device( side_effect=Mock(return_value=True), ): result = await hass.config_entries.options.async_init(config_entry.entry_id) - assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "show_revert_guide" + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "show_revert_guide" result = await hass.config_entries.options.async_configure(result["flow_id"], {}) assert result["type"] == FlowResultType.CREATE_ENTRY @@ -806,3 +930,80 @@ def test_is_multiprotocol_url() -> None: "http://core-silabs-multiprotocol:8081" ) assert not silabs_multiprotocol_addon.is_multiprotocol_url("/dev/ttyAMA1") + + +@pytest.mark.parametrize( + ( + "initial_multipan_channel", + "platform_using_multipan", + "platform_channel", + "new_multipan_channel", + ), + [ + (None, True, 15, 15), + (None, False, 15, None), + (11, True, 15, 11), + (None, True, None, None), + ], +) +async def test_import_channel( + hass: HomeAssistant, + initial_multipan_channel: int | None, + platform_using_multipan: bool, + platform_channel: int | None, + new_multipan_channel: int | None, +) -> None: + """Test channel is initialized from first platform.""" + multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager._channel = initial_multipan_channel + + mock_multiprotocol_platform = MockMultiprotocolPlatform() + mock_multiprotocol_platform.channel = platform_channel + mock_multiprotocol_platform.using_multipan = platform_using_multipan + + hass.config.components.add(TEST_DOMAIN) + mock_platform( + hass, f"{TEST_DOMAIN}.silabs_multiprotocol", mock_multiprotocol_platform + ) + hass.bus.async_fire(EVENT_COMPONENT_LOADED, {ATTR_COMPONENT: TEST_DOMAIN}) + await hass.async_block_till_done() + + assert multipan_manager.async_get_channel() == new_multipan_channel + + +@pytest.mark.parametrize( + ( + "platform_using_multipan", + "expected_calls", + ), + [ + (True, [(15, 10)]), + (False, []), + ], +) +async def test_change_channel( + hass: HomeAssistant, + mock_multiprotocol_platform: MockMultiprotocolPlatform, + platform_using_multipan: bool, + expected_calls: list[int], +) -> None: + """Test channel is initialized from first platform.""" + multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + mock_multiprotocol_platform.using_multipan = platform_using_multipan + + await multipan_manager.async_change_channel(15, 10) + assert mock_multiprotocol_platform.change_channel_calls == expected_calls + + +async def test_load_preferences(hass: HomeAssistant) -> None: + """Make sure that we can load/save data correctly.""" + multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + assert multipan_manager._channel != 11 + multipan_manager.async_set_channel(11) + + await flush_store(multipan_manager._store) + + multipan_manager2 = silabs_multiprotocol_addon.MultiprotocolAddonManager(hass) + await multipan_manager2.async_setup() + + assert multipan_manager._channel == multipan_manager2._channel diff --git a/tests/components/homeassistant_sky_connect/conftest.py b/tests/components/homeassistant_sky_connect/conftest.py index 7fcc1f86880..3677b4ea8f1 100644 --- a/tests/components/homeassistant_sky_connect/conftest.py +++ b/tests/components/homeassistant_sky_connect/conftest.py @@ -1,6 +1,6 @@ """Test fixtures for the Home Assistant SkyConnect integration.""" from collections.abc import Generator -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -34,6 +34,17 @@ def mock_zha(): yield +@pytest.fixture(autouse=True) +def mock_zha_get_last_network_settings() -> Generator[None, None, None]: + """Mock zha.api.async_get_last_network_settings.""" + + with patch( + "homeassistant.components.zha.api.async_get_last_network_settings", + AsyncMock(return_value=None), + ): + yield + + @pytest.fixture(name="addon_running") def mock_addon_running(addon_store_info, addon_info): """Mock add-on already running.""" diff --git a/tests/components/homeassistant_yellow/conftest.py b/tests/components/homeassistant_yellow/conftest.py index bc48c6b01fd..e4a666f9f04 100644 --- a/tests/components/homeassistant_yellow/conftest.py +++ b/tests/components/homeassistant_yellow/conftest.py @@ -1,7 +1,7 @@ """Test fixtures for the Home Assistant Yellow integration.""" from collections.abc import Generator from typing import Any -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -32,6 +32,17 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]: yield +@pytest.fixture(autouse=True) +def mock_zha_get_last_network_settings() -> Generator[None, None, None]: + """Mock zha.api.async_get_last_network_settings.""" + + with patch( + "homeassistant.components.zha.api.async_get_last_network_settings", + AsyncMock(return_value=None), + ): + yield + + @pytest.fixture(name="addon_running") def mock_addon_running(addon_store_info, addon_info): """Mock add-on already running.""" diff --git a/tests/components/otbr/conftest.py b/tests/components/otbr/conftest.py index f0b3ca0a18d..bb3b474519e 100644 --- a/tests/components/otbr/conftest.py +++ b/tests/components/otbr/conftest.py @@ -1,9 +1,10 @@ """Test fixtures for the Open Thread Border Router integration.""" -from unittest.mock import patch +from unittest.mock import Mock, patch import pytest from homeassistant.components import otbr +from homeassistant.core import HomeAssistant from . import CONFIG_ENTRY_DATA, DATASET_CH16 @@ -31,3 +32,12 @@ async def otbr_config_entry_fixture(hass): @pytest.fixture(autouse=True) def use_mocked_zeroconf(mock_async_zeroconf): """Mock zeroconf in all tests.""" + + +@pytest.fixture(name="multiprotocol_addon_manager_mock") +def multiprotocol_addon_manager_mock_fixture(hass: HomeAssistant): + """Mock the Silicon Labs Multiprotocol add-on manager.""" + mock_manager = Mock() + mock_manager.async_get_channel = Mock(return_value=None) + with patch.dict(hass.data, {"silabs_multiprotocol_addon_manager": mock_manager}): + yield mock_manager diff --git a/tests/components/otbr/test_config_flow.py b/tests/components/otbr/test_config_flow.py index faec90282df..cfb47a28bcf 100644 --- a/tests/components/otbr/test_config_flow.py +++ b/tests/components/otbr/test_config_flow.py @@ -2,7 +2,7 @@ import asyncio from http import HTTPStatus from typing import Any -from unittest.mock import Mock, patch +from unittest.mock import patch import aiohttp import pytest @@ -309,7 +309,9 @@ async def test_hassio_discovery_flow_router_not_setup_has_preferred( async def test_hassio_discovery_flow_router_not_setup_has_preferred_2( - hass: HomeAssistant, aioclient_mock: AiohttpClientMocker + hass: HomeAssistant, + aioclient_mock: AiohttpClientMocker, + multiprotocol_addon_manager_mock, ) -> None: """Test the hassio discovery flow when the border router has no dataset. @@ -321,8 +323,7 @@ async def test_hassio_discovery_flow_router_not_setup_has_preferred_2( aioclient_mock.put(f"{url}/node/dataset/active", status=HTTPStatus.CREATED) aioclient_mock.put(f"{url}/node/state", status=HTTPStatus.OK) - networksettings = Mock() - networksettings.network_info.channel = 15 + multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 with patch( "homeassistant.components.otbr.config_flow.async_get_preferred_dataset", @@ -330,13 +331,7 @@ async def test_hassio_discovery_flow_router_not_setup_has_preferred_2( ), patch( "homeassistant.components.otbr.async_setup_entry", return_value=True, - ) as mock_setup_entry, patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=networksettings, - ): + ) as mock_setup_entry: result = await hass.config_entries.flow.async_init( otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA ) diff --git a/tests/components/otbr/test_init.py b/tests/components/otbr/test_init.py index 3d646287ce1..990c015244f 100644 --- a/tests/components/otbr/test_init.py +++ b/tests/components/otbr/test_init.py @@ -1,7 +1,7 @@ """Test the Open Thread Border Router integration.""" import asyncio from http import HTTPStatus -from unittest.mock import ANY, AsyncMock, MagicMock, Mock, patch +from unittest.mock import ANY, AsyncMock, MagicMock, patch import aiohttp import pytest @@ -59,7 +59,9 @@ async def test_import_dataset(hass: HomeAssistant) -> None: ) -async def test_import_share_radio_channel_collision(hass: HomeAssistant) -> None: +async def test_import_share_radio_channel_collision( + hass: HomeAssistant, multiprotocol_addon_manager_mock +) -> None: """Test the active dataset is imported at setup. This imports a dataset with different channel than ZHA when ZHA and OTBR share @@ -67,8 +69,7 @@ async def test_import_share_radio_channel_collision(hass: HomeAssistant) -> None """ issue_registry = ir.async_get(hass) - networksettings = Mock() - networksettings.network_info.channel = 15 + multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 config_entry = MockConfigEntry( data=CONFIG_ENTRY_DATA, @@ -81,13 +82,7 @@ async def test_import_share_radio_channel_collision(hass: HomeAssistant) -> None "python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=DATASET_CH16 ), patch( "homeassistant.components.thread.dataset_store.DatasetStore.async_add" - ) as mock_add, patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=networksettings, - ): + ) as mock_add: assert await hass.config_entries.async_setup(config_entry.entry_id) mock_add.assert_called_once_with(otbr.DOMAIN, DATASET_CH16.hex()) @@ -99,7 +94,7 @@ async def test_import_share_radio_channel_collision(hass: HomeAssistant) -> None @pytest.mark.parametrize("dataset", [DATASET_CH15, DATASET_NO_CHANNEL]) async def test_import_share_radio_no_channel_collision( - hass: HomeAssistant, dataset: bytes + hass: HomeAssistant, multiprotocol_addon_manager_mock, dataset: bytes ) -> None: """Test the active dataset is imported at setup. @@ -107,8 +102,7 @@ async def test_import_share_radio_no_channel_collision( """ issue_registry = ir.async_get(hass) - networksettings = Mock() - networksettings.network_info.channel = 15 + multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 config_entry = MockConfigEntry( data=CONFIG_ENTRY_DATA, @@ -121,13 +115,7 @@ async def test_import_share_radio_no_channel_collision( "python_otbr_api.OTBR.get_active_dataset_tlvs", return_value=dataset ), patch( "homeassistant.components.thread.dataset_store.DatasetStore.async_add" - ) as mock_add, patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=networksettings, - ): + ) as mock_add: assert await hass.config_entries.async_setup(config_entry.entry_id) mock_add.assert_called_once_with(otbr.DOMAIN, dataset.hex()) diff --git a/tests/components/otbr/test_silabs_multiprotocol.py b/tests/components/otbr/test_silabs_multiprotocol.py new file mode 100644 index 00000000000..8dd07db6f22 --- /dev/null +++ b/tests/components/otbr/test_silabs_multiprotocol.py @@ -0,0 +1,175 @@ +"""Test OTBR Silicon Labs Multiprotocol support.""" +from unittest.mock import patch + +import pytest +from python_otbr_api import ActiveDataSet, tlv_parser + +from homeassistant.components import otbr +from homeassistant.components.otbr import ( + silabs_multiprotocol as otbr_silabs_multiprotocol, +) +from homeassistant.components.thread import dataset_store +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError + +from . import DATASET_CH16 + +OTBR_MULTIPAN_URL = "http://core-silabs-multiprotocol:8081" +OTBR_NON_MULTIPAN_URL = "/dev/ttyAMA1" +DATASET_CH16_PENDING = ( + "0E080000000000020000" # ACTIVETIMESTAMP + "340400006699" # DELAYTIMER + "000300000F" # CHANNEL + "35060004001FFFE0" # CHANNELMASK + "0208F642646DA209B1C0" # EXTPANID + "0708FDF57B5A0FE2AAF6" # MESHLOCALPREFIX + "0510DE98B5BA1A528FEE049D4B4B01835375" # NETWORKKEY + "030D4F70656E546872656164204841" # NETWORKNAME + "010225A4" # PANID + "0410F5DD18371BFD29E1A601EF6FFAD94C03" # PSKC + "0C0402A0F7F8" # SECURITYPOLICY +) + + +async def test_async_change_channel(hass: HomeAssistant, otbr_config_entry) -> None: + """Test test_async_change_channel.""" + + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 1 + assert list(store.datasets.values())[0].tlv == DATASET_CH16.hex() + + with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel, patch( + "python_otbr_api.OTBR.get_pending_dataset_tlvs", + return_value=bytes.fromhex(DATASET_CH16_PENDING), + ): + await otbr_silabs_multiprotocol.async_change_channel(hass, 15, delay=5 * 300) + mock_set_channel.assert_awaited_once_with(15, delay=5 * 300 * 1000) + + pending_dataset = tlv_parser.parse_tlv(DATASET_CH16_PENDING) + pending_dataset.pop(tlv_parser.MeshcopTLVType.DELAYTIMER) + + assert len(store.datasets) == 1 + assert list(store.datasets.values())[0].tlv == tlv_parser.encode_tlv( + pending_dataset + ) + + +async def test_async_change_channel_no_pending( + hass: HomeAssistant, otbr_config_entry +) -> None: + """Test test_async_change_channel when the pending dataset already expired.""" + + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 1 + assert list(store.datasets.values())[0].tlv == DATASET_CH16.hex() + + with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel, patch( + "python_otbr_api.OTBR.get_active_dataset_tlvs", + return_value=bytes.fromhex(DATASET_CH16_PENDING), + ), patch( + "python_otbr_api.OTBR.get_pending_dataset_tlvs", + return_value=None, + ): + await otbr_silabs_multiprotocol.async_change_channel(hass, 15, delay=5 * 300) + mock_set_channel.assert_awaited_once_with(15, delay=5 * 300 * 1000) + + pending_dataset = tlv_parser.parse_tlv(DATASET_CH16_PENDING) + pending_dataset.pop(tlv_parser.MeshcopTLVType.DELAYTIMER) + + assert len(store.datasets) == 1 + assert list(store.datasets.values())[0].tlv == tlv_parser.encode_tlv( + pending_dataset + ) + + +async def test_async_change_channel_no_update( + hass: HomeAssistant, otbr_config_entry +) -> None: + """Test test_async_change_channel when we didn't get a dataset from the OTBR.""" + + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 1 + assert list(store.datasets.values())[0].tlv == DATASET_CH16.hex() + + with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel, patch( + "python_otbr_api.OTBR.get_active_dataset_tlvs", + return_value=None, + ), patch( + "python_otbr_api.OTBR.get_pending_dataset_tlvs", + return_value=None, + ): + await otbr_silabs_multiprotocol.async_change_channel(hass, 15, delay=5 * 300) + mock_set_channel.assert_awaited_once_with(15, delay=5 * 300 * 1000) + + assert list(store.datasets.values())[0].tlv == DATASET_CH16.hex() + + +async def test_async_change_channel_no_otbr(hass: HomeAssistant) -> None: + """Test async_change_channel when otbr is not configured.""" + + 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) -> None: + """Test test_async_get_channel.""" + + with patch( + "python_otbr_api.OTBR.get_active_dataset", + return_value=ActiveDataSet(channel=11), + ) as mock_get_active_dataset: + assert await otbr_silabs_multiprotocol.async_get_channel(hass) == 11 + mock_get_active_dataset.assert_awaited_once_with() + + +async def test_async_get_channel_no_dataset( + hass: HomeAssistant, otbr_config_entry +) -> None: + """Test test_async_get_channel.""" + + with patch( + "python_otbr_api.OTBR.get_active_dataset", + return_value=None, + ) as mock_get_active_dataset: + assert await otbr_silabs_multiprotocol.async_get_channel(hass) is None + mock_get_active_dataset.assert_awaited_once_with() + + +async def test_async_get_channel_error(hass: HomeAssistant, otbr_config_entry) -> None: + """Test test_async_get_channel.""" + + with patch( + "python_otbr_api.OTBR.get_active_dataset", + side_effect=HomeAssistantError, + ) as mock_get_active_dataset: + assert await otbr_silabs_multiprotocol.async_get_channel(hass) is None + mock_get_active_dataset.assert_awaited_once_with() + + +async def test_async_get_channel_no_otbr(hass: HomeAssistant) -> None: + """Test test_async_get_channel when otbr is not configured.""" + + with patch("python_otbr_api.OTBR.get_active_dataset") as mock_get_active_dataset: + await otbr_silabs_multiprotocol.async_get_channel(hass) + mock_get_active_dataset.assert_not_awaited() + + +@pytest.mark.parametrize( + ("url", "expected"), + [(OTBR_MULTIPAN_URL, True), (OTBR_NON_MULTIPAN_URL, False)], +) +async def test_async_using_multipan( + hass: HomeAssistant, otbr_config_entry, url: str, expected: bool +) -> None: + """Test async_change_channel when otbr is not configured.""" + data: otbr.OTBRData = hass.data[otbr.DOMAIN] + data.url = url + + assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is expected + + +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 diff --git a/tests/components/otbr/test_util.py b/tests/components/otbr/test_util.py index af5306b3581..f8ed79b91ee 100644 --- a/tests/components/otbr/test_util.py +++ b/tests/components/otbr/test_util.py @@ -1,5 +1,4 @@ """Test OTBR Utility functions.""" -from unittest.mock import Mock, patch from homeassistant.components import otbr from homeassistant.core import HomeAssistant @@ -8,51 +7,19 @@ OTBR_MULTIPAN_URL = "http://core-silabs-multiprotocol:8081" OTBR_NON_MULTIPAN_URL = "/dev/ttyAMA1" -async def test_get_allowed_channel(hass: HomeAssistant) -> None: +async def test_get_allowed_channel( + hass: HomeAssistant, multiprotocol_addon_manager_mock +) -> None: """Test get_allowed_channel.""" - zha_networksettings = Mock() - zha_networksettings.network_info.channel = 15 - - # OTBR multipan + No ZHA -> no restriction + # OTBR multipan + No configured channel -> no restriction + multiprotocol_addon_manager_mock.async_get_channel.return_value = None assert await otbr.util.get_allowed_channel(hass, OTBR_MULTIPAN_URL) is None - # OTBR multipan + ZHA multipan empty settings -> no restriction - with patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=None, - ): - assert await otbr.util.get_allowed_channel(hass, OTBR_MULTIPAN_URL) is None + # OTBR multipan + multipan using channel 15 -> 15 + multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 + assert await otbr.util.get_allowed_channel(hass, OTBR_MULTIPAN_URL) == 15 - # OTBR multipan + ZHA not multipan using channel 15 -> no restriction - with patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="/dev/ttyAMA1", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=zha_networksettings, - ): - assert await otbr.util.get_allowed_channel(hass, OTBR_MULTIPAN_URL) is None - - # OTBR multipan + ZHA multipan using channel 15 -> 15 - with patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=zha_networksettings, - ): - assert await otbr.util.get_allowed_channel(hass, OTBR_MULTIPAN_URL) == 15 - - # OTBR not multipan + ZHA multipan using channel 15 -> no restriction - with patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=zha_networksettings, - ): - assert await otbr.util.get_allowed_channel(hass, OTBR_NON_MULTIPAN_URL) is None + # OTBR no multipan + multipan using channel 15 -> no restriction + multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 + assert await otbr.util.get_allowed_channel(hass, OTBR_NON_MULTIPAN_URL) is None diff --git a/tests/components/otbr/test_websocket_api.py b/tests/components/otbr/test_websocket_api.py index bfc3f09d6fe..1feebe9c02c 100644 --- a/tests/components/otbr/test_websocket_api.py +++ b/tests/components/otbr/test_websocket_api.py @@ -1,5 +1,5 @@ """Test OTBR Websocket API.""" -from unittest.mock import Mock, patch +from unittest.mock import patch import pytest import python_otbr_api @@ -273,6 +273,7 @@ async def test_set_network_no_entry( async def test_set_network_channel_conflict( hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, + multiprotocol_addon_manager_mock, otbr_config_entry, websocket_client, ) -> None: @@ -281,24 +282,16 @@ async def test_set_network_channel_conflict( dataset_store = await thread.dataset_store.async_get_store(hass) dataset_id = list(dataset_store.datasets)[0] - networksettings = Mock() - networksettings.network_info.channel = 15 + multiprotocol_addon_manager_mock.async_get_channel.return_value = 15 - with patch( - "homeassistant.components.otbr.util.zha_api.async_get_radio_path", - return_value="socket://core-silabs-multiprotocol:9999", - ), patch( - "homeassistant.components.otbr.util.zha_api.async_get_network_settings", - return_value=networksettings, - ): - await websocket_client.send_json_auto_id( - { - "type": "otbr/set_network", - "dataset_id": dataset_id, - } - ) + await websocket_client.send_json_auto_id( + { + "type": "otbr/set_network", + "dataset_id": dataset_id, + } + ) - msg = await websocket_client.receive_json() + msg = await websocket_client.receive_json() assert not msg["success"] assert msg["error"]["code"] == "channel_conflict" diff --git a/tests/components/zha/test_silabs_multiprotocol.py b/tests/components/zha/test_silabs_multiprotocol.py new file mode 100644 index 00000000000..beae0230901 --- /dev/null +++ b/tests/components/zha/test_silabs_multiprotocol.py @@ -0,0 +1,118 @@ +"""Test ZHA Silicon Labs Multiprotocol support.""" +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import call, patch + +import pytest +import zigpy.backups +import zigpy.state + +from homeassistant.components import zha +from homeassistant.components.zha import api, silabs_multiprotocol +from homeassistant.core import HomeAssistant + +if TYPE_CHECKING: + from zigpy.application import ControllerApplication + + +@pytest.fixture(autouse=True) +def required_platform_only(): + """Only set up the required and required base platforms to speed up tests.""" + with patch("homeassistant.components.zha.PLATFORMS", ()): + yield + + +async def test_async_get_channel_active(hass: HomeAssistant, setup_zha) -> None: + """Test reading channel with an active ZHA installation.""" + await setup_zha() + + assert await silabs_multiprotocol.async_get_channel(hass) == 15 + + +async def test_async_get_channel_missing( + hass: HomeAssistant, setup_zha, zigpy_app_controller: ControllerApplication +) -> None: + """Test reading channel with an inactive ZHA installation, no valid channel.""" + await setup_zha() + + gateway = api._get_gateway(hass) + await zha.async_unload_entry(hass, gateway.config_entry) + + # Network settings were never loaded for whatever reason + zigpy_app_controller.state.network_info = zigpy.state.NetworkInfo() + zigpy_app_controller.state.node_info = zigpy.state.NodeInfo() + + with patch( + "bellows.zigbee.application.ControllerApplication.__new__", + return_value=zigpy_app_controller, + ): + assert await silabs_multiprotocol.async_get_channel(hass) is None + + +async def test_async_get_channel_no_zha(hass: HomeAssistant) -> None: + """Test reading channel with no ZHA config entries and no database.""" + assert await silabs_multiprotocol.async_get_channel(hass) is None + + +async def test_async_using_multipan_active(hass: HomeAssistant, setup_zha) -> None: + """Test async_using_multipan with an active ZHA installation.""" + await setup_zha() + + assert await silabs_multiprotocol.async_using_multipan(hass) is False + + +async def test_async_using_multipan_no_zha(hass: HomeAssistant) -> None: + """Test async_using_multipan with no ZHA config entries and no database.""" + assert await silabs_multiprotocol.async_using_multipan(hass) is False + + +async def test_change_channel( + hass: HomeAssistant, setup_zha, zigpy_app_controller: ControllerApplication +) -> None: + """Test changing the channel.""" + await setup_zha() + + with patch.object( + zigpy_app_controller, "move_network_to_channel", autospec=True + ) as mock_move_network_to_channel: + task = await silabs_multiprotocol.async_change_channel(hass, 20) + await task + + assert mock_move_network_to_channel.mock_calls == [call(20)] + + +async def test_change_channel_no_zha( + hass: HomeAssistant, zigpy_app_controller: ControllerApplication +) -> None: + """Test changing the channel with no ZHA config entries and no database.""" + with patch.object( + zigpy_app_controller, "move_network_to_channel", autospec=True + ) as mock_move_network_to_channel: + task = await silabs_multiprotocol.async_change_channel(hass, 20) + assert task is None + + assert mock_move_network_to_channel.mock_calls == [] + + +@pytest.mark.parametrize(("delay", "sleep"), [(0, 0), (5, 0), (15, 15 - 10.27)]) +async def test_change_channel_delay( + hass: HomeAssistant, + setup_zha, + zigpy_app_controller: ControllerApplication, + delay: float, + sleep: float, +) -> None: + """Test changing the channel with a delay.""" + await setup_zha() + + with patch.object( + zigpy_app_controller, "move_network_to_channel", autospec=True + ) as mock_move_network_to_channel, patch( + "homeassistant.components.zha.silabs_multiprotocol.asyncio.sleep", autospec=True + ) as mock_sleep: + task = await silabs_multiprotocol.async_change_channel(hass, 20, delay=delay) + await task + + assert mock_move_network_to_channel.mock_calls == [call(20)] + assert mock_sleep.mock_calls == [call(sleep)]