From 20e92d24cd4d044578c05a7a9affb7c793d1a8cb Mon Sep 17 00:00:00 2001 From: Christian Glombek Date: Mon, 27 Jul 2026 06:55:14 +0200 Subject: [PATCH 1/2] Import Thread datasets from Matter border routers A router implementing the Network Infrastructure Manager device type carries the Thread Border Router Management cluster, which lets an authorised member of the Matter fabric read the active operational dataset over an authenticated session. That is an alternative to reading the same credentials from a vendor specific API such as the OpenThread Border Router REST interface. When a node exposes a border router endpoint, read its active dataset by command and add it to the Thread dataset store with source "matter". The read is keyed on the active dataset timestamp so node updates do not reissue the command while the dataset is unchanged, and an attribute subscription on that timestamp re-imports when the dataset changes underneath a quiet node, which is exactly what a scheduled migration does: without it the change would go unnoticed until the next interview. Three details are load bearing and are covered by tests: - The Matter client returns command responses as plain dicts with octet strings base64 encoded, so the dataset cannot be read off the response as an attribute. - A border router with no Thread stack running reports ExtAddress as NullValue rather than omitting it. NullValue is not falsy, and the store rejects a preferred border agent ID that is not accompanied by an extended address, so it has to be excluded explicitly. - The get_node mock in the Matter test helpers found the requested node and then returned None; this is its first caller, so the helper is fixed here. Thread is declared as an after dependency rather than a dependency. Matter does not require Thread, and making it a hard dependency would set up the Thread integration, and so zeroconf, for every Matter config entry. Assisted-By: Claude Opus 5 --- homeassistant/components/matter/adapter.py | 173 ++++++++++++- homeassistant/components/matter/manifest.json | 2 +- .../components/matter/thread_border_router.py | 146 +++++++++++ .../fixtures/nodes/thread_border_router.json | 46 ++++ .../matter/test_thread_border_router.py | 229 ++++++++++++++++++ 5 files changed, 593 insertions(+), 3 deletions(-) create mode 100644 homeassistant/components/matter/thread_border_router.py create mode 100644 tests/components/matter/fixtures/nodes/thread_border_router.json create mode 100644 tests/components/matter/test_thread_border_router.py 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..cd62d88549dba5 --- /dev/null +++ b/tests/components/matter/test_thread_border_router.py @@ -0,0 +1,229 @@ +"""Test importing Thread datasets from Matter border routers.""" + +from base64 import b64encode +from collections.abc import Generator +from unittest.mock import AsyncMock, MagicMock, patch + +from chip.clusters import Objects as clusters +from chip.clusters.Types import NullValue +from matter_server.common.models import EventType +import pytest + +from homeassistant.components.matter.thread_border_router import async_import_dataset +from homeassistant.components.thread import dataset_store +from homeassistant.core import HomeAssistant + +from .common import ( + set_node_attribute, + setup_integration_with_node_fixture, + trigger_subscription_callback, +) + +from tests.components.thread import DATASET_1 + + +@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. + """ + with patch.object(dataset_store, "BORDER_AGENT_DISCOVERY_TIMEOUT", 0.05): + 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 From 9199bfb97a7231165b48fdc5b65c56f14907fd71 Mon Sep 17 00:00:00 2001 From: Christian Glombek Date: Mon, 27 Jul 2026 15:27:18 +0200 Subject: [PATCH 2/2] Test Matter dataset import coexistence with otbr and its failure paths The store already handles two integrations writing to it: async_add() deduplicates on dataset and extended PAN ID, and the preference setter only acts when none is stored or the border agent ID matches. Assert that importing over Matter leaves an otbr-seeded entry alone, and cover the NodeNotReady retry (including cancellation on unload and replacement on re-arm), state cleanup on endpoint and node removal, and the response shapes the dataset read accepts. Assisted-By: Claude Opus 5 --- .../matter/test_thread_border_router.py | 361 +++++++++++++++++- 1 file changed, 358 insertions(+), 3 deletions(-) diff --git a/tests/components/matter/test_thread_border_router.py b/tests/components/matter/test_thread_border_router.py index cd62d88549dba5..b65f5cd59fa040 100644 --- a/tests/components/matter/test_thread_border_router.py +++ b/tests/components/matter/test_thread_border_router.py @@ -2,15 +2,19 @@ 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 dataset_store +from homeassistant.components.thread import async_add_dataset, dataset_store from homeassistant.core import HomeAssistant from .common import ( @@ -19,7 +23,8 @@ trigger_subscription_callback, ) -from tests.components.thread import DATASET_1 +from tests.common import async_fire_time_changed +from tests.components.thread import DATASET_1, DATASET_2 @pytest.fixture(autouse=True) @@ -30,7 +35,9 @@ def short_border_agent_discovery_timeout() -> Generator[None]: teardown waits for it; the wait is the thread integration's concern, not this one's. """ - with patch.object(dataset_store, "BORDER_AGENT_DISCOVERY_TIMEOUT", 0.05): + # 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 @@ -227,3 +234,351 @@ async def test_null_ext_address_omits_preference_pair( 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