✨(backend) add connection-test API

Currently users have no way to reliably test their connection before
joining a room. We then want to create a connection test page
to adress this issue.

The testing will require a dedicated LiveKit token without going
through the room API, which is tied to registered meetings, lobby rules,
and longer-lived access tokens.

We introduce a new API endpoint GET /api/v1.0/connection-test/
to issue a dedicated token for diagnostics, even for anonymous users.
Each request creates a new room so users never share
the same LiveKit room during tests. Tokens are short-lived
(default 10 minutes) to limit reuse,
and the endpoint is throttled to prevent abuse.

src/backend/core/api/connection_test.py

@@ -0,0 +1,42 @@
+"""Connection test API endpoint."""
+
+from datetime import timedelta
+from uuid import uuid4
+
+from django.conf import settings
+
+from rest_framework.decorators import api_view, throttle_classes
+from rest_framework.response import Response
+
+from core.api.throttling import (
+    ConnectionTestAnonRateThrottle,
+    ConnectionTestUserRateThrottle,
+)
+from core.utils import generate_token
+
+CONNECTION_TEST_USERNAME = "Test connexion"
+
+
+@api_view(["GET"])
+@throttle_classes([ConnectionTestUserRateThrottle, ConnectionTestAnonRateThrottle])
+def get_connection_test_config(request):
+    """Return a short-lived LiveKit token for an ephemeral connection test room."""
+    room = f"{settings.CONNECTION_TEST_ROOM_PREFIX}{uuid4()}"
+    expires_in = settings.CONNECTION_TEST_TOKEN_TTL_SECONDS
+
+    return Response(
+        {
+            "livekit": {
+                "url": settings.LIVEKIT_CONFIGURATION["url"],
+                "room": room,
+                "token": generate_token(
+                    room=room,
+                    user=request.user,
+                    username=CONNECTION_TEST_USERNAME,
+                    is_admin_or_owner=False,
+                    ttl=timedelta(seconds=expires_in),
+                ),
+                "expires_in": expires_in,
+            },
+        }
+    )

src/backend/core/api/throttling.py

@@ -73,3 +73,15 @@ class CreationCallbackAnonRateThrottle(MonitoredAnonRateThrottle):
     """Throttle Anonymous user requesting room generation callback"""
 
     scope = "creation_callback"
+
+
+class ConnectionTestUserRateThrottle(MonitoredUserRateThrottle):
+    """Throttle authenticated users requesting connection test tokens."""
+
+    scope = "connection_test"
+
+
+class ConnectionTestAnonRateThrottle(MonitoredAnonRateThrottle):
+    """Throttle anonymous users requesting connection test tokens."""
+
+    scope = "connection_test"

src/backend/core/services/livekit_events.py

@@ -218,9 +218,21 @@ class LiveKitEventsService:
 
         # Silently ignoring EGRESS_ABORTED, EGRESS_FAILED
 
+    @staticmethod
+    def _is_connection_test_room(room_name: str) -> bool:
+        """Return True for ephemeral rooms created by the connection test endpoint."""
+        return room_name.startswith(settings.CONNECTION_TEST_ROOM_PREFIX)
+
     def _handle_room_started(self, data):
         """Handle 'room_started' event."""
 
+        if self._is_connection_test_room(data.room.name):
+            logger.info(
+                "Ignoring room_started event for connection test room '%s'.",
+                data.room.name,
+            )
+            return
+
         try:
             room_id = uuid.UUID(data.room.name)
         except ValueError as e:
@@ -246,6 +258,13 @@ class LiveKitEventsService:
     def _handle_room_finished(self, data):
         """Handle 'room_finished' event."""
 
+        if self._is_connection_test_room(data.room.name):
+            logger.info(
+                "Ignoring room_finished event for connection test room '%s'.",
+                data.room.name,
+            )
+            return
+
         try:
             room_id = uuid.UUID(data.room.name)
         except ValueError as e:

src/backend/core/tests/services/test_livekit_events.py

@@ -6,6 +6,8 @@ Test LiveKitEvents service.
 import uuid
 from unittest import mock
 
+from django.test.utils import override_settings
+
 import pytest
 from livekit.api import EgressStatus
 
@@ -550,6 +552,23 @@ def test_handle_room_finished_raises_error_when_telephony_deletion_fails(
     mock_clear_cache.assert_not_called()
 
 
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test-")
+@mock.patch.object(LobbyService, "clear_room_cache")
+@mock.patch.object(TelephonyService, "delete_dispatch_rule")
+def test_handle_room_finished_ignores_connection_test_room(
+    mock_delete_dispatch_rule, mock_clear_cache, service, settings
+):
+    """Should ignore room_finished events for connection test rooms."""
+    settings.ROOM_TELEPHONY_ENABLED = True
+    mock_data = mock.MagicMock()
+    mock_data.room.name = f"{settings.CONNECTION_TEST_ROOM_PREFIX}{uuid.uuid4()}"
+
+    service._handle_room_finished(mock_data)
+
+    mock_delete_dispatch_rule.assert_not_called()
+    mock_clear_cache.assert_not_called()
+
+
 def test_handle_room_finished_raises_error_for_invalid_room_name(service):
     """Should raise ActionFailedError when room name format is invalid when room finishes."""
     mock_data = mock.MagicMock()
@@ -600,6 +619,15 @@ def test_handle_room_started_raises_error_for_invalid_room_name(service):
         service._handle_room_started(mock_data)
 
 
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test-")
+def test_handle_room_started_ignores_connection_test_room(service, settings):
+    """Should ignore room_started events for connection test rooms."""
+    mock_data = mock.MagicMock()
+    mock_data.room.name = f"{settings.CONNECTION_TEST_ROOM_PREFIX}{uuid.uuid4()}"
+
+    service._handle_room_started(mock_data)
+
+
 def test_handle_room_started_raises_error_for_nonexistent_room(service):
     """Should raise ActionFailedError when a room starts that doesn't exist in the database."""
     mock_data = mock.MagicMock()

src/backend/core/tests/test_api_connection_test.py

@@ -0,0 +1,66 @@
+"""Test connection test API endpoint."""
+
+import uuid
+
+from django.test.utils import override_settings
+
+import jwt
+import pytest
+from rest_framework.test import APIClient
+
+from core.api.connection_test import CONNECTION_TEST_USERNAME
+
+pytestmark = pytest.mark.django_db
+
+
+@override_settings(
+    CONNECTION_TEST_TOKEN_TTL_SECONDS=600,
+    CONNECTION_TEST_ROOM_PREFIX="connection-test-",
+)
+def test_api_connection_test_returns_ephemeral_livekit_config():
+    """Each request gets a dedicated room and a short-lived token."""
+    client = APIClient()
+    response_a = client.get("/api/v1.0/connection-test/")
+    response_b = client.get("/api/v1.0/connection-test/")
+
+    assert response_a.status_code == 200
+    assert response_b.status_code == 200
+
+    data_a = response_a.json()
+    data_b = response_b.json()
+
+    room_a = data_a["livekit"]["room"]
+    room_b = data_b["livekit"]["room"]
+
+    assert room_a.startswith("connection-test-")
+    assert room_b.startswith("connection-test-")
+    uuid.UUID(room_a.removeprefix("connection-test-"))
+    uuid.UUID(room_b.removeprefix("connection-test-"))
+    assert room_a != room_b
+    assert data_a["livekit"]["url"]
+    assert data_a["livekit"]["token"]
+    assert data_a["livekit"]["expires_in"] == 600
+    assert data_a["livekit"]["token"] != data_b["livekit"]["token"]
+
+
+@override_settings(CONNECTION_TEST_TOKEN_TTL_SECONDS=300)
+def test_api_connection_test_token_is_short_lived_for_user(settings):
+    """Connection test tokens expire quickly for users."""
+    client = APIClient()
+    response = client.get("/api/v1.0/connection-test/")
+
+    assert response.status_code == 200
+
+    config = response.json()["livekit"]
+    payload = jwt.decode(
+        config["token"],
+        settings.LIVEKIT_CONFIGURATION["api_secret"],
+        algorithms=["HS256"],
+        options={"verify_exp": False},
+    )
+
+    assert config["expires_in"] == 300
+    assert payload["video"]["room"] == config["room"]
+    assert payload["name"] == CONNECTION_TEST_USERNAME
+    assert payload["video"]["roomAdmin"] is False
+    assert payload["exp"] - payload["nbf"] == 300

src/backend/core/urls.py

@@ -8,6 +8,7 @@ from rest_framework.routers import DefaultRouter, SimpleRouter
 
 from core.addons import viewsets as addons_viewsets
 from core.api import get_frontend_configuration, viewsets
+from core.api.connection_test import get_connection_test_config
 from core.external_api import viewsets as external_viewsets
 
 # - Main endpoints
@@ -46,6 +47,11 @@ urlpatterns = [
                 *router.urls,
                 *oidc_urls,
                 path("config/", get_frontend_configuration, name="config"),
+                path(
+                    "connection-test/",
+                    get_connection_test_config,
+                    name="connection_test",
+                ),
             ]
         ),
     ),

src/backend/core/utils.py

@@ -12,6 +12,7 @@ import mimetypes
 import random
 import secrets
 import string
+from datetime import timedelta
 from functools import lru_cache
 from typing import List, Optional
 from uuid import uuid4
@@ -68,6 +69,7 @@ def generate_token(
     sources: Optional[List[str]] = None,
     is_admin_or_owner: bool = False,
     participant_id: Optional[str] = None,
+    ttl: Optional[timedelta] = None,
 ) -> str:
     """Generate a LiveKit access token for a user in a specific room.
 
@@ -83,6 +85,7 @@ def generate_token(
         is_admin_or_owner (bool): Whether user has admin privileges
         participant_id (Optional[str]): Stable identifier for anonymous users;
                          used as identity when user.is_anonymous.
+        ttl (Optional[timedelta]): Token validity duration. Defaults to LiveKit SDK default.
 
     Returns:
         str: The LiveKit JWT access token.
@@ -131,6 +134,8 @@ def generate_token(
             {"color": color, "room_admin": "true" if is_admin_or_owner else "false"}
         )
     )
+    if ttl is not None:
+        token = token.with_ttl(ttl)
 
     return token.to_jwt()
 

src/backend/meet/settings.py

@@ -349,6 +349,11 @@ class Base(Configuration):
                 environ_name="CREATION_CALLBACK_THROTTLE_RATES",
                 environ_prefix=None,
             ),
+            "connection_test": values.Value(
+                default="30/minute",
+                environ_name="CONNECTION_TEST_THROTTLE_RATES",
+                environ_prefix=None,
+            ),
         },
     }
     MONITORED_THROTTLE_FAILURE_CALLBACK = (
@@ -655,6 +660,16 @@ class Base(Configuration):
         environ_prefix=None,
         default=False,
     )
+    CONNECTION_TEST_TOKEN_TTL_SECONDS = values.PositiveIntegerValue(
+        600,
+        environ_name="CONNECTION_TEST_TOKEN_TTL_SECONDS",
+        environ_prefix=None,
+    )
+    CONNECTION_TEST_ROOM_PREFIX = values.Value(
+        "connection-test-",
+        environ_name="CONNECTION_TEST_ROOM_PREFIX",
+        environ_prefix=None,
+    )
     LIVEKIT_VERIFY_SSL = values.BooleanValue(
         True, environ_name="LIVEKIT_VERIFY_SSL", environ_prefix=None
     )