Add camera websocket endpoint for ICE servers

homeassistant/components/camera/webrtc.py

@@ -345,13 +345,34 @@ async def ws_candidate(
     connection.send_message(websocket_api.result_message(msg["id"]))
 
 
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "camera/webrtc/ice_servers",
+    }
+)
+@websocket_api.async_response
+async def ws_ice_servers(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Handle get WebRTC ICE servers websocket command."""
+    ice_servers = [
+        server.to_dict()
+        for servers in hass.data.get(DATA_ICE_SERVERS, [])
+        for server in servers()
+    ]
+    connection.send_result(msg["id"], ice_servers)
+
+
 @callback
 def async_register_ws(hass: HomeAssistant) -> None:
     """Register camera webrtc ws endpoints."""
 
     websocket_api.async_register_command(hass, ws_webrtc_offer)
     websocket_api.async_register_command(hass, ws_get_client_config)
     websocket_api.async_register_command(hass, ws_candidate)
+    websocket_api.async_register_command(hass, ws_ice_servers)
 
 
 async def async_get_supported_provider(

tests/components/camera/test_webrtc.py

@@ -757,6 +757,62 @@ async def test_ws_webrtc_candidate_invalid_stream_type(
     }
 
 
+@pytest.mark.usefixtures("mock_test_webrtc_cameras")
+async def test_ws_ice_servers(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """Test get WebRTC ICE servers."""
+    await async_setup_component(hass, "camera", {})
+
+    client = await hass_ws_client(hass)
+    await client.send_json_auto_id({"type": "camera/webrtc/ice_servers"})
+    msg = await client.receive_json()
+
+    # Assert WebSocket response
+    assert msg["type"] == TYPE_RESULT
+    assert msg["success"]
+    assert msg["result"] == [
+        {
+            "urls": [
+                "stun:stun.home-assistant.io:80",
+                "stun:stun.home-assistant.io:3478",
+            ]
+        },
+    ]
+
+    @callback
+    def get_ice_server() -> list[RTCIceServer]:
+        return [
+            RTCIceServer(
+                urls=["stun:example2.com", "turn:example2.com"],
+                username="user",
+                credential="pass",
+            )
+        ]
+
+    async_register_ice_servers(hass, get_ice_server)
+
+    await client.send_json_auto_id({"type": "camera/webrtc/ice_servers"})
+    msg = await client.receive_json()
+
+    # Assert WebSocket response
+    assert msg["type"] == TYPE_RESULT
+    assert msg["success"]
+    assert msg["result"] == [
+        {
+            "urls": [
+                "stun:stun.home-assistant.io:80",
+                "stun:stun.home-assistant.io:3478",
+            ]
+        },
+        {
+            "urls": ["stun:example2.com", "turn:example2.com"],
+            "username": "user",
+            "credential": "pass",
+        },
+    ]
+
+
 async def test_webrtc_provider_optional_interface(hass: HomeAssistant) -> None:
     """Test optional interface for WebRTC provider."""