mirror of
https://github.com/suitenumerique/meet.git
synced 2026-07-28 21:01:55 +00:00
lebaudantoine
0d5136206f
Add a new setting that controls whether an authenticated user can rename or modify their display name, or if they must use the one returned by the SSO. When the setting is disabled, only anonymous users can set a display name freely; authenticated users always use their SSO display name. Requested by many self-hosters.
265 lines
8.9 KiB
Python
265 lines
8.9 KiB
Python
"""
|
|
Test utils functions
|
|
"""
|
|
|
|
# pylint: disable=W0621
|
|
import json
|
|
from unittest import mock
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AnonymousUser
|
|
|
|
import jwt
|
|
import pytest
|
|
from livekit.api import TwirpError
|
|
|
|
from core.factories import UserFactory
|
|
from core.utils import (
|
|
NotificationError,
|
|
create_livekit_client,
|
|
generate_token,
|
|
notify_participants,
|
|
)
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def decode_token(token: str) -> dict:
|
|
"""Decode a LiveKit JWT access token for inspection."""
|
|
return jwt.decode(
|
|
token,
|
|
settings.LIVEKIT_CONFIGURATION["api_secret"],
|
|
algorithms=["HS256"],
|
|
)
|
|
|
|
|
|
def test_generate_token_authenticated_uses_full_name():
|
|
"""The token's display name should default to the user's full name."""
|
|
user = UserFactory(full_name="Jane Doe")
|
|
|
|
token = generate_token(room="my-room", user=user)
|
|
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Jane Doe"
|
|
assert claims["sub"] == str(user.sub)
|
|
|
|
|
|
def test_generate_token_authenticated_fallback_user_representation():
|
|
"""
|
|
When the user has no full name, the token's display name should fall back
|
|
to the user's string representation.
|
|
"""
|
|
user = UserFactory(full_name=None)
|
|
|
|
token = generate_token(room="my-room", user=user)
|
|
|
|
claims = decode_token(token)
|
|
assert claims["name"] == str(user)
|
|
|
|
|
|
def test_generate_token_explicit_username_overrides_default():
|
|
"""An explicitly provided username should take precedence over the full name."""
|
|
user = UserFactory(full_name="Jane Doe")
|
|
|
|
token = generate_token(room="my-room", user=user, username="Custom Name")
|
|
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Custom Name"
|
|
|
|
|
|
def test_authenticated_username_ignored_when_editing_disabled(settings):
|
|
"""With editing disabled, an authenticated user's username is ignored."""
|
|
settings.AUTHENTICATED_PARTICIPANTS_CAN_EDIT_DISPLAY_NAME = False
|
|
user = UserFactory(full_name="Jane Doe")
|
|
token = generate_token(room="my-room", user=user, username="Custom Name")
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Jane Doe"
|
|
|
|
|
|
def test_authenticated_default_name_unaffected_when_editing_disabled(settings):
|
|
"""Disabling editing doesn't disturb the default full-name path."""
|
|
settings.AUTHENTICATED_PARTICIPANTS_CAN_EDIT_DISPLAY_NAME = False
|
|
user = UserFactory(full_name="Jane Doe")
|
|
token = generate_token(room="my-room", user=user)
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Jane Doe"
|
|
|
|
|
|
def test_anonymous_uses_username_when_provided():
|
|
"""An anonymous user's provided username is used as the display name."""
|
|
token = generate_token(room="my-room", user=AnonymousUser(), username="Guest42")
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Guest42"
|
|
|
|
|
|
def test_anonymous_username_used_even_when_editing_disabled(settings):
|
|
"""The setting governs authenticated users only; anonymous can still set a name."""
|
|
settings.AUTHENTICATED_PARTICIPANTS_CAN_EDIT_DISPLAY_NAME = False
|
|
token = generate_token(room="my-room", user=AnonymousUser(), username="Guest42")
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Guest42"
|
|
|
|
|
|
def test_anonymous_falls_back_to_anonymous_label():
|
|
"""With no username, an anonymous user is labelled 'Anonymous'."""
|
|
token = generate_token(room="my-room", user=AnonymousUser())
|
|
claims = decode_token(token)
|
|
assert claims["name"] == "Anonymous"
|
|
|
|
|
|
@mock.patch("asyncio.get_running_loop")
|
|
@mock.patch("core.utils.LiveKitAPI")
|
|
def test_create_livekit_client_ssl_enabled(
|
|
mock_livekit_api, mock_get_running_loop, settings
|
|
):
|
|
"""Test LiveKitAPI client creation with SSL verification enabled."""
|
|
mock_get_running_loop.return_value = mock.MagicMock()
|
|
settings.LIVEKIT_VERIFY_SSL = True
|
|
|
|
create_livekit_client()
|
|
|
|
mock_livekit_api.assert_called_once_with(
|
|
**settings.LIVEKIT_CONFIGURATION, session=None
|
|
)
|
|
|
|
|
|
@mock.patch("core.utils.aiohttp.ClientSession")
|
|
@mock.patch("asyncio.get_running_loop")
|
|
@mock.patch("core.utils.LiveKitAPI")
|
|
def test_create_livekit_client_ssl_disabled(
|
|
mock_livekit_api, mock_get_running_loop, mock_client_session, settings
|
|
):
|
|
"""Test LiveKitAPI client creation with SSL verification disabled."""
|
|
mock_get_running_loop.return_value = mock.MagicMock()
|
|
mock_session_instance = mock.MagicMock()
|
|
mock_client_session.return_value = mock_session_instance
|
|
settings.LIVEKIT_VERIFY_SSL = False
|
|
|
|
create_livekit_client()
|
|
|
|
mock_livekit_api.assert_called_once_with(
|
|
**settings.LIVEKIT_CONFIGURATION, session=mock_session_instance
|
|
)
|
|
|
|
|
|
@mock.patch("asyncio.get_running_loop")
|
|
@mock.patch("core.utils.LiveKitAPI")
|
|
def test_create_livekit_client_custom_configuration(
|
|
mock_livekit_api, mock_get_running_loop, settings
|
|
):
|
|
"""Test LiveKitAPI client creation with custom configuration."""
|
|
settings.LIVEKIT_VERIFY_SSL = True
|
|
|
|
mock_get_running_loop.return_value = mock.MagicMock()
|
|
custom_configuration = {
|
|
"api_key": "mock_key",
|
|
"api_secret": "mock_secret",
|
|
"url": "http://mock-url.com",
|
|
}
|
|
|
|
create_livekit_client(custom_configuration)
|
|
|
|
mock_livekit_api.assert_called_once_with(**custom_configuration, session=None)
|
|
|
|
|
|
@mock.patch("core.utils.create_livekit_client")
|
|
def test_notify_participants_error(mock_create_livekit_client):
|
|
"""Test participant notification with API error."""
|
|
|
|
# Set up the mock LiveKitAPI and its behavior
|
|
mock_api_instance = mock.Mock()
|
|
mock_api_instance.room = mock.Mock()
|
|
mock_api_instance.room.send_data = mock.AsyncMock(
|
|
side_effect=TwirpError(msg="test error", code=123, status=123)
|
|
)
|
|
|
|
class MockResponse:
|
|
"""LiveKit API response mock with non-empty rooms list."""
|
|
|
|
rooms = ["room-1"]
|
|
|
|
mock_api_instance.room.list_rooms = mock.AsyncMock(return_value=MockResponse())
|
|
|
|
mock_api_instance.aclose = mock.AsyncMock()
|
|
mock_create_livekit_client.return_value = mock_api_instance
|
|
|
|
# Call the function and expect an exception
|
|
with pytest.raises(NotificationError, match="Failed to notify room participants"):
|
|
notify_participants(room_name="room-number-1", notification_data={"foo": "foo"})
|
|
|
|
# Verify that the service checked for existing rooms
|
|
mock_api_instance.room.list_rooms.assert_called_once()
|
|
|
|
# Verify send_data was called
|
|
mock_api_instance.room.send_data.assert_called_once()
|
|
|
|
# Verify aclose was still called after the exception
|
|
mock_api_instance.aclose.assert_called_once()
|
|
|
|
|
|
@mock.patch("core.utils.create_livekit_client")
|
|
def test_notify_participants_success_no_room(mock_create_livekit_client):
|
|
"""Test the notify_participants function when the LiveKit room doesn't exist."""
|
|
|
|
# Set up the mock LiveKitAPI and its behavior
|
|
mock_api_instance = mock.Mock()
|
|
mock_api_instance.room = mock.Mock()
|
|
mock_api_instance.room.send_data = mock.AsyncMock()
|
|
|
|
# Create a proper response object with an empty rooms list
|
|
class MockResponse:
|
|
"""LiveKit API response mock with empty rooms list."""
|
|
|
|
rooms = []
|
|
|
|
mock_api_instance.room.list_rooms = mock.AsyncMock(return_value=MockResponse())
|
|
mock_api_instance.aclose = mock.AsyncMock()
|
|
mock_create_livekit_client.return_value = mock_api_instance
|
|
|
|
notify_participants(room_name="room-number-1", notification_data={"foo": "foo"})
|
|
|
|
# Verify that the service checked for existing rooms
|
|
mock_api_instance.room.list_rooms.assert_called_once()
|
|
|
|
# Verify the send_data method was not called since no room exists
|
|
mock_api_instance.room.send_data.assert_not_called()
|
|
|
|
# Verify the connection was properly closed
|
|
mock_api_instance.aclose.assert_called_once()
|
|
|
|
|
|
@mock.patch("core.utils.create_livekit_client")
|
|
def test_notify_participants_success(mock_create_livekit_client):
|
|
"""Test successful participant notification."""
|
|
|
|
# Set up the mock LiveKitAPI and its behavior
|
|
mock_api_instance = mock.Mock()
|
|
mock_api_instance.room = mock.Mock()
|
|
mock_api_instance.room.send_data = mock.AsyncMock()
|
|
|
|
class MockResponse:
|
|
"""LiveKit API response mock with non-empty rooms list."""
|
|
|
|
rooms = ["room-1"]
|
|
|
|
mock_api_instance.room.list_rooms = mock.AsyncMock(return_value=MockResponse())
|
|
|
|
mock_api_instance.aclose = mock.AsyncMock()
|
|
mock_create_livekit_client.return_value = mock_api_instance
|
|
|
|
# Call the function
|
|
notify_participants(room_name="room-number-1", notification_data={"foo": "foo"})
|
|
|
|
# Verify that the service checked for existing rooms
|
|
mock_api_instance.room.list_rooms.assert_called_once()
|
|
|
|
# Verify the send_data method was called
|
|
mock_api_instance.room.send_data.assert_called_once()
|
|
send_data_request = mock_api_instance.room.send_data.call_args[0][0]
|
|
assert send_data_request.room == "room-number-1"
|
|
assert json.loads(send_data_request.data.decode("utf-8")) == {"foo": "foo"}
|
|
assert send_data_request.kind == 0 # RELIABLE mode in Livekit protocol
|
|
|
|
# Verify aclose was called
|
|
mock_api_instance.aclose.assert_called_once()
|