diff --git a/homeassistant/components/matter/api.py b/homeassistant/components/matter/api.py index a391f3115700..03ad7f526747 100644 --- a/homeassistant/components/matter/api.py +++ b/homeassistant/components/matter/api.py @@ -380,6 +380,7 @@ def _serialize_topology( return result +@websocket_api.require_admin @websocket_api.websocket_command( { vol.Required(TYPE): "matter/network_topology", @@ -402,6 +403,7 @@ async def websocket_network_topology( connection.send_result(msg[ID], _serialize_topology(hass, matter, topology)) +@websocket_api.require_admin @websocket_api.websocket_command( { vol.Required(TYPE): "matter/subscribe_network_topology", @@ -420,23 +422,44 @@ async def websocket_subscribe_network_topology( if not _topology_supported(connection, msg, matter): return + initial_sent = False + # updates are full snapshots, so only the newest buffered one matters + buffered: NetworkTopology | None = None + @callback def forward_topology(event: EventType, topology: NetworkTopology) -> None: + nonlocal buffered + if not initial_sent: + buffered = topology + return connection.send_message( websocket_api.event_message( msg[ID], _serialize_topology(hass, matter, topology) ) ) - # the initial fetch also opts this client in to topology events server-side - topology = await matter.matter_client.get_network_topology() - connection.subscriptions[msg[ID]] = matter.matter_client.subscribe_events( + # subscribe before the fetch: the fetch opts this client in server-side, + # and an update may arrive before the command result does + unsubscribe = matter.matter_client.subscribe_events( callback=forward_topology, event_filter=EventType.NETWORK_TOPOLOGY_UPDATED, ) + try: + topology = await matter.matter_client.get_network_topology() + except Exception: + unsubscribe() + raise + connection.subscriptions[msg[ID]] = unsubscribe connection.send_result(msg[ID]) connection.send_message( websocket_api.event_message( msg[ID], _serialize_topology(hass, matter, topology) ) ) + if buffered is not None: + connection.send_message( + websocket_api.event_message( + msg[ID], _serialize_topology(hass, matter, buffered) + ) + ) + initial_sent = True diff --git a/tests/components/matter/test_api.py b/tests/components/matter/test_api.py index 25b25668eeda..151515f7dae7 100644 --- a/tests/components/matter/test_api.py +++ b/tests/components/matter/test_api.py @@ -617,7 +617,6 @@ async def test_subscribe_network_topology( assert entry is not None topology = _mock_topology() - matter_client.get_network_topology = AsyncMock(return_value=topology) subscription_callback: Callable[[EventType, NetworkTopology], None] | None = None unsubscribe = MagicMock() @@ -635,6 +634,17 @@ async def test_subscribe_network_topology( matter_client.subscribe_events.side_effect = capture_subscription + during_fetch = _mock_topology() + during_fetch.collected_at = 1767888030000 + + async def fetch_topology() -> NetworkTopology: + # an update arriving while the initial fetch is in flight is buffered + assert subscription_callback is not None + subscription_callback(EventType.NETWORK_TOPOLOGY_UPDATED, during_fetch) + return topology + + matter_client.get_network_topology = AsyncMock(side_effect=fetch_topology) + ws_client = await hass_ws_client(hass) await ws_client.send_json({ID: 1, TYPE: "matter/subscribe_network_topology"}) msg = await ws_client.receive_json() @@ -648,6 +658,11 @@ async def test_subscribe_network_topology( assert msg["type"] == "event" assert msg["event"] == _expected_topology(topology, [entry.id, None, None]) + # the update buffered during the fetch is flushed right after + msg = await ws_client.receive_json() + assert msg["type"] == "event" + assert msg["event"] == _expected_topology(during_fetch, [entry.id, None, None]) + # a topology update from the server is forwarded to the subscription updated = _mock_topology() updated.collected_at = 1767888060000 @@ -664,3 +679,35 @@ async def test_subscribe_network_topology( assert msg["success"] unsubscribe.assert_called_once_with() + + +@pytest.mark.usefixtures("integration") +async def test_subscribe_network_topology_fetch_failure( + hass: HomeAssistant, + hass_ws_client: WebSocketGenerator, + matter_client: MagicMock, +) -> None: + """Test the event subscription is cleaned up when the initial fetch fails.""" + matter_client.server_info.schema_version = 13 + unsubscribe = MagicMock() + + def capture_subscription( + callback: Callable[[EventType, NetworkTopology], None], + event_filter: EventType | None = None, + node_filter: int | None = None, + attr_path_filter: str | None = None, + ) -> MagicMock: + return unsubscribe + + matter_client.subscribe_events.side_effect = capture_subscription + matter_client.get_network_topology = AsyncMock( + side_effect=ServerVersionTooOld("Command not available") + ) + + ws_client = await hass_ws_client(hass) + await ws_client.send_json({ID: 1, TYPE: "matter/subscribe_network_topology"}) + msg = await ws_client.receive_json() + + assert not msg["success"] + assert msg["error"]["code"] == "not_supported" + unsubscribe.assert_called_once_with()