evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add silabs_multiprotocol platform (#92904)

Browse source 
* Add silabs_multiprotocol platform

* Add new files

* Add ZHA tests

* Prevent ZHA from creating database during tests

* Add delay parameter to async_change_channel

* Add the updated dataset to the dataset store

* Allow MultipanProtocol.async_change_channel to return a task

* Notify user about the duration of migration

* Update tests
This commit is contained in:
Erik Montnemery 2023-06-01 12:32:14 +02:00 • committed by GitHub
parent 4f153a8f90
commit 15e5cf01bb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 1072 additions and 148 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

261
homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py
View file

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

19
homeassistant/components/homeassistant_hardware/strings.json
View file

@ -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": {

19
homeassistant/components/homeassistant_sky_connect/strings.json
View file

@ -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": {

19
homeassistant/components/homeassistant_yellow/strings.json
View file

@ -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": {

87
homeassistant/components/otbr/silabs_multiprotocol.py Normal file
View file

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

50
homeassistant/components/otbr/util.py
View file

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

2
homeassistant/components/zha/config_flow.py
View file

@ -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()

81
homeassistant/components/zha/silabs_multiprotocol.py Normal file
View file

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

13
tests/components/homeassistant_hardware/conftest.py
View file

@ -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."""

209
tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py
View file

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

13
tests/components/homeassistant_sky_connect/conftest.py
View file

@ -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."""

13
tests/components/homeassistant_yellow/conftest.py
View file

@ -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."""

12
tests/components/otbr/conftest.py
View file

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

17
tests/components/otbr/test_config_flow.py
View file

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

30
tests/components/otbr/test_init.py
View file

@ -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())

175
tests/components/otbr/test_silabs_multiprotocol.py Normal file
View file

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

55
tests/components/otbr/test_util.py
View file

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

27
tests/components/otbr/test_websocket_api.py
View file

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

118
tests/components/zha/test_silabs_multiprotocol.py Normal file
View file

@ -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)]