jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add support for multiple otbr config entries (#124289)

Browse Source 
* Add support for multiple otbr config entries

* Fix test

* Drop useless fixture

* Address review comments

* Change unique id from xa to id

* Improve error text

* Store data in ConfigEntry.runtime_data

* Remove useless function
This commit is contained in:
Erik Montnemery 2024-08-27 19:07:35 +02:00 committed by GitHub
parent 52b6f00363
commit e2d84f9a58
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 563 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
homeassistant/components/otbr/__init__.py
View File

@ -2,6 +2,8 @@
from __future__ import annotations
import logging
import aiohttp
import python_otbr_api
@ -14,22 +16,28 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from . import websocket_api
from .const import DATA_OTBR, DOMAIN
from .util import OTBRData, update_issues
from .const import DOMAIN
from .util import (
GetBorderAgentIdNotSupported,
OTBRData,
update_issues,
update_unique_id,
)
_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)
if len(config_entries := hass.config_entries.async_entries(DOMAIN)):
for config_entry in config_entries[1:]:
await hass.config_entries.async_remove(config_entry.entry_id)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: OTBRConfigEntry) -> bool:
"""Set up an Open Thread Border Router config entry."""
api = python_otbr_api.OTBR(entry.data["url"], async_get_clientsession(hass), 10)
@ -38,13 +46,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
border_agent_id = await otbrdata.get_border_agent_id()
dataset_tlvs = await otbrdata.get_active_dataset_tlvs()
extended_address = await otbrdata.get_extended_address()
except (
HomeAssistantError,
aiohttp.ClientError,
TimeoutError,
) as err:
raise ConfigEntryNotReady("Unable to connect") from err
if border_agent_id is None:
except GetBorderAgentIdNotSupported:
ir.async_create_issue(
hass,
DOMAIN,
@ -55,6 +57,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
translation_key="get_get_border_agent_id_unsupported",
)
return False
except (
HomeAssistantError,
aiohttp.ClientError,
TimeoutError,
) as err:
raise ConfigEntryNotReady("Unable to connect") from err
await update_unique_id(hass, entry, border_agent_id)
if dataset_tlvs:
await update_issues(hass, otbrdata, dataset_tlvs)
await async_add_dataset(
@ -66,18 +75,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
hass.data[DATA_OTBR] = otbrdata
entry.runtime_data = otbrdata
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: OTBRConfigEntry) -> bool:
"""Unload a config entry."""
hass.data.pop(DATA_OTBR)
return True
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def async_reload_entry(hass: HomeAssistant, entry: OTBRConfigEntry) -> None:
"""Handle an options update."""
await hass.config_entries.async_reload(entry.entry_id)

89
homeassistant/components/otbr/config_flow.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from contextlib import suppress
import logging
from typing import cast
from typing import TYPE_CHECKING, cast
import aiohttp
import python_otbr_api
@ -33,9 +33,16 @@ from .util import (
get_allowed_channel,
)
if TYPE_CHECKING:
from . import OTBRConfigEntry
_LOGGER = logging.getLogger(__name__)
class AlreadyConfigured(HomeAssistantError):
"""Raised when the router is already configured."""
def _is_yellow(hass: HomeAssistant) -> bool:
"""Return True if Home Assistant is running on a Home Assistant Yellow."""
try:
@ -70,9 +77,8 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
async def _connect_and_set_dataset(self, otbr_url: str) -> None:
async def _set_dataset(self, api: python_otbr_api.OTBR, otbr_url: str) -> None:
"""Connect to the OTBR and create or apply a dataset if it doesn't have one."""
api = python_otbr_api.OTBR(otbr_url, async_get_clientsession(self.hass), 10)
if await api.get_active_dataset_tlvs() is None:
allowed_channel = await get_allowed_channel(self.hass, otbr_url)
@ -89,7 +95,9 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
await api.set_active_dataset_tlvs(bytes.fromhex(thread_dataset_tlv))
else:
_LOGGER.debug(
"not importing TLV with channel %s", thread_dataset_channel
"not importing TLV with channel %s for %s",
thread_dataset_channel,
otbr_url,
)
pan_id = generate_random_pan_id()
await api.create_active_dataset(
@ -101,27 +109,65 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
)
await api.set_enabled(True)
async def _is_border_agent_id_configured(self, border_agent_id: bytes) -> bool:
"""Return True if another config entry's OTBR has the same border agent id."""
config_entry: OTBRConfigEntry
for config_entry in self.hass.config_entries.async_loaded_entries(DOMAIN):
data = config_entry.runtime_data
try:
other_border_agent_id = await data.get_border_agent_id()
except HomeAssistantError:
_LOGGER.debug(
"Could not read border agent id from %s", data.url, exc_info=True
)
continue
_LOGGER.debug(
"border agent id for existing url %s: %s",
data.url,
other_border_agent_id.hex(),
)
if border_agent_id == other_border_agent_id:
return True
return False
async def _connect_and_configure_router(self, otbr_url: str) -> bytes:
"""Connect to the router and configure it if needed.
Will raise if the router's border agent id is in use by another config entry.
Returns the router's border agent id.
"""
api = python_otbr_api.OTBR(otbr_url, async_get_clientsession(self.hass), 10)
border_agent_id = await api.get_border_agent_id()
_LOGGER.debug("border agent id for url %s: %s", otbr_url, border_agent_id.hex())
if await self._is_border_agent_id_configured(border_agent_id):
raise AlreadyConfigured
await self._set_dataset(api, otbr_url)
return border_agent_id
async def async_step_user(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Set up by user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
errors = {}
if user_input is not None:
url = user_input[CONF_URL].rstrip("/")
try:
await self._connect_and_set_dataset(url)
border_agent_id = await self._connect_and_configure_router(url)
except AlreadyConfigured:
errors["base"] = "already_configured"
except (
python_otbr_api.OTBRError,
aiohttp.ClientError,
TimeoutError,
):
) as exc:
_LOGGER.debug("Failed to communicate with OTBR@%s: %s", url, exc)
errors["base"] = "cannot_connect"
else:
await self.async_set_unique_id(DOMAIN)
await self.async_set_unique_id(border_agent_id.hex())
return self.async_create_entry(
title="Open Thread Border Router",
data={CONF_URL: url},
@ -140,34 +186,35 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
url = f"http://{config['host']}:{config['port']}"
config_entry_data = {"url": url}
if self._async_in_progress(include_uninitialized=True):
# We currently don't handle multiple config entries, abort if hassio
# discovers multiple addons with otbr support
return self.async_abort(reason="single_instance_allowed")
if current_entries := self._async_current_entries():
for current_entry in current_entries:
if current_entry.source != SOURCE_HASSIO:
continue
current_url = yarl.URL(current_entry.data["url"])
if (
if not (unique_id := current_entry.unique_id):
# The first version did not set a unique_id
# so if the entry does not have a unique_id
# we have to assume it's the first version
current_entry.unique_id
and (current_entry.unique_id != discovery_info.uuid)
# This check can be removed in HA Core 2025.9
unique_id = discovery_info.uuid
if (
unique_id != discovery_info.uuid
or current_url.host != config["host"]
or current_url.port == config["port"]
):
continue
# Update URL with the new port
self.hass.config_entries.async_update_entry(
current_entry, data=config_entry_data
current_entry,
data=config_entry_data,
unique_id=unique_id, # Remove in HA Core 2025.9
)
return self.async_abort(reason="single_instance_allowed")
return self.async_abort(reason="already_configured")
try:
await self._connect_and_set_dataset(url)
await self._connect_and_configure_router(url)
except AlreadyConfigured:
return self.async_abort(reason="already_configured")
except (
python_otbr_api.OTBRError,
aiohttp.ClientError,

8
homeassistant/components/otbr/const.py
View File

@ -2,14 +2,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from homeassistant.util.hass_dict import HassKey
if TYPE_CHECKING:
from .util import OTBRData
DOMAIN = "otbr"
DATA_OTBR: HassKey[OTBRData] = HassKey(DOMAIN)
DEFAULT_CHANNEL = 15

21
homeassistant/components/otbr/silabs_multiprotocol.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine
from functools import wraps
import logging
from typing import Any, Concatenate
from typing import TYPE_CHECKING, Any, Concatenate
import aiohttp
from python_otbr_api import tlv_parser
@ -18,9 +18,12 @@ from homeassistant.components.thread import async_add_dataset
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .const import DATA_OTBR, DOMAIN
from .const import DOMAIN
from .util import OTBRData
if TYPE_CHECKING:
from . import OTBRConfigEntry
_LOGGER = logging.getLogger(__name__)
@ -45,16 +48,14 @@ def async_get_otbr_data[**_P, _R, _R_Def](
hass: HomeAssistant, *args: _P.args, **kwargs: _P.kwargs
) -> _R | _R_Def:
"""Fetch OTBR data and pass to orig_func."""
if DATA_OTBR not in hass.data:
return retval
data = hass.data[DATA_OTBR]
if not is_multiprotocol_url(data.url):
return retval
config_entry: OTBRConfigEntry
for config_entry in hass.config_entries.async_loaded_entries(DOMAIN):
data = config_entry.runtime_data
if is_multiprotocol_url(data.url):
return await orig_func(hass, data, *args, **kwargs)
return retval
return async_get_otbr_data_wrapper
return _async_get_otbr_data

1
homeassistant/components/otbr/strings.json
View File

@ -9,6 +9,7 @@
}
},
"error": {
"already_configured": "The Thread border router is already configured",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {

38
homeassistant/components/otbr/util.py
View File

@ -7,7 +7,7 @@ import dataclasses
from functools import wraps
import logging
import random
from typing import Any, Concatenate, cast
from typing import TYPE_CHECKING, Any, Concatenate, cast
import aiohttp
import python_otbr_api
@ -22,12 +22,16 @@ from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon
multi_pan_addon_using_device,
)
from homeassistant.components.homeassistant_yellow import RADIO_DEVICE as YELLOW_RADIO
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import issue_registry as ir
from .const import DOMAIN
if TYPE_CHECKING:
from . import OTBRConfigEntry
_LOGGER = logging.getLogger(__name__)
INFO_URL_SKY_CONNECT = (
@ -48,6 +52,10 @@ INSECURE_PASSPHRASES = (
)
class GetBorderAgentIdNotSupported(HomeAssistantError):
"""Raised from python_otbr_api.GetBorderAgentIdNotSupportedError."""
def compose_default_network_name(pan_id: int) -> str:
"""Generate a default network name."""
return f"ha-thread-{pan_id:04x}"
@ -83,7 +91,7 @@ class OTBRData:
entry_id: str
@_handle_otbr_error
async def factory_reset(self) -> None:
async def factory_reset(self, hass: HomeAssistant) -> None:
"""Reset the router."""
try:
await self.api.factory_reset()
@ -92,14 +100,19 @@ class OTBRData:
"OTBR does not support factory reset, attempting to delete dataset"
)
await self.delete_active_dataset()
await update_unique_id(
hass,
hass.config_entries.async_get_entry(self.entry_id),
await self.get_border_agent_id(),
)
@_handle_otbr_error
async def get_border_agent_id(self) -> bytes | None:
async def get_border_agent_id(self) -> bytes:
"""Get the border agent ID or None if not supported by the router."""
try:
return await self.api.get_border_agent_id()
except python_otbr_api.GetBorderAgentIdNotSupportedError:
return None
except python_otbr_api.GetBorderAgentIdNotSupportedError as exc:
raise GetBorderAgentIdNotSupported from exc
@_handle_otbr_error
async def set_enabled(self, enabled: bool) -> None:
@ -258,3 +271,18 @@ async def update_issues(
"""Raise or clear repair issues related to network settings."""
await _warn_on_channel_collision(hass, otbrdata, dataset_tlvs)
_warn_on_default_network_settings(hass, otbrdata, dataset_tlvs)
async def update_unique_id(
hass: HomeAssistant, entry: OTBRConfigEntry | None, border_agent_id: bytes
) -> None:
"""Update the config entry's unique_id if not matching."""
border_agent_id_hex = border_agent_id.hex()
if entry and entry.source == SOURCE_USER and entry.unique_id != border_agent_id_hex:
_LOGGER.debug(
"Updating unique_id of entry %s from %s to %s",
entry.entry_id,
entry.unique_id,
border_agent_id_hex,
)
hass.config_entries.async_update_entry(entry, unique_id=border_agent_id_hex)

50
homeassistant/components/otbr/websocket_api.py
View File

@ -2,7 +2,7 @@
from collections.abc import Callable, Coroutine
from functools import wraps
from typing import Any, cast
from typing import TYPE_CHECKING, Any, cast
import python_otbr_api
from python_otbr_api import PENDING_DATASET_DELAY_TIMER, tlv_parser
@ -17,7 +17,7 @@ from homeassistant.components.thread import async_add_dataset, async_get_dataset
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from .const import DATA_OTBR, DEFAULT_CHANNEL, DOMAIN
from .const import DEFAULT_CHANNEL, DOMAIN
from .util import (
OTBRData,
compose_default_network_name,
@ -26,6 +26,9 @@ from .util import (
update_issues,
)
if TYPE_CHECKING:
from . import OTBRConfigEntry
@callback
def async_setup(hass: HomeAssistant) -> None:
@ -47,12 +50,17 @@ async def websocket_info(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
) -> None:
"""Get OTBR info."""
if DATA_OTBR not in hass.data:
config_entries: list[OTBRConfigEntry]
config_entries = hass.config_entries.async_loaded_entries(DOMAIN)
if not config_entries:
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
return
data = hass.data[DATA_OTBR]
response: dict[str, dict[str, Any]] = {}
for config_entry in config_entries:
data = config_entry.runtime_data
try:
border_agent_id = await data.get_border_agent_id()
dataset = await data.get_active_dataset()
@ -67,12 +75,11 @@ async def websocket_info(
assert border_agent_id is not None
extended_pan_id = (
dataset.extended_pan_id.lower() if dataset and dataset.extended_pan_id else None
dataset.extended_pan_id.lower()
if dataset and dataset.extended_pan_id
else None
)
connection.send_result(
msg["id"],
{
extended_address: {
response[extended_address] = {
"active_dataset_tlvs": dataset_tlvs.hex() if dataset_tlvs else None,
"border_agent_id": border_agent_id.hex(),
"channel": dataset.channel if dataset else None,
@ -80,8 +87,8 @@ async def websocket_info(
"extended_pan_id": extended_pan_id,
"url": data.url,
}
},
)
connection.send_result(msg["id"], response)
def async_get_otbr_data(
@ -99,22 +106,29 @@ def async_get_otbr_data(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
) -> None:
"""Fetch OTBR data and pass to orig_func."""
if DATA_OTBR not in hass.data:
config_entries: list[OTBRConfigEntry]
config_entries = hass.config_entries.async_loaded_entries(DOMAIN)
if not config_entries:
connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
return
data = hass.data[DATA_OTBR]
for config_entry in config_entries:
data = config_entry.runtime_data
try:
extended_address = await data.get_extended_address()
except HomeAssistantError as exc:
connection.send_error(msg["id"], "get_extended_address_failed", str(exc))
connection.send_error(
msg["id"], "get_extended_address_failed", str(exc)
)
return
if extended_address.hex() != msg["extended_address"]:
connection.send_error(msg["id"], "unknown_router", "")
return
continue
await orig_func(hass, connection, msg, data)
return
connection.send_error(msg["id"], "unknown_router", "")
return async_check_extended_address_func
@ -144,7 +158,7 @@ async def websocket_create_network(
return
try:
await data.factory_reset()
await data.factory_reset(hass)
except HomeAssistantError as exc:
connection.send_error(msg["id"], "factory_reset_failed", str(exc))
return

1
tests/components/otbr/__init__.py
View File

@ -31,6 +31,7 @@ DATASET_INSECURE_PASSPHRASE = bytes.fromhex(
TEST_BORDER_AGENT_EXTENDED_ADDRESS = bytes.fromhex("AEEB2F594B570BBF")
TEST_BORDER_AGENT_ID = bytes.fromhex("230C6A1AC57F6F4BE262ACF32E5EF52C")
TEST_BORDER_AGENT_ID_2 = bytes.fromhex("230C6A1AC57F6F4BE262ACF32E5EF52D")
ROUTER_DISCOVERY_HASS = {
"type_": "_meshcop._udp.local.",

5
tests/components/otbr/conftest.py
View File

@ -77,16 +77,18 @@ async def otbr_config_entry_multipan_fixture(
get_active_dataset_tlvs: AsyncMock,
get_border_agent_id: AsyncMock,
get_extended_address: AsyncMock,
) -> None:
) -> str:
"""Mock Open Thread Border Router config entry."""
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
return config_entry.entry_id
@pytest.fixture(name="otbr_config_entry_thread")
@ -102,6 +104,7 @@ async def otbr_config_entry_thread_fixture(
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)

320
tests/components/otbr/test_config_flow.py
View File

@ -13,7 +13,7 @@ from homeassistant.components import hassio, otbr
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import DATASET_CH15, DATASET_CH16
from . import DATASET_CH15, DATASET_CH16, TEST_BORDER_AGENT_ID, TEST_BORDER_AGENT_ID_2
from tests.common import MockConfigEntry, MockModule, mock_integration
from tests.test_util.aiohttp import AiohttpClientMocker
@ -57,12 +57,91 @@ def addon_info_fixture():
"http://custom_url:1234//",
],
)
@pytest.mark.usefixtures(
"get_active_dataset_tlvs",
"get_border_agent_id",
)
async def test_user_flow(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, url: str
) -> None:
"""Test the user flow."""
await _finish_user_flow(hass, url)
@pytest.mark.usefixtures(
"get_active_dataset_tlvs",
"get_extended_address",
)
async def test_user_flow_additional_entry(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test more than a single entry is allowed."""
url1 = "http://custom_url:1234"
url2 = "http://custom_url_2:1234"
aioclient_mock.get(f"{url1}/node/ba-id", json=TEST_BORDER_AGENT_ID.hex())
aioclient_mock.get(f"{url2}/node/ba-id", json=TEST_BORDER_AGENT_ID_2.hex())
mock_integration(hass, MockModule("hassio"))
# Setup a config entry
config_entry = MockConfigEntry(
data={"url": url2},
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
unique_id=TEST_BORDER_AGENT_ID_2.hex(),
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
# Do a user flow
await _finish_user_flow(hass)
@pytest.mark.usefixtures(
"get_active_dataset_tlvs",
"get_extended_address",
)
async def test_user_flow_additional_entry_fail_get_address(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test more than a single entry is allowed.
This tets the behavior when we can't read the extended address from the existing
config entry.
"""
url1 = "http://custom_url:1234"
url2 = "http://custom_url_2:1234"
aioclient_mock.get(f"{url2}/node/ba-id", json=TEST_BORDER_AGENT_ID_2.hex())
mock_integration(hass, MockModule("hassio"))
# Setup a config entry
config_entry = MockConfigEntry(
data={"url": url2},
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
unique_id=TEST_BORDER_AGENT_ID_2.hex(),
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
# Do a user flow
aioclient_mock.clear_requests()
aioclient_mock.get(f"{url1}/node/ba-id", json=TEST_BORDER_AGENT_ID.hex())
aioclient_mock.get(f"{url2}/node/ba-id", status=HTTPStatus.NOT_FOUND)
await _finish_user_flow(hass)
assert f"Could not read border agent id from {url2}" in caplog.text
async def _finish_user_flow(
hass: HomeAssistant, url: str = "http://custom_url:1234"
) -> None:
"""Finish a user flow."""
stripped_url = "http://custom_url:1234"
aioclient_mock.get(f"{stripped_url}/node/dataset/active", text="aa")
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
@ -88,13 +167,56 @@ async def test_user_flow(
assert result["options"] == {}
assert len(mock_setup_entry.mock_calls) == 1
config_entry = hass.config_entries.async_entries(otbr.DOMAIN)[0]
config_entry = result["result"]
assert config_entry.data == expected_data
assert config_entry.options == {}
assert config_entry.title == "Open Thread Border Router"
assert config_entry.unique_id == otbr.DOMAIN
assert config_entry.unique_id == TEST_BORDER_AGENT_ID.hex()
@pytest.mark.usefixtures(
"get_active_dataset_tlvs",
"get_border_agent_id",
"get_extended_address",
)
async def test_user_flow_additional_entry_same_address(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test more than a single entry is allowed."""
mock_integration(hass, MockModule("hassio"))
# Setup a config entry
config_entry = MockConfigEntry(
data={"url": "http://custom_url:1234"},
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
unique_id=TEST_BORDER_AGENT_ID.hex(),
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
# Start user flow
url = "http://custom_url:1234"
aioclient_mock.get(f"{url}/node/dataset/active", text="aa")
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"url": url,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "already_configured"}
@pytest.mark.usefixtures("get_border_agent_id")
async def test_user_flow_router_not_setup(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
@ -158,10 +280,11 @@ async def test_user_flow_router_not_setup(
assert config_entry.data == expected_data
assert config_entry.options == {}
assert config_entry.title == "Open Thread Border Router"
assert config_entry.unique_id == otbr.DOMAIN
assert config_entry.unique_id == TEST_BORDER_AGENT_ID.hex()
async def test_user_flow_404(
@pytest.mark.usefixtures("get_border_agent_id")
async def test_user_flow_get_dataset_404(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the user flow."""
@ -192,7 +315,30 @@ async def test_user_flow_404(
aiohttp.ClientError,
],
)
async def test_user_flow_connect_error(hass: HomeAssistant, error) -> None:
async def test_user_flow_get_ba_id_connect_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, error
) -> None:
"""Test the user flow."""
await _test_user_flow_connect_error(hass, "get_border_agent_id", error)
@pytest.mark.usefixtures("get_border_agent_id")
@pytest.mark.parametrize(
"error",
[
TimeoutError,
python_otbr_api.OTBRError,
aiohttp.ClientError,
],
)
async def test_user_flow_get_dataset_connect_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, error
) -> None:
"""Test the user flow."""
await _test_user_flow_connect_error(hass, "get_active_dataset_tlvs", error)
async def _test_user_flow_connect_error(hass: HomeAssistant, func, error) -> None:
"""Test the user flow."""
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
@ -201,7 +347,7 @@ async def test_user_flow_connect_error(hass: HomeAssistant, error) -> None:
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with patch("python_otbr_api.OTBR.get_active_dataset_tlvs", side_effect=error):
with patch(f"python_otbr_api.OTBR.{func}", side_effect=error):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
@ -212,6 +358,7 @@ async def test_user_flow_connect_error(hass: HomeAssistant, error) -> None:
assert result["errors"] == {"base": "cannot_connect"}
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, addon_info
) -> None:
@ -244,6 +391,7 @@ async def test_hassio_discovery_flow(
assert config_entry.unique_id == HASSIO_DATA.uuid
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_yellow(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, addon_info
) -> None:
@ -301,6 +449,7 @@ async def test_hassio_discovery_flow_yellow(
),
],
)
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_sky_connect(
device: str,
title: str,
@ -346,6 +495,7 @@ async def test_hassio_discovery_flow_sky_connect(
assert config_entry.unique_id == HASSIO_DATA.uuid
@pytest.mark.usefixtures("get_active_dataset_tlvs", "get_extended_address")
async def test_hassio_discovery_flow_2x_addons(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, addon_info
) -> None:
@ -354,6 +504,101 @@ async def test_hassio_discovery_flow_2x_addons(
url2 = "http://core-silabs-multiprotocol_2:8081"
aioclient_mock.get(f"{url1}/node/dataset/active", text="aa")
aioclient_mock.get(f"{url2}/node/dataset/active", text="bb")
aioclient_mock.get(f"{url1}/node/ba-id", json=TEST_BORDER_AGENT_ID.hex())
aioclient_mock.get(f"{url2}/node/ba-id", json=TEST_BORDER_AGENT_ID_2.hex())
async def _addon_info(hass: HomeAssistant, slug: str) -> dict[str, Any]:
await asyncio.sleep(0)
if slug == "otbr":
return {
"available": True,
"hostname": None,
"options": {
"device": (
"/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_"
"9e2adbd75b8beb119fe564a0f320645d-if00-port0"
)
},
"state": None,
"update_available": False,
"version": None,
}
return {
"available": True,
"hostname": None,
"options": {
"device": (
"/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_"
"9e2adbd75b8beb119fe564a0f320645d-if00-port1"
)
},
"state": None,
"update_available": False,
"version": None,
}
addon_info.side_effect = _addon_info
result1 = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA
)
result2 = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA_2
)
results = [result1, result2]
expected_data = {
"url": f"http://{HASSIO_DATA.config['host']}:{HASSIO_DATA.config['port']}",
}
expected_data_2 = {
"url": f"http://{HASSIO_DATA_2.config['host']}:{HASSIO_DATA_2.config['port']}",
}
assert results[0]["type"] is FlowResultType.CREATE_ENTRY
assert (
results[0]["title"] == "Home Assistant SkyConnect (Silicon Labs Multiprotocol)"
)
assert results[0]["data"] == expected_data
assert results[0]["options"] == {}
assert results[1]["type"] is FlowResultType.CREATE_ENTRY
assert (
results[1]["title"] == "Home Assistant SkyConnect (Silicon Labs Multiprotocol)"
)
assert results[1]["data"] == expected_data_2
assert results[1]["options"] == {}
assert len(hass.config_entries.async_entries(otbr.DOMAIN)) == 2
config_entry = hass.config_entries.async_entries(otbr.DOMAIN)[0]
assert config_entry.data == expected_data
assert config_entry.options == {}
assert (
config_entry.title == "Home Assistant SkyConnect (Silicon Labs Multiprotocol)"
)
assert config_entry.unique_id == HASSIO_DATA.uuid
config_entry = hass.config_entries.async_entries(otbr.DOMAIN)[1]
assert config_entry.data == expected_data_2
assert config_entry.options == {}
assert (
config_entry.title == "Home Assistant SkyConnect (Silicon Labs Multiprotocol)"
)
assert config_entry.unique_id == HASSIO_DATA_2.uuid
@pytest.mark.usefixtures("get_active_dataset_tlvs", "get_extended_address")
async def test_hassio_discovery_flow_2x_addons_same_ext_address(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, addon_info
) -> None:
"""Test the hassio discovery flow when the user has 2 addons with otbr support."""
url1 = "http://core-silabs-multiprotocol:8081"
url2 = "http://core-silabs-multiprotocol_2:8081"
aioclient_mock.get(f"{url1}/node/dataset/active", text="aa")
aioclient_mock.get(f"{url2}/node/dataset/active", text="bb")
aioclient_mock.get(f"{url1}/node/ba-id", json=TEST_BORDER_AGENT_ID.hex())
aioclient_mock.get(f"{url2}/node/ba-id", json=TEST_BORDER_AGENT_ID.hex())
async def _addon_info(hass: HomeAssistant, slug: str) -> dict[str, Any]:
await asyncio.sleep(0)
@ -387,10 +632,6 @@ async def test_hassio_discovery_flow_2x_addons(
addon_info.side_effect = _addon_info
with patch(
"homeassistant.components.otbr.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result1 = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA
)
@ -411,9 +652,8 @@ async def test_hassio_discovery_flow_2x_addons(
assert results[0]["data"] == expected_data
assert results[0]["options"] == {}
assert results[1]["type"] is FlowResultType.ABORT
assert results[1]["reason"] == "single_instance_allowed"
assert results[1]["reason"] == "already_configured"
assert len(hass.config_entries.async_entries(otbr.DOMAIN)) == 1
assert len(mock_setup_entry.mock_calls) == 1
config_entry = hass.config_entries.async_entries(otbr.DOMAIN)[0]
assert config_entry.data == expected_data
@ -424,6 +664,7 @@ async def test_hassio_discovery_flow_2x_addons(
assert config_entry.unique_id == HASSIO_DATA.uuid
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_router_not_setup(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, addon_info
) -> None:
@ -481,6 +722,7 @@ async def test_hassio_discovery_flow_router_not_setup(
assert config_entry.unique_id == HASSIO_DATA.uuid
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_router_not_setup_has_preferred(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, addon_info
) -> None:
@ -533,6 +775,7 @@ async def test_hassio_discovery_flow_router_not_setup_has_preferred(
assert config_entry.unique_id == HASSIO_DATA.uuid
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_router_not_setup_has_preferred_2(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
@ -596,6 +839,7 @@ async def test_hassio_discovery_flow_router_not_setup_has_preferred_2(
assert config_entry.unique_id == HASSIO_DATA.uuid
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_404(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
@ -610,6 +854,7 @@ async def test_hassio_discovery_flow_404(
assert result["reason"] == "unknown"
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_new_port_missing_unique_id(
hass: HomeAssistant,
) -> None:
@ -633,7 +878,7 @@ async def test_hassio_discovery_flow_new_port_missing_unique_id(
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
assert result["reason"] == "already_configured"
expected_data = {
"url": f"http://{HASSIO_DATA.config['host']}:{HASSIO_DATA.config['port']}",
@ -642,6 +887,7 @@ async def test_hassio_discovery_flow_new_port_missing_unique_id(
assert config_entry.data == expected_data
@pytest.mark.usefixtures("get_border_agent_id")
async def test_hassio_discovery_flow_new_port(hass: HomeAssistant) -> None:
"""Test the port can be updated."""
mock_integration(hass, MockModule("hassio"))
@ -664,7 +910,7 @@ async def test_hassio_discovery_flow_new_port(hass: HomeAssistant) -> None:
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
assert result["reason"] == "already_configured"
expected_data = {
"url": f"http://{HASSIO_DATA.config['host']}:{HASSIO_DATA.config['port']}",
@ -673,6 +919,12 @@ async def test_hassio_discovery_flow_new_port(hass: HomeAssistant) -> None:
assert config_entry.data == expected_data
@pytest.mark.usefixtures(
"addon_info",
"get_active_dataset_tlvs",
"get_border_agent_id",
"get_extended_address",
)
async def test_hassio_discovery_flow_new_port_other_addon(hass: HomeAssistant) -> None:
"""Test the port is not updated if we get data for another addon hosting OTBR."""
mock_integration(hass, MockModule("hassio"))
@ -691,22 +943,34 @@ async def test_hassio_discovery_flow_new_port_other_addon(hass: HomeAssistant) -
otbr.DOMAIN, context={"source": "hassio"}, data=HASSIO_DATA
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
# Another entry will be created
assert result["type"] is FlowResultType.CREATE_ENTRY
# Make sure the data was not updated
# Make sure the data of the existing entry was not updated
expected_data = {
"url": f"http://openthread_border_router:{HASSIO_DATA.config['port']+1}",
}
config_entry = hass.config_entries.async_entries(otbr.DOMAIN)[0]
config_entry = hass.config_entries.async_get_entry(config_entry.entry_id)
assert config_entry.data == expected_data
@pytest.mark.parametrize(("source", "data"), [("hassio", HASSIO_DATA), ("user", None)])
async def test_config_flow_single_entry(
hass: HomeAssistant, source: str, data: Any
@pytest.mark.parametrize(
("source", "data", "expected_result"),
[
("hassio", HASSIO_DATA, FlowResultType.CREATE_ENTRY),
("user", None, FlowResultType.FORM),
],
)
@pytest.mark.usefixtures(
"addon_info",
"get_active_dataset_tlvs",
"get_border_agent_id",
"get_extended_address",
)
async def test_config_flow_additional_entry(
hass: HomeAssistant, source: str, data: Any, expected_result: FlowResultType
) -> None:
"""Test only a single entry is allowed."""
"""Test more than a single entry is allowed."""
mock_integration(hass, MockModule("hassio"))
# Setup the config entry
@ -719,13 +983,11 @@ async def test_config_flow_single_entry(
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.homeassistant_yellow.async_setup_entry",
"homeassistant.components.otbr.async_setup_entry",
return_value=True,
) as mock_setup_entry:
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": source}, data=data
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
mock_setup_entry.assert_not_called()
assert result["type"] is expected_result

45
tests/components/otbr/test_init.py
View File

@ -11,6 +11,7 @@ from zeroconf.asyncio import AsyncServiceInfo
from homeassistant.components import otbr, thread
from homeassistant.components.thread import discovery
from homeassistant.config_entries import SOURCE_HASSIO, SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
@ -18,7 +19,6 @@ from homeassistant.setup import async_setup_component
from . import (
BASE_URL,
CONFIG_ENTRY_DATA_MULTIPAN,
CONFIG_ENTRY_DATA_THREAD,
DATASET_CH15,
DATASET_CH16,
DATASET_INSECURE_NW_KEY,
@ -71,6 +71,7 @@ async def test_import_dataset(
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
@ -138,6 +139,7 @@ async def test_import_share_radio_channel_collision(
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
with (
@ -177,6 +179,7 @@ async def test_import_share_radio_no_channel_collision(
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
with (
@ -214,6 +217,7 @@ async def test_import_insecure_dataset(
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
with (
@ -252,6 +256,7 @@ async def test_config_entry_not_ready(
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
get_active_dataset_tlvs.side_effect = error
@ -268,6 +273,7 @@ async def test_border_agent_id_not_supported(
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
get_border_agent_id.side_effect = python_otbr_api.GetBorderAgentIdNotSupportedError
@ -281,6 +287,7 @@ async def test_config_entry_update(hass: HomeAssistant) -> None:
domain=otbr.DOMAIN,
options={},
title="My OTBR",
unique_id=TEST_BORDER_AGENT_EXTENDED_ADDRESS.hex(),
)
config_entry.add_to_hass(hass)
mock_api = MagicMock()
@ -314,25 +321,33 @@ async def test_remove_entry(
await hass.config_entries.async_remove(config_entry.entry_id)
async def test_remove_extra_entries(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
@pytest.mark.parametrize(
("source", "unique_id", "updated_unique_id"),
[
(SOURCE_HASSIO, None, None),
(SOURCE_HASSIO, "abcd", "abcd"),
(SOURCE_USER, None, TEST_BORDER_AGENT_ID.hex()),
(SOURCE_USER, "abcd", TEST_BORDER_AGENT_ID.hex()),
],
)
async def test_update_unique_id(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
source: str,
unique_id: str | None,
updated_unique_id: str | None,
) -> None:
"""Test we remove additional config entries."""
"""Test we update the unique id if extended address has changed."""
config_entry1 = MockConfigEntry(
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA_MULTIPAN,
domain=otbr.DOMAIN,
options={},
source=source,
title="Open Thread Border Router",
unique_id=unique_id,
)
config_entry2 = MockConfigEntry(
data=CONFIG_ENTRY_DATA_THREAD,
domain=otbr.DOMAIN,
options={},
title="Open Thread Border Router",
)
config_entry1.add_to_hass(hass)
config_entry2.add_to_hass(hass)
assert len(hass.config_entries.async_entries(otbr.DOMAIN)) == 2
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, otbr.DOMAIN, {})
assert len(hass.config_entries.async_entries(otbr.DOMAIN)) == 1
config_entry = hass.config_entries.async_get_entry(config_entry.entry_id)
assert config_entry.unique_id == updated_unique_id

21
tests/components/otbr/test_silabs_multiprotocol.py
View File

@ -5,7 +5,6 @@ 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,
)
@ -127,10 +126,11 @@ async def test_async_change_channel_no_otbr(hass: HomeAssistant) -> None:
async def test_async_change_channel_non_matching_url(
hass: HomeAssistant, otbr_config_entry_multipan
hass: HomeAssistant, otbr_config_entry_multipan: str
) -> None:
"""Test async_change_channel when otbr is not configured."""
hass.data[otbr.DATA_OTBR].url = OTBR_NON_MULTIPAN_URL
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
config_entry.runtime_data.url = OTBR_NON_MULTIPAN_URL
with patch("python_otbr_api.OTBR.set_channel") as mock_set_channel:
await otbr_silabs_multiprotocol.async_change_channel(hass, 16, delay=0)
mock_set_channel.assert_not_awaited()
@ -184,10 +184,11 @@ async def test_async_get_channel_no_otbr(hass: HomeAssistant) -> None:
async def test_async_get_channel_non_matching_url(
hass: HomeAssistant, otbr_config_entry_multipan
hass: HomeAssistant, otbr_config_entry_multipan: str
) -> None:
"""Test async_change_channel when otbr is not configured."""
hass.data[otbr.DATA_OTBR].url = OTBR_NON_MULTIPAN_URL
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
config_entry.runtime_data.url = OTBR_NON_MULTIPAN_URL
with patch("python_otbr_api.OTBR.get_active_dataset") as mock_get_active_dataset:
assert await otbr_silabs_multiprotocol.async_get_channel(hass) is None
mock_get_active_dataset.assert_not_awaited()
@ -198,10 +199,11 @@ async def test_async_get_channel_non_matching_url(
[(OTBR_MULTIPAN_URL, True), (OTBR_NON_MULTIPAN_URL, False)],
)
async def test_async_using_multipan(
hass: HomeAssistant, otbr_config_entry_multipan, url: str, expected: bool
hass: HomeAssistant, otbr_config_entry_multipan: str, url: str, expected: bool
) -> None:
"""Test async_change_channel when otbr is not configured."""
hass.data[otbr.DATA_OTBR].url = url
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
config_entry.runtime_data.url = url
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is expected
@ -213,8 +215,9 @@ async def test_async_using_multipan_no_otbr(hass: HomeAssistant) -> None:
async def test_async_using_multipan_non_matching_url(
hass: HomeAssistant, otbr_config_entry_multipan
hass: HomeAssistant, otbr_config_entry_multipan: str
) -> None:
"""Test async_change_channel when otbr is not configured."""
hass.data[otbr.DATA_OTBR].url = OTBR_NON_MULTIPAN_URL
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
config_entry.runtime_data.url = OTBR_NON_MULTIPAN_URL
assert await otbr_silabs_multiprotocol.async_using_multipan(hass) is False

33
tests/components/otbr/test_util.py
View File

@ -1,6 +1,6 @@
"""Test OTBR Utility functions."""
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pytest
import python_otbr_api
@ -31,24 +31,37 @@ async def test_get_allowed_channel(
assert await otbr.util.get_allowed_channel(hass, OTBR_NON_MULTIPAN_URL) is None
async def test_factory_reset(hass: HomeAssistant, otbr_config_entry_multipan) -> None:
async def test_factory_reset(
hass: HomeAssistant,
otbr_config_entry_multipan: str,
get_border_agent_id: AsyncMock,
) -> None:
"""Test factory_reset."""
new_ba_id = b"new_ba_id"
get_border_agent_id.return_value = new_ba_id
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
assert config_entry.unique_id != new_ba_id.hex()
with (
patch("python_otbr_api.OTBR.factory_reset") as factory_reset_mock,
patch(
"python_otbr_api.OTBR.delete_active_dataset"
) as delete_active_dataset_mock,
):
await hass.data[otbr.DATA_OTBR].factory_reset()
await config_entry.runtime_data.factory_reset(hass)
delete_active_dataset_mock.assert_not_called()
factory_reset_mock.assert_called_once_with()
# Check the unique_id is updated
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
assert config_entry.unique_id == new_ba_id.hex()
async def test_factory_reset_not_supported(
hass: HomeAssistant, otbr_config_entry_multipan
hass: HomeAssistant, otbr_config_entry_multipan: str
) -> None:
"""Test factory_reset."""
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
with (
patch(
"python_otbr_api.OTBR.factory_reset",
@ -58,16 +71,17 @@ async def test_factory_reset_not_supported(
"python_otbr_api.OTBR.delete_active_dataset"
) as delete_active_dataset_mock,
):
await hass.data[otbr.DATA_OTBR].factory_reset()
await config_entry.runtime_data.factory_reset(hass)
delete_active_dataset_mock.assert_called_once_with()
factory_reset_mock.assert_called_once_with()
async def test_factory_reset_error_1(
hass: HomeAssistant, otbr_config_entry_multipan
hass: HomeAssistant, otbr_config_entry_multipan: str
) -> None:
"""Test factory_reset."""
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
with (
patch(
"python_otbr_api.OTBR.factory_reset",
@ -80,16 +94,17 @@ async def test_factory_reset_error_1(
HomeAssistantError,
),
):
await hass.data[otbr.DATA_OTBR].factory_reset()
await config_entry.runtime_data.factory_reset(hass)
delete_active_dataset_mock.assert_not_called()
factory_reset_mock.assert_called_once_with()
async def test_factory_reset_error_2(
hass: HomeAssistant, otbr_config_entry_multipan
hass: HomeAssistant, otbr_config_entry_multipan: str
) -> None:
"""Test factory_reset."""
config_entry = hass.config_entries.async_get_entry(otbr_config_entry_multipan)
with (
patch(
"python_otbr_api.OTBR.factory_reset",
@ -103,7 +118,7 @@ async def test_factory_reset_error_2(
HomeAssistantError,
),
):
await hass.data[otbr.DATA_OTBR].factory_reset()
await config_entry.runtime_data.factory_reset(hass)
delete_active_dataset_mock.assert_called_once_with()
factory_reset_mock.assert_called_once_with()