diff --git a/homeassistant/components/matter/adapter.py b/homeassistant/components/matter/adapter.py index 72246f72d0030c..0ef31981757036 100644 --- a/homeassistant/components/matter/adapter.py +++ b/homeassistant/components/matter/adapter.py @@ -1,20 +1,32 @@ """Matter to Home Assistant adapter.""" -from typing import TYPE_CHECKING, cast +from collections.abc import Callable +from typing import TYPE_CHECKING, Any, cast from chip.clusters import Objects as clusters from matter_server.client.models.device_types import BridgedNode +from matter_server.common.errors import NodeNotReady from matter_server.common.models import EventType, ServerInfoMessage from homeassistant.const import Platform -from homeassistant.core import HomeAssistant +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers import device_registry as dr from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.event import async_call_later from homeassistant.helpers.typing import UNDEFINED, UndefinedType from .const import DOMAIN, ID_TYPE_DEVICE_ID, ID_TYPE_SERIAL, LOGGER from .discovery import async_discover_entities from .helpers import MatterConfigEntry, get_device_endpoint, get_device_id +from .thread_border_router import ( + async_import_dataset, + get_active_dataset_timestamp, + get_active_dataset_timestamp_path, + get_border_router_endpoints, + get_extended_address, +) + +THREAD_DATASET_RETRY_DELAY = 30 # seconds if TYPE_CHECKING: from matter_server.client import MatterClient @@ -44,6 +56,17 @@ def __init__( self.config_entry = config_entry self.platform_handlers: dict[Platform, AddEntitiesCallback] = {} self.discovered_entities: set[str] = set() + # (node_id, endpoint_id) -> (active dataset timestamp, extended + # address) as of the last import; the address is part of the key + # because a router can regenerate it without touching the dataset, + # and the store tracks it per entry. + self._thread_dataset_timestamps: dict[tuple[int, int], tuple[Any, Any]] = {} + # border router endpoints with an ActiveDatasetTimestamp subscription + self._thread_dataset_subscriptions: dict[ + tuple[int, int], Callable[[], None] + ] = {} + # pending NodeNotReady retries, one timer per endpoint + self._thread_dataset_retries: dict[tuple[int, int], Callable[[], None]] = {} def register_platform_handler( self, platform: Platform, add_entities: AddEntitiesCallback @@ -53,6 +76,17 @@ def register_platform_handler( async def setup_nodes(self) -> None: """Set up all existing nodes and subscribe to new nodes.""" + + def unsubscribe_thread_dataset_updates() -> None: + while self._thread_dataset_subscriptions: + _, unsubscribe = self._thread_dataset_subscriptions.popitem() + unsubscribe() + while self._thread_dataset_retries: + _, cancel = self._thread_dataset_retries.popitem() + cancel() + + self.config_entry.async_on_unload(unsubscribe_thread_dataset_updates) + for node in self.matter_client.get_nodes(): self._setup_node(node) @@ -81,9 +115,18 @@ def endpoint_added_callback(event: EventType, data: dict[str, int]) -> None: ): self._setup_endpoint(node.endpoints[0]) self._setup_endpoint(endpoint) + # An endpoint can be delivered on its own, without any node-level + # event; a border router arriving this way still has to be read. + self._schedule_thread_dataset_import(node) def endpoint_removed_callback(event: EventType, data: dict[str, int]) -> None: """Handle endpoint removed event.""" + key = (data["node_id"], data["endpoint_id"]) + if unsubscribe := self._thread_dataset_subscriptions.pop(key, None): + unsubscribe() + if cancel_retry := self._thread_dataset_retries.pop(key, None): + cancel_retry() + self._thread_dataset_timestamps.pop(key, None) server_info = cast(ServerInfoMessage, self.matter_client.server_info) try: node = self.matter_client.get_node(data["node_id"]) @@ -106,6 +149,18 @@ def endpoint_removed_callback(event: EventType, data: dict[str, int]) -> None: def node_removed_callback(event: EventType, node_id: int) -> None: """Handle node removed event.""" + # The client may already have evicted the node, in which case the + # endpoint iteration below never happens; the Thread import state + # has to go regardless, or a node reusing the id with an unchanged + # dataset would have its import suppressed. + for key in [ + k for k in self._thread_dataset_subscriptions if k[0] == node_id + ]: + self._thread_dataset_subscriptions.pop(key)() + for key in [k for k in self._thread_dataset_timestamps if k[0] == node_id]: + del self._thread_dataset_timestamps[key] + for key in [k for k in self._thread_dataset_retries if k[0] == node_id]: + self._thread_dataset_retries.pop(key)() try: node = self.matter_client.get_node(node_id) except KeyError: @@ -132,6 +187,7 @@ def node_removed_callback(event: EventType, node_id: int) -> None: callback=node_removed_callback, event_filter=EventType.NODE_REMOVED ) ) + self.config_entry.async_on_unload( self.matter_client.subscribe_events( callback=node_added_callback, event_filter=EventType.NODE_ADDED @@ -161,6 +217,119 @@ def _setup_node(self, node: MatterNode) -> None: node.node_id, err, ) + # Outside the catch-all above so the datasets of a node whose entity + # setup failed are still read; the read runs in a background task, so + # its errors surface on their own terms either way. + self._schedule_thread_dataset_import(node) + + def _schedule_thread_dataset_import(self, node: MatterNode) -> None: + """Import Thread datasets from any border router endpoints on this node. + + Reading the dataset needs an await and this runs from synchronous + callbacks, so the work is scheduled as a background task. + """ + # An unavailable node cannot answer the read, and setup_nodes() visits + # cached offline nodes too: without this gate every visit ends in + # NodeNotReady and re-arms the retry, polling an offline border router + # indefinitely. The node-updated path schedules the import once the + # node comes back. + if not node.available: + return + for endpoint in get_border_router_endpoints(node): + key = (node.node_id, endpoint.endpoint_id) + self._subscribe_thread_dataset_updates(node, endpoint) + state = ( + get_active_dataset_timestamp(endpoint), + get_extended_address(endpoint), + ) + # _setup_node also runs on every node update; only re-read when + # the timestamp shows the dataset changed or the router's + # extended address moved underneath the same dataset. + if self._thread_dataset_timestamps.get(key, object()) == state: + continue + self._thread_dataset_timestamps[key] = state + self.config_entry.async_create_background_task( + self.hass, + self._import_thread_dataset(key, endpoint), + name=f"matter_thread_dataset_{node.node_id}_{endpoint.endpoint_id}", + ) + + async def _import_thread_dataset( + self, key: tuple[int, int], endpoint: MatterEndpoint + ) -> None: + """Import the dataset, forgetting the timestamp when the read fails. + + The timestamp is recorded before the read to deduplicate concurrent + triggers, but a failed read must not count as done, or a transient + error (a server reconnect, say) would suppress the import until the + dataset changes again. + """ + try: + await async_import_dataset(self.hass, self.matter_client, endpoint) + except NodeNotReady: + # The node is mid-resubscription after a restart; the availability + # event can even arrive while it is still not ready, so a plain + # retrigger is not enough. Try again once things have settled. + self._thread_dataset_timestamps.pop(key, None) + + @callback + def _retry(_now: Any) -> None: + self._thread_dataset_retries.pop(key, None) + try: + node = self.matter_client.get_node(key[0]) + except KeyError: + return # node removed meanwhile + self._schedule_thread_dataset_import(node) + + # One timer per endpoint, replaced on re-arm: registering each + # one-shot timer for unload instead would retain a callback per + # retry cycle for the entry's lifetime. + if cancel_previous := self._thread_dataset_retries.pop(key, None): + cancel_previous() + self._thread_dataset_retries[key] = async_call_later( + self.hass, THREAD_DATASET_RETRY_DELAY, _retry + ) + except ValueError: + # A malformed response is not an unprovisioned router: forget the + # timestamp so the next trigger re-reads instead of trusting it. + self._thread_dataset_timestamps.pop(key, None) + LOGGER.warning( + "Border router %s returned an unusable dataset response", key + ) + except Exception: + self._thread_dataset_timestamps.pop(key, None) + raise + + def _subscribe_thread_dataset_updates( + self, node: MatterNode, endpoint: MatterEndpoint + ) -> None: + """Re-import this border router's dataset when its timestamp changes. + + Node updates only happen on interviews, so a dataset changed by a + scheduled migration would otherwise go unnoticed until the next + interview or restart. The client dispatches attribute events by path; + the timestamp bookkeeping in _schedule_thread_dataset_import decides + whether a read is due. + """ + key = (node.node_id, endpoint.endpoint_id) + if key in self._thread_dataset_subscriptions: + return + + def timestamp_updated_callback(event: EventType, data: Any) -> None: + try: + updated_node = self.matter_client.get_node(node.node_id) + except KeyError: + return # race condition + self._schedule_thread_dataset_import(updated_node) + + # Kept per endpoint so removal can unsubscribe; anything still + # subscribed when the entry unloads is released in one sweep there. + self._thread_dataset_subscriptions[key] = self.matter_client.subscribe_events( + callback=timestamp_updated_callback, + event_filter=EventType.ATTRIBUTE_UPDATED, + node_filter=node.node_id, + attr_path_filter=get_active_dataset_timestamp_path(endpoint), + ) def _create_device_registry( self, diff --git a/homeassistant/components/matter/manifest.json b/homeassistant/components/matter/manifest.json index c007c8ee8eeffd..bd29e41f7e8507 100644 --- a/homeassistant/components/matter/manifest.json +++ b/homeassistant/components/matter/manifest.json @@ -1,7 +1,7 @@ { "domain": "matter", "name": "Matter", - "after_dependencies": ["bluetooth", "hassio"], + "after_dependencies": ["bluetooth", "hassio", "thread"], "codeowners": ["@home-assistant/matter"], "config_flow": true, "dependencies": ["websocket_api"], diff --git a/homeassistant/components/matter/thread_border_router.py b/homeassistant/components/matter/thread_border_router.py new file mode 100644 index 00000000000000..dfcffa4230f48e --- /dev/null +++ b/homeassistant/components/matter/thread_border_router.py @@ -0,0 +1,146 @@ +"""Import Thread operational datasets from Matter border routers. + +A router that implements the Network Infrastructure Manager device type carries +the Thread Border Router Management cluster, which lets an authorised member of +the fabric read the active operational dataset over Matter. That replaces +reading the same credentials from a vendor specific API such as the OpenThread +Border Router REST interface. +""" + +from base64 import b64decode +from typing import TYPE_CHECKING, Any + +from chip.clusters import Objects as clusters +from chip.clusters.Types import NullValue +from matter_server.client.models import device_types +from matter_server.common.helpers.util import create_attribute_path + +from homeassistant.components.thread import async_add_dataset +from homeassistant.core import HomeAssistant + +from .const import DOMAIN, LOGGER + +if TYPE_CHECKING: + from matter_server.client import MatterClient + from matter_server.client.models.node import MatterEndpoint, MatterNode + +# A border router may present either device type; both carry the TBRM cluster. +BORDER_ROUTER_DEVICE_TYPES = { + device_types.NetworkInfrastructureManager.device_type, + device_types.ThreadBorderRouter.device_type, +} + + +def get_active_dataset_timestamp_path(endpoint: MatterEndpoint) -> str: + """Return the attribute path of ActiveDatasetTimestamp on this endpoint.""" + return create_attribute_path( + endpoint.endpoint_id, + clusters.ThreadBorderRouterManagement.id, + clusters.ThreadBorderRouterManagement.Attributes.ActiveDatasetTimestamp.attribute_id, + ) + + +def get_border_router_endpoints(node: MatterNode) -> list[MatterEndpoint]: + """Return the endpoints of a node that expose a Thread border router.""" + return [ + endpoint + for endpoint in node.endpoints.values() + if endpoint.has_cluster(clusters.ThreadBorderRouterManagement) + and any( + device_type.device_type in BORDER_ROUTER_DEVICE_TYPES + for device_type in endpoint.device_types + ) + ] + + +def get_extended_address(endpoint: MatterEndpoint) -> Any: + """Return the border router's Thread extended address attribute value. + + NullValue when the Thread stack is not running; None when the attribute + is absent. + """ + return endpoint.get_attribute_value( + None, clusters.ThreadNetworkDiagnostics.Attributes.ExtAddress + ) + + +def get_active_dataset_timestamp(endpoint: MatterEndpoint) -> int | None: + """Return the active dataset timestamp, which changes when the dataset does.""" + timestamp: int | None = endpoint.get_attribute_value( + None, clusters.ThreadBorderRouterManagement.Attributes.ActiveDatasetTimestamp + ) + return timestamp + + +def _dataset_from_response(response: Any) -> bytes: + """Return the dataset carried by a DatasetResponse. + + The Matter client hands back command responses as plain dicts, with octet + strings base64 encoded rather than as bytes, so the payload cannot be read + off the response as an attribute. Object and bytes forms are still accepted + so this keeps working if that representation changes. + + Raises ValueError for a response that carries no readable dataset: a + malformed reply must not be mistaken for an unprovisioned border router, + or the import would be considered done and not tried again. + """ + if isinstance(response, dict): + raw = response.get("dataset") + else: + raw = getattr(response, "dataset", None) + if isinstance(raw, str): + return b64decode(raw, validate=True) + if isinstance(raw, bytes): + return raw + raise ValueError("response carries no dataset payload") + + +async def async_import_dataset( + hass: HomeAssistant, matter_client: MatterClient, endpoint: MatterEndpoint +) -> None: + """Read the active dataset from a border router and add it to the store. + + The dataset is only reachable by command; the identifiers used to mark the + preferred border agent are plain attributes on the same endpoint. + """ + response: Any = await matter_client.send_device_command( + node_id=endpoint.node.node_id, + endpoint_id=endpoint.endpoint_id, + command=clusters.ThreadBorderRouterManagement.Commands.GetActiveDatasetRequest(), + ) + dataset = _dataset_from_response(response) + + if not dataset: + # A border router that has not formed or joined a network answers with + # an empty dataset, which the dataset store would reject for lacking an + # active timestamp. + LOGGER.debug( + "Border router on node %s endpoint %s has no active dataset", + endpoint.node.node_id, + endpoint.endpoint_id, + ) + return + + border_agent_id: bytes | None = endpoint.get_attribute_value( + None, clusters.ThreadBorderRouterManagement.Attributes.BorderAgentID + ) + ext_address: int | None = get_extended_address(endpoint) + + # The store refuses a preferred border agent ID that is not accompanied by an + # extended address, so only mark a preference when both are known. A border + # router with no Thread stack running reports ExtAddress as NullValue rather + # than omitting it, which is not falsy and must be excluded explicitly. + preferred: dict[str, str] = {} + if border_agent_id and ext_address not in (None, NullValue): + preferred = { + "preferred_border_agent_id": border_agent_id.hex(), + "preferred_extended_address": ext_address.to_bytes(8, "big").hex(), + } + + await async_add_dataset(hass, DOMAIN, dataset.hex(), **preferred) + + LOGGER.debug( + "Imported Thread dataset from node %s endpoint %s", + endpoint.node.node_id, + endpoint.endpoint_id, + ) diff --git a/tests/components/matter/fixtures/nodes/thread_border_router.json b/tests/components/matter/fixtures/nodes/thread_border_router.json new file mode 100644 index 00000000000000..5d86c3bc15ccc6 --- /dev/null +++ b/tests/components/matter/fixtures/nodes/thread_border_router.json @@ -0,0 +1,46 @@ +{ + "node_id": 90, + "date_commissioned": "2026-07-27T10:00:00.000000", + "last_interview": "2026-07-27T10:00:00.000000", + "interview_version": 6, + "available": true, + "is_bridge": false, + "attributes": { + "0/29/0": [ + { + "0": 22, + "1": 5 + } + ], + "0/29/3": [1], + "0/40/1": "OpenWrt", + "0/40/2": 65521, + "0/40/3": "OpenWrt OTBR", + "0/40/5": "Border Router", + "0/40/6": "XX", + "0/40/7": 1, + "0/40/8": "1.0", + "0/40/9": 1, + "0/40/10": "2026.7.0", + "0/40/18": "DEADBEEF00000090", + "1/29/0": [ + { + "0": 144, + "1": 2 + } + ], + "1/29/1": [53, 1105, 1106, 1107], + "1/53/0": null, + "1/53/1": 0, + "1/1106/0": "OpenWrt OTBR", + "1/1106/1": "AQIDBAUGBwgJCgsMDQ4PEA==", + "1/1106/2": 5, + "1/1106/3": true, + "1/1106/4": 1, + "1/1106/5": null, + "1/1106/65532": 0, + "1/1106/65533": 1, + "1/53/63": 1234605616436508552 + }, + "attribute_subscriptions": [] +} diff --git a/tests/components/matter/test_thread_border_router.py b/tests/components/matter/test_thread_border_router.py new file mode 100644 index 00000000000000..b65f5cd59fa040 --- /dev/null +++ b/tests/components/matter/test_thread_border_router.py @@ -0,0 +1,584 @@ +"""Test importing Thread datasets from Matter border routers.""" + +from base64 import b64encode +from collections.abc import Generator +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +from chip.clusters import Objects as clusters +from chip.clusters.Types import NullValue +from freezegun.api import FrozenDateTimeFactory +from matter_server.common.errors import NodeNotReady +from matter_server.common.models import EventType +import pytest + +from homeassistant.components.matter.adapter import THREAD_DATASET_RETRY_DELAY +from homeassistant.components.matter.thread_border_router import async_import_dataset +from homeassistant.components.thread import async_add_dataset, dataset_store +from homeassistant.core import HomeAssistant + +from .common import ( + set_node_attribute, + setup_integration_with_node_fixture, + trigger_subscription_callback, +) + +from tests.common import async_fire_time_changed +from tests.components.thread import DATASET_1, DATASET_2 + + +@pytest.fixture(autouse=True) +def short_border_agent_discovery_timeout() -> Generator[None]: + """Keep the dataset store's 30 s discovery wait out of these tests. + + Every import starts _set_preferred_dataset_if_only_network, and test + teardown waits for it; the wait is the thread integration's concern, not + this one's. + """ + # Zero rather than merely small: under a frozen clock a nonzero asyncio + # timeout never elapses on its own. + with patch.object(dataset_store, "BORDER_AGENT_DISCOVERY_TIMEOUT", 0): + yield + + +# Reuse a known-good dataset from the Thread integration's own tests; the store +# parses it and rejects anything lacking an active timestamp. +DATASET_TLV = bytes.fromhex(DATASET_1) + +BORDER_AGENT_ID = "0102030405060708090a0b0c0d0e0f10" +EXT_ADDRESS_HEX = "1122334455667788" + + +@pytest.fixture(autouse=True) +def mock_thread_discovery(mock_async_zeroconf: MagicMock) -> MagicMock: + """Adding a dataset starts Thread discovery, which would open a real socket.""" + return mock_async_zeroconf + + +@pytest.fixture(name="dataset_response") +def dataset_response_fixture(matter_client: MagicMock) -> dict[str, str]: + """Make GetActiveDatasetRequest return a dataset. + + The Matter client returns command responses as dicts with octet strings + base64 encoded, which is what a real border router produces. + """ + response = {"dataset": b64encode(DATASET_TLV).decode()} + matter_client.send_device_command.return_value = response + return response + + +async def test_dataset_imported_from_border_router( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """A border router's active dataset is added to the Thread dataset store.""" + await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + # The dataset is only reachable by command, not as an attribute. + assert matter_client.send_device_command.called + command = matter_client.send_device_command.call_args.kwargs["command"] + assert isinstance( + command, + clusters.ThreadBorderRouterManagement.Commands.GetActiveDatasetRequest, + ) + + store = await dataset_store.async_get_store(hass) + entries = list(store.datasets.values()) + assert len(entries) == 1 + + entry = entries[0] + assert entry.source == "matter" + assert entry.tlv == DATASET_TLV.hex() + # Both preference fields must be set together, or the store raises. + assert entry.preferred_border_agent_id == BORDER_AGENT_ID + assert entry.preferred_extended_address == EXT_ADDRESS_HEX + + +async def test_empty_dataset_is_not_imported( + hass: HomeAssistant, matter_client: MagicMock +) -> None: + """An unprovisioned border router returns an empty dataset and is skipped.""" + matter_client.send_device_command.return_value = {"dataset": ""} + + await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 0 + + +async def test_non_border_router_node_is_ignored( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """A node without the TBRM cluster never triggers a dataset read.""" + await setup_integration_with_node_fixture(hass, "eve_contact_sensor", matter_client) + await hass.async_block_till_done() + + for call in matter_client.send_device_command.call_args_list: + assert not isinstance( + call.kwargs.get("command"), + clusters.ThreadBorderRouterManagement.Commands.GetActiveDatasetRequest, + ) + + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 0 + + +async def test_dataset_not_reread_when_timestamp_unchanged( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """Node updates must not re-issue the command while the dataset is unchanged.""" + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + def dataset_reads() -> int: + return sum( + isinstance( + call.kwargs.get("command"), + clusters.ThreadBorderRouterManagement.Commands.GetActiveDatasetRequest, + ) + for call in matter_client.send_device_command.call_args_list + ) + + assert dataset_reads() == 1 + + # A node update with an unchanged dataset must not cause another read. + set_node_attribute(node, 1, 1106, 3, False) + await trigger_subscription_callback( + hass, matter_client, EventType.NODE_UPDATED, node + ) + assert dataset_reads() == 1 + + # A new active dataset timestamp must. + set_node_attribute(node, 1, 1106, 4, 2) + await trigger_subscription_callback( + hass, matter_client, EventType.NODE_UPDATED, node + ) + assert dataset_reads() == 2 + + +async def test_dataset_reread_on_attribute_update( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """A pushed timestamp change re-imports without waiting for an interview. + + Scheduled migrations change the dataset while the node is otherwise quiet: + the device reports the new ActiveDatasetTimestamp through the attribute + subscription, and no interview happens. + """ + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + def dataset_reads() -> int: + return sum( + isinstance( + call.kwargs.get("command"), + clusters.ThreadBorderRouterManagement.Commands.GetActiveDatasetRequest, + ) + for call in matter_client.send_device_command.call_args_list + ) + + assert dataset_reads() == 1 + + # An event without a timestamp change must not cause a read. (The real + # client only dispatches this callback for the subscribed attribute path; + # the test helper fires every subscription, so this also exercises the + # timestamp bookkeeping.) + await trigger_subscription_callback( + hass, matter_client, EventType.ATTRIBUTE_UPDATED, None + ) + assert dataset_reads() == 1 + + # A pushed ActiveDatasetTimestamp change must. + set_node_attribute(node, 1, 1106, 4, 3) + await trigger_subscription_callback( + hass, matter_client, EventType.ATTRIBUTE_UPDATED, 3 + ) + assert dataset_reads() == 2 + + +async def test_null_ext_address_omits_preference_pair( + hass: HomeAssistant, matter_client: MagicMock +) -> None: + """A border router with no Thread stack reports ExtAddress as NullValue. + + NullValue is neither None nor falsy, so it has to be excluded explicitly or + the store raises for a border agent ID without an extended address. + """ + matter_client.send_device_command = AsyncMock( + return_value={"dataset": b64encode(DATASET_TLV).decode()} + ) + + endpoint = MagicMock() + endpoint.node.node_id = 1 + endpoint.endpoint_id = 1 + endpoint.get_attribute_value.side_effect = lambda _, attr: { + clusters.ThreadBorderRouterManagement.Attributes.BorderAgentID: b"\x01" * 16, + clusters.ThreadNetworkDiagnostics.Attributes.ExtAddress: NullValue, + }[attr] + + with patch( + "homeassistant.components.matter.thread_border_router.async_add_dataset" + ) as add_dataset: + await async_import_dataset(hass, matter_client, endpoint) + + add_dataset.assert_called_once() + kwargs = add_dataset.call_args.kwargs + assert "preferred_border_agent_id" not in kwargs + assert "preferred_extended_address" not in kwargs + + +async def test_coexists_with_the_otbr_integration( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """Importing over Matter must not disturb a dataset the otbr integration owns. + + Both integrations write to the same store. The otbr integration reaches a + border router over its REST API, this one reaches it over Matter, and a user + can have both. + """ + await async_add_dataset( + hass, + "otbr", + DATASET_TLV.hex(), + preferred_border_agent_id="aabbccddeeff00112233445566778899", + preferred_extended_address="8877665544332211", + ) + + await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + store = await dataset_store.async_get_store(hass) + entries = list(store.datasets.values()) + + # The same network must not be stored twice just because a second + # integration reported it. + assert len(entries) == 1 + + entry = entries[0] + assert entry.source == "otbr" + # A different border router must not take over the preference. The Matter + # node reports its own border agent, which is not the one otbr registered. + assert entry.preferred_border_agent_id == "aabbccddeeff00112233445566778899" + assert entry.preferred_extended_address == "8877665544332211" + + +async def test_dataset_read_retried_after_node_not_ready( + hass: HomeAssistant, + matter_client: MagicMock, + freezer: FrozenDateTimeFactory, +) -> None: + """A read failing with NodeNotReady is retried after the node settles. + + The failed read must not count as done: the timestamp is forgotten, so + the delayed retry actually re-reads instead of deduplicating itself away. + """ + # Seed a preferred dataset up front: with a preference in place the store + # skips its border-agent discovery task, whose wait never elapses under + # this test's frozen clock. + await async_add_dataset(hass, "test", DATASET_2) + store = await dataset_store.async_get_store(hass) + store.preferred_dataset = next(iter(store.datasets.values())).id + + matter_client.send_device_command.side_effect = NodeNotReady("node not ready") + + await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + assert len(store.datasets) == 1 + + matter_client.send_device_command.side_effect = None + matter_client.send_device_command.return_value = { + "dataset": b64encode(DATASET_TLV).decode() + } + + freezer.tick(THREAD_DATASET_RETRY_DELAY + 1) + async_fire_time_changed(hass) + await hass.async_block_till_done() + + assert len(store.datasets) == 2 + matter_entry = next(e for e in store.datasets.values() if e.source == "matter") + assert matter_entry.tlv == DATASET_TLV.hex() + + +async def test_dataset_retry_cancelled_on_unload( + hass: HomeAssistant, + matter_client: MagicMock, + freezer: FrozenDateTimeFactory, +) -> None: + """Unloading the entry cancels a pending NodeNotReady retry.""" + matter_client.send_device_command.side_effect = NodeNotReady("node not ready") + + await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + reads_before_unload = matter_client.send_device_command.call_count + + entry = hass.config_entries.async_entries("matter")[0] + assert await hass.config_entries.async_unload(entry.entry_id) + + matter_client.send_device_command.side_effect = None + freezer.tick(THREAD_DATASET_RETRY_DELAY + 1) + async_fire_time_changed(hass) + await hass.async_block_till_done() + + assert matter_client.send_device_command.call_count == reads_before_unload + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 0 + + +async def test_endpoint_removal_forgets_import_state( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """ENDPOINT_REMOVED releases the bookkeeping for that border router. + + With the subscription and timestamp gone, the same node re-announcing an + unchanged dataset is read again instead of deduplicated against state of + an endpoint that no longer exists. + """ + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + reads = matter_client.send_device_command.call_count + assert reads > 0 + + await trigger_subscription_callback( + hass, + matter_client, + event=EventType.ENDPOINT_REMOVED, + data={"node_id": node.node_id, "endpoint_id": 1}, + ) + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_UPDATED, data=node + ) + await hass.async_block_till_done() + assert matter_client.send_device_command.call_count == reads + 1 + + +async def test_node_removal_forgets_import_state( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """NODE_REMOVED cleans up every endpoint of the node.""" + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + reads = matter_client.send_device_command.call_count + + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_REMOVED, data=node.node_id + ) + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_ADDED, data=node + ) + await hass.async_block_till_done() + assert matter_client.send_device_command.call_count == reads + 1 + + +async def test_node_removal_after_client_eviction_forgets_import_state( + hass: HomeAssistant, matter_client: MagicMock, dataset_response: dict[str, str] +) -> None: + """The import state goes even when the client already evicted the node.""" + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + reads = matter_client.send_device_command.call_count + + original_get_node = matter_client.get_node.side_effect + matter_client.get_node.side_effect = KeyError(node.node_id) + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_REMOVED, data=node.node_id + ) + matter_client.get_node.side_effect = original_get_node + + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_ADDED, data=node + ) + await hass.async_block_till_done() + assert matter_client.send_device_command.call_count == reads + 1 + + +async def test_retry_skipped_when_node_removed_meanwhile( + hass: HomeAssistant, + matter_client: MagicMock, + freezer: FrozenDateTimeFactory, +) -> None: + """A pending retry whose node is gone by firing time does nothing.""" + await async_add_dataset(hass, "test", DATASET_2) + store = await dataset_store.async_get_store(hass) + store.preferred_dataset = next(iter(store.datasets.values())).id + + matter_client.send_device_command.side_effect = NodeNotReady("node not ready") + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + reads = matter_client.send_device_command.call_count + + original_get_node = matter_client.get_node.side_effect + matter_client.get_node.side_effect = KeyError(node.node_id) + freezer.tick(THREAD_DATASET_RETRY_DELAY + 1) + async_fire_time_changed(hass) + await hass.async_block_till_done() + matter_client.get_node.side_effect = original_get_node + + assert matter_client.send_device_command.call_count == reads + assert len(store.datasets) == 1 + + +async def test_retry_rearmed_not_duplicated( + hass: HomeAssistant, + matter_client: MagicMock, + freezer: FrozenDateTimeFactory, +) -> None: + """A second failure while a retry is pending replaces the timer. + + One timer per endpoint: when the delay finally elapses there is exactly + one retry read, not one per earlier failure. + """ + await async_add_dataset(hass, "test", DATASET_2) + store = await dataset_store.async_get_store(hass) + store.preferred_dataset = next(iter(store.datasets.values())).id + + matter_client.send_device_command.side_effect = NodeNotReady("node not ready") + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_UPDATED, data=node + ) + await hass.async_block_till_done() + reads_while_failing = matter_client.send_device_command.call_count + + matter_client.send_device_command.side_effect = None + matter_client.send_device_command.return_value = { + "dataset": b64encode(DATASET_TLV).decode() + } + freezer.tick(THREAD_DATASET_RETRY_DELAY + 1) + async_fire_time_changed(hass) + await hass.async_block_till_done() + + assert matter_client.send_device_command.call_count == reads_while_failing + 1 + assert len(store.datasets) == 2 + + +async def test_unusable_dataset_responses_are_skipped( + hass: HomeAssistant, matter_client: MagicMock +) -> None: + """Non-dict, undecodable and raw-bytes response shapes are all handled.""" + endpoint = MagicMock() + endpoint.node.node_id = 1 + endpoint.endpoint_id = 1 + endpoint.get_attribute_value.side_effect = lambda _, attr: { + clusters.ThreadBorderRouterManagement.Attributes.BorderAgentID: b"\x01" * 16, + clusters.ThreadNetworkDiagnostics.Attributes.ExtAddress: 0x1122334455667788, + }[attr] + + with patch( + "homeassistant.components.matter.thread_border_router.async_add_dataset" + ) as add_dataset: + # Malformed responses raise so the adapter does not consider the + # import done: an object without a usable dataset attribute, a string + # that is not base64, and non-alphabet characters prefixed to an + # otherwise valid value (the default decoder would silently discard + # them and import the corrupted remainder). + for malformed in ( + SimpleNamespace(dataset=None), + {"dataset": "%%%not-base64%%%"}, + {"dataset": "!" + b64encode(DATASET_TLV).decode()}, + ): + matter_client.send_device_command = AsyncMock(return_value=malformed) + with pytest.raises(ValueError): + await async_import_dataset(hass, matter_client, endpoint) + add_dataset.assert_not_called() + + # Raw bytes are accepted as they are. + matter_client.send_device_command = AsyncMock( + return_value={"dataset": DATASET_TLV} + ) + await async_import_dataset(hass, matter_client, endpoint) + add_dataset.assert_called_once() + + +async def test_offline_border_router_not_polled( + hass: HomeAssistant, + matter_client: MagicMock, + freezer: FrozenDateTimeFactory, +) -> None: + """An unavailable node is not read, and a pending retry stops on it. + + Without the availability gate every NodeNotReady retry re-arms itself, so + a cached offline border router would be polled every retry interval + indefinitely. + """ + await async_add_dataset(hass, "test", DATASET_2) + store = await dataset_store.async_get_store(hass) + store.preferred_dataset = next(iter(store.datasets.values())).id + + matter_client.send_device_command.side_effect = NodeNotReady("node not ready") + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + reads = matter_client.send_device_command.call_count + + node.node_data.available = False + freezer.tick(THREAD_DATASET_RETRY_DELAY + 1) + async_fire_time_changed(hass) + await hass.async_block_till_done() + assert matter_client.send_device_command.call_count == reads + + matter_client.send_device_command.side_effect = None + matter_client.send_device_command.return_value = { + "dataset": b64encode(DATASET_TLV).decode() + } + node.node_data.available = True + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_UPDATED, data=node + ) + await hass.async_block_till_done() + assert len(store.datasets) == 2 + + +async def test_malformed_response_is_retried_on_next_trigger( + hass: HomeAssistant, matter_client: MagicMock +) -> None: + """A malformed response must not count as done for the timestamp. + + Unlike a legitimately empty dataset, it leaves nothing imported and the + next trigger re-reads instead of deduplicating against the cached state. + """ + matter_client.send_device_command.return_value = {"dataset": "%%%not-base64%%%"} + node = await setup_integration_with_node_fixture( + hass, "thread_border_router", matter_client + ) + await hass.async_block_till_done() + + store = await dataset_store.async_get_store(hass) + assert len(store.datasets) == 0 + + matter_client.send_device_command.return_value = { + "dataset": b64encode(DATASET_TLV).decode() + } + await trigger_subscription_callback( + hass, matter_client, event=EventType.NODE_UPDATED, data=node + ) + await hass.async_block_till_done() + assert len(store.datasets) == 1