diff --git a/homeassistant/components/homeassistant_hardware/util.py b/homeassistant/components/homeassistant_hardware/util.py index 53cbcbae5d4..bd1ff642d10 100644 --- a/homeassistant/components/homeassistant_hardware/util.py +++ b/homeassistant/components/homeassistant_hardware/util.py @@ -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) diff --git a/homeassistant/components/otbr/__init__.py b/homeassistant/components/otbr/__init__.py index 4b95be1d40d..0756f32ab18 100644 --- a/homeassistant/components/otbr/__init__.py +++ b/homeassistant/components/otbr/__init__.py @@ -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 diff --git a/homeassistant/components/otbr/config_flow.py b/homeassistant/components/otbr/config_flow.py index aff79ca4651..514f6c7617c 100644 --- a/homeassistant/components/otbr/config_flow.py +++ b/homeassistant/components/otbr/config_flow.py @@ -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, diff --git a/homeassistant/components/otbr/homeassistant_hardware.py b/homeassistant/components/otbr/homeassistant_hardware.py new file mode 100644 index 00000000000..94193be1359 --- /dev/null +++ b/homeassistant/components/otbr/homeassistant_hardware.py @@ -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, + ) diff --git a/homeassistant/components/otbr/types.py b/homeassistant/components/otbr/types.py new file mode 100644 index 00000000000..eff6aa980d6 --- /dev/null +++ b/homeassistant/components/otbr/types.py @@ -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] diff --git a/homeassistant/components/otbr/util.py b/homeassistant/components/otbr/util.py index 351e23c7736..30e456e11a8 100644 --- a/homeassistant/components/otbr/util.py +++ b/homeassistant/components/otbr/util.py @@ -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.""" diff --git a/tests/components/homeassistant_hardware/test_util.py b/tests/components/homeassistant_hardware/test_util.py index 047de3e452c..52739f16886 100644 --- a/tests/components/homeassistant_hardware/test_util.py +++ b/tests/components/homeassistant_hardware/test_util.py @@ -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 diff --git a/tests/components/otbr/__init__.py b/tests/components/otbr/__init__.py index 7d52318b477..5f778169e55 100644 --- a/tests/components/otbr/__init__.py +++ b/tests/components/otbr/__init__.py @@ -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" +) diff --git a/tests/components/otbr/conftest.py b/tests/components/otbr/conftest.py index 5ab3e442183..9140fcf6847 100644 --- a/tests/components/otbr/conftest.py +++ b/tests/components/otbr/conftest.py @@ -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( diff --git a/tests/components/otbr/test_config_flow.py b/tests/components/otbr/test_config_flow.py index d14fbc5cbd1..8384b905b9c 100644 --- a/tests/components/otbr/test_config_flow.py +++ b/tests/components/otbr/test_config_flow.py @@ -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 diff --git a/tests/components/otbr/test_homeassistant_hardware.py b/tests/components/otbr/test_homeassistant_hardware.py new file mode 100644 index 00000000000..7f831656d06 --- /dev/null +++ b/tests/components/otbr/test_homeassistant_hardware.py @@ -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 diff --git a/tests/components/otbr/test_init.py b/tests/components/otbr/test_init.py index faf13786107..b14527165e6 100644 --- a/tests/components/otbr/test_init.py +++ b/tests/components/otbr/test_init.py @@ -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)