Require admin for Matter topology commands and fix subscription race

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

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()