mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 16:57:53 +00:00
OTBR firmware API for Home Assistant Hardware (#138330)
* Implement `async_register_firmware_info_provider` for OTBR * Keep track of the current device for OTBR Keep track of the current device, part 2 * Fix unit tests * Revert keeping track of the current device * Fix existing unit tests * Increase test coverage * Remove unused code from tests * Reload OTBR when the addon reloads * Only reload if the current entry is running * Runtime test * Add a unit test for the reloading * Clarify the purpose of `ConfigEntryState.SETUP_IN_PROGRESS` * Simplify typing
This commit is contained in:
parent
c0068e0891
commit
81cac25bd0
@ -12,7 +12,7 @@ import logging
|
||||
from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType
|
||||
from universal_silabs_flasher.flasher import Flasher
|
||||
|
||||
from homeassistant.components.hassio import AddonError, AddonState
|
||||
from homeassistant.components.hassio import AddonError, AddonManager, AddonState
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.hassio import is_hassio
|
||||
@ -143,6 +143,31 @@ class FirmwareInfo:
|
||||
return all(states)
|
||||
|
||||
|
||||
async def get_otbr_addon_firmware_info(
|
||||
hass: HomeAssistant, otbr_addon_manager: AddonManager
|
||||
) -> FirmwareInfo | None:
|
||||
"""Get firmware info from the OTBR add-on."""
|
||||
try:
|
||||
otbr_addon_info = await otbr_addon_manager.async_get_addon_info()
|
||||
except AddonError:
|
||||
return None
|
||||
|
||||
if otbr_addon_info.state == AddonState.NOT_INSTALLED:
|
||||
return None
|
||||
|
||||
if (otbr_path := otbr_addon_info.options.get("device")) is None:
|
||||
return None
|
||||
|
||||
# Only create a new entry if there are no existing OTBR ones
|
||||
return FirmwareInfo(
|
||||
device=otbr_path,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[OwningAddon(slug=otbr_addon_manager.addon_slug)],
|
||||
)
|
||||
|
||||
|
||||
async def guess_hardware_owners(
|
||||
hass: HomeAssistant, device_path: str
|
||||
) -> list[FirmwareInfo]:
|
||||
@ -155,28 +180,19 @@ async def guess_hardware_owners(
|
||||
# It may be possible for the OTBR addon to be present without the integration
|
||||
if is_hassio(hass):
|
||||
otbr_addon_manager = get_otbr_addon_manager(hass)
|
||||
otbr_addon_fw_info = await get_otbr_addon_firmware_info(
|
||||
hass, otbr_addon_manager
|
||||
)
|
||||
otbr_path = (
|
||||
otbr_addon_fw_info.device if otbr_addon_fw_info is not None else None
|
||||
)
|
||||
|
||||
try:
|
||||
otbr_addon_info = await otbr_addon_manager.async_get_addon_info()
|
||||
except AddonError:
|
||||
pass
|
||||
else:
|
||||
if otbr_addon_info.state != AddonState.NOT_INSTALLED:
|
||||
otbr_path = otbr_addon_info.options.get("device")
|
||||
|
||||
# Only create a new entry if there are no existing OTBR ones
|
||||
if otbr_path is not None and not any(
|
||||
info.source == "otbr" for info in device_guesses[otbr_path]
|
||||
):
|
||||
device_guesses[otbr_path].append(
|
||||
FirmwareInfo(
|
||||
device=otbr_path,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[OwningAddon(slug=otbr_addon_manager.addon_slug)],
|
||||
)
|
||||
)
|
||||
# Only create a new entry if there are no existing OTBR ones
|
||||
if otbr_path is not None and not any(
|
||||
info.source == "otbr" for info in device_guesses[otbr_path]
|
||||
):
|
||||
assert otbr_addon_fw_info is not None
|
||||
device_guesses[otbr_path].append(otbr_addon_fw_info)
|
||||
|
||||
if is_hassio(hass):
|
||||
multipan_addon_manager = await get_multiprotocol_addon_manager(hass)
|
||||
|
@ -7,16 +7,20 @@ import logging
|
||||
import aiohttp
|
||||
import python_otbr_api
|
||||
|
||||
from homeassistant.components.homeassistant_hardware.helpers import (
|
||||
async_notify_firmware_info,
|
||||
async_register_firmware_info_provider,
|
||||
)
|
||||
from homeassistant.components.thread import async_add_dataset
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
|
||||
from homeassistant.helpers import config_validation as cv, issue_registry as ir
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from . import websocket_api
|
||||
from . import homeassistant_hardware, websocket_api
|
||||
from .const import DOMAIN
|
||||
from .types import OTBRConfigEntry
|
||||
from .util import (
|
||||
GetBorderAgentIdNotSupported,
|
||||
OTBRData,
|
||||
@ -28,12 +32,13 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
||||
|
||||
type OTBRConfigEntry = ConfigEntry[OTBRData]
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the Open Thread Border Router component."""
|
||||
websocket_api.async_setup(hass)
|
||||
|
||||
async_register_firmware_info_provider(hass, DOMAIN, homeassistant_hardware)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@ -77,6 +82,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: OTBRConfigEntry) -> bool
|
||||
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
|
||||
entry.runtime_data = otbrdata
|
||||
|
||||
if fw_info := await homeassistant_hardware.async_get_firmware_info(hass, entry):
|
||||
await async_notify_firmware_info(hass, DOMAIN, fw_info)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
|
@ -16,7 +16,12 @@ import yarl
|
||||
from homeassistant.components.hassio import AddonError, AddonManager
|
||||
from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware
|
||||
from homeassistant.components.thread import async_get_preferred_dataset
|
||||
from homeassistant.config_entries import SOURCE_HASSIO, ConfigFlow, ConfigFlowResult
|
||||
from homeassistant.config_entries import (
|
||||
SOURCE_HASSIO,
|
||||
ConfigEntryState,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
)
|
||||
from homeassistant.const import CONF_URL
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
@ -201,12 +206,23 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
# we have to assume it's the first version
|
||||
# This check can be removed in HA Core 2025.9
|
||||
unique_id = discovery_info.uuid
|
||||
|
||||
if unique_id != discovery_info.uuid:
|
||||
continue
|
||||
|
||||
if (
|
||||
unique_id != discovery_info.uuid
|
||||
or current_url.host != config["host"]
|
||||
current_url.host != config["host"]
|
||||
or current_url.port == config["port"]
|
||||
):
|
||||
# Reload the entry since OTBR has restarted
|
||||
if current_entry.state == ConfigEntryState.LOADED:
|
||||
assert current_entry.unique_id is not None
|
||||
await self.hass.config_entries.async_reload(
|
||||
current_entry.entry_id
|
||||
)
|
||||
|
||||
continue
|
||||
|
||||
# Update URL with the new port
|
||||
self.hass.config_entries.async_update_entry(
|
||||
current_entry,
|
||||
|
76
homeassistant/components/otbr/homeassistant_hardware.py
Normal file
76
homeassistant/components/otbr/homeassistant_hardware.py
Normal file
@ -0,0 +1,76 @@
|
||||
"""Home Assistant Hardware firmware utilities."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from yarl import URL
|
||||
|
||||
from homeassistant.components.hassio import AddonManager
|
||||
from homeassistant.components.homeassistant_hardware.util import (
|
||||
ApplicationType,
|
||||
FirmwareInfo,
|
||||
OwningAddon,
|
||||
OwningIntegration,
|
||||
get_otbr_addon_firmware_info,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.hassio import is_hassio
|
||||
|
||||
from .const import DOMAIN
|
||||
from .types import OTBRConfigEntry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_get_firmware_info(
|
||||
hass: HomeAssistant, config_entry: OTBRConfigEntry
|
||||
) -> FirmwareInfo | None:
|
||||
"""Return firmware information for the OpenThread Border Router."""
|
||||
owners: list[OwningIntegration | OwningAddon] = [
|
||||
OwningIntegration(config_entry_id=config_entry.entry_id)
|
||||
]
|
||||
|
||||
device = None
|
||||
|
||||
if is_hassio(hass) and (host := URL(config_entry.data["url"]).host) is not None:
|
||||
otbr_addon_manager = AddonManager(
|
||||
hass=hass,
|
||||
logger=_LOGGER,
|
||||
addon_name="OpenThread Border Router",
|
||||
addon_slug=host.replace("-", "_"),
|
||||
)
|
||||
|
||||
if (
|
||||
addon_fw_info := await get_otbr_addon_firmware_info(
|
||||
hass, otbr_addon_manager
|
||||
)
|
||||
) is not None:
|
||||
device = addon_fw_info.device
|
||||
owners.extend(addon_fw_info.owners)
|
||||
|
||||
firmware_version = None
|
||||
|
||||
if config_entry.state in (
|
||||
# This function is called during OTBR config entry setup so we need to account
|
||||
# for both config entry states
|
||||
ConfigEntryState.LOADED,
|
||||
ConfigEntryState.SETUP_IN_PROGRESS,
|
||||
):
|
||||
try:
|
||||
firmware_version = await config_entry.runtime_data.get_coprocessor_version()
|
||||
except HomeAssistantError:
|
||||
firmware_version = None
|
||||
|
||||
if device is None:
|
||||
return None
|
||||
|
||||
return FirmwareInfo(
|
||||
device=device,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=firmware_version,
|
||||
source=DOMAIN,
|
||||
owners=owners,
|
||||
)
|
7
homeassistant/components/otbr/types.py
Normal file
7
homeassistant/components/otbr/types.py
Normal file
@ -0,0 +1,7 @@
|
||||
"""The Open Thread Border Router integration types."""
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
||||
from .util import OTBRData
|
||||
|
||||
type OTBRConfigEntry = ConfigEntry[OTBRData]
|
@ -163,6 +163,11 @@ class OTBRData:
|
||||
"""Get extended address (EUI-64)."""
|
||||
return await self.api.get_extended_address()
|
||||
|
||||
@_handle_otbr_error
|
||||
async def get_coprocessor_version(self) -> str:
|
||||
"""Get coprocessor firmware version."""
|
||||
return await self.api.get_coprocessor_version()
|
||||
|
||||
|
||||
async def get_allowed_channel(hass: HomeAssistant, otbr_url: str) -> int | None:
|
||||
"""Return the allowed channel, or None if there's no restriction."""
|
||||
|
@ -2,7 +2,12 @@
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from homeassistant.components.hassio import AddonError, AddonInfo, AddonState
|
||||
from homeassistant.components.hassio import (
|
||||
AddonError,
|
||||
AddonInfo,
|
||||
AddonManager,
|
||||
AddonState,
|
||||
)
|
||||
from homeassistant.components.homeassistant_hardware.helpers import (
|
||||
async_register_firmware_info_provider,
|
||||
)
|
||||
@ -11,6 +16,7 @@ from homeassistant.components.homeassistant_hardware.util import (
|
||||
FirmwareInfo,
|
||||
OwningAddon,
|
||||
OwningIntegration,
|
||||
get_otbr_addon_firmware_info,
|
||||
guess_firmware_info,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
|
||||
@ -247,3 +253,30 @@ async def test_firmware_info(hass: HomeAssistant) -> None:
|
||||
)
|
||||
|
||||
assert (await firmware_info2.is_running(hass)) is False
|
||||
|
||||
|
||||
async def test_get_otbr_addon_firmware_info_failure(hass: HomeAssistant) -> None:
|
||||
"""Test getting OTBR addon firmware info failure due to bad API call."""
|
||||
|
||||
otbr_addon_manager = AsyncMock(spec_set=AddonManager)
|
||||
otbr_addon_manager.async_get_addon_info.side_effect = AddonError()
|
||||
|
||||
assert (await get_otbr_addon_firmware_info(hass, otbr_addon_manager)) is None
|
||||
|
||||
|
||||
async def test_get_otbr_addon_firmware_info_failure_bad_options(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting OTBR addon firmware info failure due to bad addon options."""
|
||||
|
||||
otbr_addon_manager = AsyncMock(spec_set=AddonManager)
|
||||
otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
|
||||
available=True,
|
||||
hostname="core_some_addon_slug",
|
||||
options={}, # `device` is missing
|
||||
state=AddonState.RUNNING,
|
||||
update_available=False,
|
||||
version="1.0.0",
|
||||
)
|
||||
|
||||
assert (await get_otbr_addon_firmware_info(hass, otbr_addon_manager)) is None
|
||||
|
@ -33,6 +33,8 @@ TEST_BORDER_AGENT_EXTENDED_ADDRESS = bytes.fromhex("AEEB2F594B570BBF")
|
||||
TEST_BORDER_AGENT_ID = bytes.fromhex("230C6A1AC57F6F4BE262ACF32E5EF52C")
|
||||
TEST_BORDER_AGENT_ID_2 = bytes.fromhex("230C6A1AC57F6F4BE262ACF32E5EF52D")
|
||||
|
||||
COPROCESSOR_VERSION = "OPENTHREAD/thread-reference-20200818-1740-g33cc75ed3; NRF52840; Jun 2 2022 14:25:49"
|
||||
|
||||
ROUTER_DISCOVERY_HASS = {
|
||||
"type_": "_meshcop._udp.local.",
|
||||
"name": "HomeAssistant OpenThreadBorderRouter #0BBF._meshcop._udp.local.",
|
||||
@ -60,3 +62,7 @@ ROUTER_DISCOVERY_HASS = {
|
||||
},
|
||||
"interface_index": None,
|
||||
}
|
||||
|
||||
TEST_COPROCESSOR_VERSION = (
|
||||
"SL-OPENTHREAD/2.4.4.0_GitHub-7074a43e4; EFR32; Oct 21 2024 14:40:57"
|
||||
)
|
||||
|
@ -15,6 +15,7 @@ from . import (
|
||||
DATASET_CH16,
|
||||
TEST_BORDER_AGENT_EXTENDED_ADDRESS,
|
||||
TEST_BORDER_AGENT_ID,
|
||||
TEST_COPROCESSOR_VERSION,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
@ -71,12 +72,23 @@ def get_extended_address_fixture() -> Generator[AsyncMock]:
|
||||
yield get_extended_address
|
||||
|
||||
|
||||
@pytest.fixture(name="get_coprocessor_version")
|
||||
def get_coprocessor_version_fixture() -> Generator[AsyncMock]:
|
||||
"""Mock get_coprocessor_version."""
|
||||
with patch(
|
||||
"python_otbr_api.OTBR.get_coprocessor_version",
|
||||
return_value=TEST_COPROCESSOR_VERSION,
|
||||
) as get_coprocessor_version:
|
||||
yield get_coprocessor_version
|
||||
|
||||
|
||||
@pytest.fixture(name="otbr_config_entry_multipan")
|
||||
async def otbr_config_entry_multipan_fixture(
|
||||
hass: HomeAssistant,
|
||||
get_active_dataset_tlvs: AsyncMock,
|
||||
get_border_agent_id: AsyncMock,
|
||||
get_extended_address: AsyncMock,
|
||||
get_coprocessor_version: AsyncMock,
|
||||
) -> str:
|
||||
"""Mock Open Thread Border Router config entry."""
|
||||
config_entry = MockConfigEntry(
|
||||
@ -97,6 +109,7 @@ async def otbr_config_entry_thread_fixture(
|
||||
get_active_dataset_tlvs: AsyncMock,
|
||||
get_border_agent_id: AsyncMock,
|
||||
get_extended_address: AsyncMock,
|
||||
get_coprocessor_version: AsyncMock,
|
||||
) -> None:
|
||||
"""Mock Open Thread Border Router config entry."""
|
||||
config_entry = MockConfigEntry(
|
||||
|
@ -10,9 +10,18 @@ import pytest
|
||||
import python_otbr_api
|
||||
|
||||
from homeassistant.components import otbr
|
||||
from homeassistant.components.homeassistant_hardware.helpers import (
|
||||
async_register_firmware_info_callback,
|
||||
)
|
||||
from homeassistant.components.homeassistant_hardware.util import (
|
||||
ApplicationType,
|
||||
FirmwareInfo,
|
||||
OwningAddon,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import DATASET_CH15, DATASET_CH16, TEST_BORDER_AGENT_ID, TEST_BORDER_AGENT_ID_2
|
||||
|
||||
@ -32,6 +41,19 @@ HASSIO_DATA_2 = HassioServiceInfo(
|
||||
uuid="23456",
|
||||
)
|
||||
|
||||
HASSIO_DATA_OTBR = HassioServiceInfo(
|
||||
config={
|
||||
"host": "core-openthread-border-router",
|
||||
"port": 8081,
|
||||
"device": "/dev/ttyUSB1",
|
||||
"firmware": "SL-OPENTHREAD/2.4.4.0_GitHub-7074a43e4; EFR32; Oct 21 2024 14:40:57\r",
|
||||
"addon": "OpenThread Border Router",
|
||||
},
|
||||
name="OpenThread Border Router",
|
||||
slug="core_openthread_border_router",
|
||||
uuid="c58ba80fc88548008776bf8da903ef21",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="otbr_addon_info")
|
||||
def otbr_addon_info_fixture(addon_info: AsyncMock, addon_installed) -> AsyncMock:
|
||||
@ -97,6 +119,7 @@ async def test_user_flow_additional_entry(
|
||||
@pytest.mark.usefixtures(
|
||||
"get_active_dataset_tlvs",
|
||||
"get_extended_address",
|
||||
"get_coprocessor_version",
|
||||
)
|
||||
async def test_user_flow_additional_entry_fail_get_address(
|
||||
hass: HomeAssistant,
|
||||
@ -174,6 +197,7 @@ async def _finish_user_flow(
|
||||
"get_active_dataset_tlvs",
|
||||
"get_border_agent_id",
|
||||
"get_extended_address",
|
||||
"get_coprocessor_version",
|
||||
)
|
||||
async def test_user_flow_additional_entry_same_address(
|
||||
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
|
||||
@ -563,7 +587,11 @@ async def test_hassio_discovery_flow_2x_addons(
|
||||
assert config_entry.unique_id == HASSIO_DATA_2.uuid
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("get_active_dataset_tlvs", "get_extended_address")
|
||||
@pytest.mark.usefixtures(
|
||||
"get_active_dataset_tlvs",
|
||||
"get_extended_address",
|
||||
"get_coprocessor_version",
|
||||
)
|
||||
async def test_hassio_discovery_flow_2x_addons_same_ext_address(
|
||||
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_addon_info
|
||||
) -> None:
|
||||
@ -963,3 +991,55 @@ async def test_config_flow_additional_entry(
|
||||
)
|
||||
|
||||
assert result["type"] is expected_result
|
||||
|
||||
|
||||
@pytest.mark.usefixtures(
|
||||
"get_border_agent_id", "get_extended_address", "get_coprocessor_version"
|
||||
)
|
||||
async def test_hassio_discovery_reload(
|
||||
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, otbr_addon_info
|
||||
) -> None:
|
||||
"""Test the hassio discovery flow."""
|
||||
await async_setup_component(hass, "homeassistant_hardware", {})
|
||||
|
||||
aioclient_mock.get(
|
||||
"http://core-openthread-border-router:8081/node/dataset/active", text=""
|
||||
)
|
||||
|
||||
callback = Mock()
|
||||
async_register_firmware_info_callback(hass, "/dev/ttyUSB1", callback)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.is_hassio",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.get_otbr_addon_firmware_info",
|
||||
return_value=FirmwareInfo(
|
||||
device="/dev/ttyUSB1",
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
),
|
||||
),
|
||||
):
|
||||
await hass.config_entries.flow.async_init(
|
||||
otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA_OTBR
|
||||
)
|
||||
|
||||
# OTBR is set up and calls the firmware info notification callback
|
||||
assert len(callback.mock_calls) == 1
|
||||
assert len(hass.config_entries.async_entries(otbr.DOMAIN)) == 1
|
||||
|
||||
# If we change discovery info and emit again, the integration will be reloaded
|
||||
# and firmware information will be broadcast again
|
||||
await hass.config_entries.flow.async_init(
|
||||
otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA_OTBR
|
||||
)
|
||||
|
||||
assert len(callback.mock_calls) == 2
|
||||
assert len(hass.config_entries.async_entries(otbr.DOMAIN)) == 1
|
||||
|
254
tests/components/otbr/test_homeassistant_hardware.py
Normal file
254
tests/components/otbr/test_homeassistant_hardware.py
Normal file
@ -0,0 +1,254 @@
|
||||
"""Test Home Assistant Hardware platform for OTBR."""
|
||||
|
||||
from unittest.mock import AsyncMock, Mock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.homeassistant_hardware.helpers import (
|
||||
async_register_firmware_info_callback,
|
||||
)
|
||||
from homeassistant.components.homeassistant_hardware.util import (
|
||||
ApplicationType,
|
||||
FirmwareInfo,
|
||||
OwningAddon,
|
||||
OwningIntegration,
|
||||
)
|
||||
from homeassistant.components.otbr.homeassistant_hardware import async_get_firmware_info
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import TEST_COPROCESSOR_VERSION
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
|
||||
DEVICE_PATH = "/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_9ab1da1ea4b3ed11956f4eaca7669f5d-if00-port0"
|
||||
|
||||
|
||||
async def test_get_firmware_info(hass: HomeAssistant) -> None:
|
||||
"""Test `async_get_firmware_info`."""
|
||||
|
||||
otbr = MockConfigEntry(
|
||||
domain="otbr",
|
||||
unique_id="some_unique_id",
|
||||
data={
|
||||
"url": "http://core_openthread_border_router:8888",
|
||||
},
|
||||
version=1,
|
||||
)
|
||||
otbr.add_to_hass(hass)
|
||||
otbr.mock_state(hass, ConfigEntryState.LOADED)
|
||||
|
||||
otbr.runtime_data = AsyncMock()
|
||||
otbr.runtime_data.get_coprocessor_version.return_value = TEST_COPROCESSOR_VERSION
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.is_hassio",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.AddonManager",
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.get_otbr_addon_firmware_info",
|
||||
return_value=FirmwareInfo(
|
||||
device=DEVICE_PATH,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
),
|
||||
),
|
||||
):
|
||||
fw_info = await async_get_firmware_info(hass, otbr)
|
||||
|
||||
assert fw_info == FirmwareInfo(
|
||||
device=DEVICE_PATH,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=TEST_COPROCESSOR_VERSION,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningIntegration(config_entry_id=otbr.entry_id),
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
async def test_get_firmware_info_ignored(hass: HomeAssistant) -> None:
|
||||
"""Test `async_get_firmware_info` with ignored entry."""
|
||||
|
||||
otbr = MockConfigEntry(
|
||||
domain="otbr",
|
||||
unique_id="some_unique_id",
|
||||
data={},
|
||||
version=1,
|
||||
)
|
||||
otbr.add_to_hass(hass)
|
||||
|
||||
fw_info = await async_get_firmware_info(hass, otbr)
|
||||
assert fw_info is None
|
||||
|
||||
|
||||
async def test_get_firmware_info_no_coprocessor_version(hass: HomeAssistant) -> None:
|
||||
"""Test `async_get_firmware_info` with no coprocessor version support."""
|
||||
|
||||
otbr = MockConfigEntry(
|
||||
domain="otbr",
|
||||
unique_id="some_unique_id",
|
||||
data={
|
||||
"url": "http://core_openthread_border_router:8888",
|
||||
},
|
||||
version=1,
|
||||
)
|
||||
otbr.add_to_hass(hass)
|
||||
otbr.mock_state(hass, ConfigEntryState.LOADED)
|
||||
|
||||
otbr.runtime_data = AsyncMock()
|
||||
otbr.runtime_data.get_coprocessor_version.side_effect = HomeAssistantError()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.is_hassio",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.AddonManager",
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.get_otbr_addon_firmware_info",
|
||||
return_value=FirmwareInfo(
|
||||
device=DEVICE_PATH,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
),
|
||||
),
|
||||
):
|
||||
fw_info = await async_get_firmware_info(hass, otbr)
|
||||
|
||||
assert fw_info == FirmwareInfo(
|
||||
device=DEVICE_PATH,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningIntegration(config_entry_id=otbr.entry_id),
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("version", "expected_version"),
|
||||
[
|
||||
((TEST_COPROCESSOR_VERSION,), TEST_COPROCESSOR_VERSION),
|
||||
(HomeAssistantError(), None),
|
||||
],
|
||||
)
|
||||
async def test_hardware_firmware_info_provider_notification(
|
||||
hass: HomeAssistant,
|
||||
version: str | Exception,
|
||||
expected_version: str | None,
|
||||
get_active_dataset_tlvs: AsyncMock,
|
||||
get_border_agent_id: AsyncMock,
|
||||
get_extended_address: AsyncMock,
|
||||
get_coprocessor_version: AsyncMock,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
) -> None:
|
||||
"""Test that the OTBR provides hardware and firmware information."""
|
||||
otbr = MockConfigEntry(
|
||||
domain="otbr",
|
||||
unique_id="some_unique_id",
|
||||
data={
|
||||
"url": "http://core_openthread_border_router:8888",
|
||||
},
|
||||
version=1,
|
||||
)
|
||||
otbr.add_to_hass(hass)
|
||||
|
||||
await async_setup_component(hass, "homeassistant_hardware", {})
|
||||
|
||||
callback = Mock()
|
||||
async_register_firmware_info_callback(hass, DEVICE_PATH, callback)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.is_hassio",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.AddonManager",
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.get_otbr_addon_firmware_info",
|
||||
return_value=FirmwareInfo(
|
||||
device=DEVICE_PATH,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=None,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
),
|
||||
),
|
||||
):
|
||||
get_coprocessor_version.side_effect = version
|
||||
await hass.config_entries.async_setup(otbr.entry_id)
|
||||
|
||||
assert callback.mock_calls == [
|
||||
call(
|
||||
FirmwareInfo(
|
||||
device=DEVICE_PATH,
|
||||
firmware_type=ApplicationType.SPINEL,
|
||||
firmware_version=expected_version,
|
||||
source="otbr",
|
||||
owners=[
|
||||
OwningIntegration(config_entry_id=otbr.entry_id),
|
||||
OwningAddon(slug="core_openthread_border_router"),
|
||||
],
|
||||
)
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
async def test_get_firmware_info_remote_otbr(hass: HomeAssistant) -> None:
|
||||
"""Test `async_get_firmware_info` with no coprocessor version support."""
|
||||
|
||||
otbr = MockConfigEntry(
|
||||
domain="otbr",
|
||||
unique_id="some_unique_id",
|
||||
data={
|
||||
"url": "http://192.168.1.10:8888",
|
||||
},
|
||||
version=1,
|
||||
)
|
||||
otbr.add_to_hass(hass)
|
||||
otbr.mock_state(hass, ConfigEntryState.LOADED)
|
||||
|
||||
otbr.runtime_data = AsyncMock()
|
||||
otbr.runtime_data.get_coprocessor_version.return_value = TEST_COPROCESSOR_VERSION
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.is_hassio",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.AddonManager",
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.otbr.homeassistant_hardware.get_otbr_addon_firmware_info",
|
||||
return_value=None,
|
||||
),
|
||||
):
|
||||
fw_info = await async_get_firmware_info(hass, otbr)
|
||||
|
||||
assert fw_info is None
|
@ -26,6 +26,7 @@ from . import (
|
||||
ROUTER_DISCOVERY_HASS,
|
||||
TEST_BORDER_AGENT_EXTENDED_ADDRESS,
|
||||
TEST_BORDER_AGENT_ID,
|
||||
TEST_COPROCESSOR_VERSION,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
@ -43,6 +44,7 @@ def enable_mocks_fixture(
|
||||
get_active_dataset_tlvs: AsyncMock,
|
||||
get_border_agent_id: AsyncMock,
|
||||
get_extended_address: AsyncMock,
|
||||
get_coprocessor_version: AsyncMock,
|
||||
) -> None:
|
||||
"""Enable API mocks."""
|
||||
|
||||
@ -298,6 +300,7 @@ async def test_config_entry_update(hass: HomeAssistant) -> None:
|
||||
mock_api.get_extended_address = AsyncMock(
|
||||
return_value=TEST_BORDER_AGENT_EXTENDED_ADDRESS
|
||||
)
|
||||
mock_api.get_coprocessor_version = AsyncMock(return_value=TEST_COPROCESSOR_VERSION)
|
||||
with patch("python_otbr_api.OTBR", return_value=mock_api) as mock_otrb_api:
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user