Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a324c3680b |
@@ -37,10 +37,6 @@ def get_frontend_configuration(request):
|
||||
"""Returns the frontend configuration dict as configured in settings."""
|
||||
frontend_configuration = {
|
||||
"LANGUAGE_CODE": settings.LANGUAGE_CODE,
|
||||
"recording": {
|
||||
"is_enabled": settings.RECORDING_ENABLE,
|
||||
"available_modes": settings.RECORDING_WORKER_CLASSES.keys(),
|
||||
},
|
||||
}
|
||||
frontend_configuration.update(settings.FRONTEND_CONFIGURATION)
|
||||
return Response(frontend_configuration)
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
"""Permission handlers for the Meet core app."""
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
from rest_framework import permissions
|
||||
|
||||
from ..models import RoleChoices
|
||||
@@ -89,33 +87,3 @@ class HasAbilityPermission(IsAuthenticated):
|
||||
def has_object_permission(self, request, view, obj):
|
||||
"""Check permission for a given object."""
|
||||
return obj.get_abilities(request.user).get(view.action, False)
|
||||
|
||||
|
||||
class HasPrivilegesOnRoom(IsAuthenticated):
|
||||
"""Check if user has privileges on a given room."""
|
||||
|
||||
message = "You must have privileges to start a recording."
|
||||
|
||||
def has_object_permission(self, request, view, obj):
|
||||
"""Determine if user has privileges on room."""
|
||||
return obj.is_owner(request.user) or obj.is_administrator(request.user)
|
||||
|
||||
|
||||
class IsRecordingEnabled(permissions.BasePermission):
|
||||
"""Check if the recording feature is enabled."""
|
||||
|
||||
message = "Access denied, recording is disabled."
|
||||
|
||||
def has_permission(self, request, view):
|
||||
"""Determine if access is allowed based on settings."""
|
||||
return settings.RECORDING_ENABLE
|
||||
|
||||
|
||||
class IsStorageEventEnabled(permissions.BasePermission):
|
||||
"""Check if the storage event feature is enabled."""
|
||||
|
||||
message = "Access denied, storage event is disabled."
|
||||
|
||||
def has_permission(self, request, view):
|
||||
"""Determine if access is allowed based on settings."""
|
||||
return settings.RECORDING_STORAGE_EVENT_ENABLE
|
||||
|
||||
@@ -130,7 +130,8 @@ class RoomSerializer(serializers.ModelSerializer):
|
||||
del output["configuration"]
|
||||
|
||||
if role is not None or instance.is_public:
|
||||
slug = f"{instance.id!s}"
|
||||
slug = f"{instance.id!s}".replace("-", "")
|
||||
|
||||
username = request.query_params.get("username", None)
|
||||
|
||||
output["livekit"] = {
|
||||
@@ -155,25 +156,3 @@ class RecordingSerializer(serializers.ModelSerializer):
|
||||
model = models.Recording
|
||||
fields = ["id", "room", "created_at", "updated_at", "status"]
|
||||
read_only_fields = fields
|
||||
|
||||
|
||||
class StartRecordingSerializer(serializers.Serializer):
|
||||
"""Validate start recording requests."""
|
||||
|
||||
mode = serializers.ChoiceField(
|
||||
choices=models.RecordingModeChoices.choices,
|
||||
required=True,
|
||||
error_messages={
|
||||
"required": "Recording mode is required.",
|
||||
"invalid_choice": "Invalid recording mode. Choose between "
|
||||
"screen_recording or transcript.",
|
||||
},
|
||||
)
|
||||
|
||||
def create(self, validated_data):
|
||||
"""Not implemented as this is a validation-only serializer."""
|
||||
raise NotImplementedError("StartRecordingSerializer is validation-only")
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
"""Not implemented as this is a validation-only serializer."""
|
||||
raise NotImplementedError("StartRecordingSerializer is validation-only")
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
"""API endpoints"""
|
||||
|
||||
import uuid
|
||||
from logging import getLogger
|
||||
|
||||
from django.conf import settings
|
||||
from django.db.models import Q
|
||||
@@ -15,41 +14,16 @@ from rest_framework import (
|
||||
pagination,
|
||||
viewsets,
|
||||
)
|
||||
from rest_framework import (
|
||||
exceptions as drf_exceptions,
|
||||
)
|
||||
from rest_framework import (
|
||||
response as drf_response,
|
||||
)
|
||||
from rest_framework import (
|
||||
status as drf_status,
|
||||
)
|
||||
|
||||
from core import models, utils
|
||||
from core.recording.event.authentication import StorageEventAuthentication
|
||||
from core.recording.event.exceptions import (
|
||||
InvalidBucketError,
|
||||
InvalidFileTypeError,
|
||||
ParsingEventDataError,
|
||||
)
|
||||
from core.recording.event.parsers import get_parser
|
||||
from core.recording.worker.exceptions import (
|
||||
RecordingStartError,
|
||||
RecordingStopError,
|
||||
)
|
||||
from core.recording.worker.factories import (
|
||||
get_worker_service,
|
||||
)
|
||||
from core.recording.worker.mediator import (
|
||||
WorkerServiceMediator,
|
||||
)
|
||||
|
||||
from . import permissions, serializers
|
||||
|
||||
# pylint: disable=too-many-ancestors
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
|
||||
class NestedGenericViewSet(viewsets.GenericViewSet):
|
||||
"""
|
||||
@@ -259,89 +233,6 @@ class RoomViewSet(
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
@decorators.action(
|
||||
detail=True,
|
||||
methods=["post"],
|
||||
url_path="start-recording",
|
||||
permission_classes=[
|
||||
permissions.HasPrivilegesOnRoom,
|
||||
permissions.IsRecordingEnabled,
|
||||
],
|
||||
)
|
||||
def start_room_recording(self, request, pk=None): # pylint: disable=unused-argument
|
||||
"""Start recording a room."""
|
||||
|
||||
serializer = serializers.StartRecordingSerializer(data=request.data)
|
||||
|
||||
if not serializer.is_valid():
|
||||
return drf_response.Response(
|
||||
{"detail": "Invalid request."}, status=drf_status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
|
||||
mode = serializer.validated_data["mode"]
|
||||
room = self.get_object()
|
||||
|
||||
# May raise exception if an active or initiated recording already exist for the room
|
||||
recording = models.Recording.objects.create(room=room, mode=mode)
|
||||
|
||||
models.RecordingAccess.objects.create(
|
||||
user=self.request.user, role=models.RoleChoices.OWNER, recording=recording
|
||||
)
|
||||
|
||||
worker_service = get_worker_service(mode=recording.mode)
|
||||
worker_manager = WorkerServiceMediator(worker_service=worker_service)
|
||||
|
||||
try:
|
||||
worker_manager.start(recording)
|
||||
except RecordingStartError:
|
||||
return drf_response.Response(
|
||||
{"error": f"Recording failed to start for room {room.slug}"},
|
||||
status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
)
|
||||
|
||||
return drf_response.Response(
|
||||
{"message": f"Recording successfully started for room {room.slug}"},
|
||||
status=drf_status.HTTP_201_CREATED,
|
||||
)
|
||||
|
||||
@decorators.action(
|
||||
detail=True,
|
||||
methods=["post"],
|
||||
url_path="stop-recording",
|
||||
permission_classes=[
|
||||
permissions.HasPrivilegesOnRoom,
|
||||
permissions.IsRecordingEnabled,
|
||||
],
|
||||
)
|
||||
def stop_room_recording(self, request, pk=None): # pylint: disable=unused-argument
|
||||
"""Stop room recording."""
|
||||
|
||||
room = self.get_object()
|
||||
|
||||
try:
|
||||
recording = models.Recording.objects.get(
|
||||
room=room, status=models.RecordingStatusChoices.ACTIVE
|
||||
)
|
||||
except models.Recording.DoesNotExist as e:
|
||||
raise drf_exceptions.NotFound(
|
||||
"No active recording found for this room."
|
||||
) from e
|
||||
|
||||
worker_service = get_worker_service(mode=recording.mode)
|
||||
worker_manager = WorkerServiceMediator(worker_service=worker_service)
|
||||
|
||||
try:
|
||||
worker_manager.stop(recording)
|
||||
except RecordingStopError:
|
||||
return drf_response.Response(
|
||||
{"error": f"Recording failed to stop for room {room.slug}"},
|
||||
status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
)
|
||||
|
||||
return drf_response.Response(
|
||||
{"message": f"Recording stopped for room {room.slug}."}
|
||||
)
|
||||
|
||||
|
||||
class ResourceAccessListModelMixin:
|
||||
"""List mixin for resource access API."""
|
||||
@@ -410,47 +301,3 @@ class RecordingViewSet(
|
||||
.get_queryset()
|
||||
.filter(Q(accesses__user=user) | Q(accesses__team__in=user.get_teams()))
|
||||
)
|
||||
|
||||
@decorators.action(
|
||||
detail=False,
|
||||
methods=["post"],
|
||||
url_path="storage-hook",
|
||||
authentication_classes=[StorageEventAuthentication],
|
||||
permission_classes=[permissions.IsStorageEventEnabled],
|
||||
)
|
||||
def on_storage_event_received(self, request, pk=None): # pylint: disable=unused-argument
|
||||
"""Handle incoming storage hook events for recordings."""
|
||||
|
||||
parser = get_parser()
|
||||
|
||||
try:
|
||||
recording_id = parser.get_recording_id(request.data)
|
||||
|
||||
except ParsingEventDataError as e:
|
||||
raise drf_exceptions.PermissionDenied(f"Invalid request data: {e}") from e
|
||||
|
||||
except InvalidBucketError as e:
|
||||
raise drf_exceptions.PermissionDenied("Invalid bucket specified") from e
|
||||
|
||||
except InvalidFileTypeError as e:
|
||||
return drf_response.Response(
|
||||
{"message": f"Ignore this file type, {e}"},
|
||||
)
|
||||
|
||||
try:
|
||||
recording = models.Recording.objects.get(id=recording_id)
|
||||
except models.Recording.DoesNotExist as e:
|
||||
raise drf_exceptions.NotFound("No recording found for this event.") from e
|
||||
|
||||
if not recording.is_savable():
|
||||
raise drf_exceptions.PermissionDenied(
|
||||
f"Recording with ID {recording_id} cannot be saved because it is either,"
|
||||
" in an error state or has already been saved."
|
||||
)
|
||||
|
||||
recording.status = models.RecordingStatusChoices.SAVED
|
||||
recording.save()
|
||||
|
||||
return drf_response.Response(
|
||||
{"message": "Event processed."},
|
||||
)
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
"""Authentication Backends for the Meet core app."""
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import SuspiciousOperation
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
@@ -67,40 +66,35 @@ class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
|
||||
user_info = self.get_userinfo(access_token, id_token, payload)
|
||||
sub = user_info.get("sub")
|
||||
|
||||
if not sub:
|
||||
if sub is None:
|
||||
raise SuspiciousOperation(
|
||||
_("User info contained no recognizable user identification")
|
||||
)
|
||||
|
||||
email = user_info.get("email")
|
||||
user = self.get_existing_user(sub, email)
|
||||
|
||||
if not user and self.get_settings("OIDC_CREATE_USER", True):
|
||||
user = User.objects.create(
|
||||
sub=sub,
|
||||
email=email,
|
||||
password="!", # noqa: S106
|
||||
)
|
||||
elif not user:
|
||||
return None
|
||||
|
||||
if not user.is_active:
|
||||
raise SuspiciousOperation(_("User account is disabled"))
|
||||
try:
|
||||
user = User.objects.get(sub=sub)
|
||||
except User.DoesNotExist:
|
||||
if self.get_settings("OIDC_CREATE_USER", True):
|
||||
user = self.create_user(user_info)
|
||||
else:
|
||||
user = None
|
||||
|
||||
return user
|
||||
|
||||
def get_existing_user(self, sub, email):
|
||||
"""Fetch existing user by sub or email."""
|
||||
try:
|
||||
return User.objects.get(sub=sub)
|
||||
except User.DoesNotExist:
|
||||
if email and settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION:
|
||||
try:
|
||||
return User.objects.get(email__iexact=email)
|
||||
except User.DoesNotExist:
|
||||
pass
|
||||
except User.MultipleObjectsReturned as e:
|
||||
raise SuspiciousOperation(
|
||||
_("Multiple user accounts share a common email.")
|
||||
) from e
|
||||
return None
|
||||
def create_user(self, claims):
|
||||
"""Return a newly created User instance."""
|
||||
|
||||
sub = claims.get("sub")
|
||||
|
||||
if sub is None:
|
||||
raise SuspiciousOperation(
|
||||
_("Claims contained no recognizable user identification")
|
||||
)
|
||||
|
||||
user = User.objects.create(
|
||||
sub=sub,
|
||||
email=claims.get("email"),
|
||||
password="!", # noqa: S106
|
||||
)
|
||||
|
||||
return user
|
||||
|
||||
@@ -32,7 +32,6 @@ class ResourceFactory(factory.django.DjangoModelFactory):
|
||||
|
||||
class Meta:
|
||||
model = models.Resource
|
||||
skip_postgeneration_save = True
|
||||
|
||||
is_public = factory.Faker("boolean", chance_of_getting_true=50)
|
||||
|
||||
@@ -46,8 +45,6 @@ class ResourceFactory(factory.django.DjangoModelFactory):
|
||||
else:
|
||||
UserResourceAccessFactory(resource=self, user=item[0], role=item[1])
|
||||
|
||||
self.save()
|
||||
|
||||
|
||||
class UserResourceAccessFactory(factory.django.DjangoModelFactory):
|
||||
"""Create fake resource user accesses for testing."""
|
||||
@@ -75,11 +72,9 @@ class RecordingFactory(factory.django.DjangoModelFactory):
|
||||
|
||||
class Meta:
|
||||
model = models.Recording
|
||||
skip_postgeneration_save = True
|
||||
|
||||
room = factory.SubFactory(RoomFactory)
|
||||
status = models.RecordingStatusChoices.INITIATED
|
||||
mode = models.RecordingModeChoices.SCREEN_RECORDING
|
||||
worker_id = None
|
||||
|
||||
@factory.post_generation
|
||||
@@ -94,8 +89,6 @@ class RecordingFactory(factory.django.DjangoModelFactory):
|
||||
recording=self, user=item[0], role=item[1]
|
||||
)
|
||||
|
||||
self.save()
|
||||
|
||||
|
||||
class UserRecordingAccessFactory(factory.django.DjangoModelFactory):
|
||||
"""Create fake recording user accesses for testing."""
|
||||
|
||||
@@ -1,89 +0,0 @@
|
||||
|
||||
from django.db import migrations
|
||||
from django.db.models import Count
|
||||
from core.models import RoleChoices
|
||||
|
||||
def merge_duplicate_user_accounts(apps, schema_editor):
|
||||
"""Merge user accounts that share the same email address.
|
||||
|
||||
Historical Context:
|
||||
Previously, ProConnect authentication could return users with the same email
|
||||
but different sub, leading to duplicate user accounts. While the application
|
||||
now prevents this scenario, this migration is needed to clean up existing
|
||||
duplicate accounts to ensure users can continue to connect without being blocked
|
||||
by unique email constraints.
|
||||
|
||||
Performance of this migration is poor, this implementation prioritizes readability
|
||||
and maintainability. Consider refactoring this code to avoid individual db queries
|
||||
on each iteration.
|
||||
"""
|
||||
|
||||
User = apps.get_model('core', 'User')
|
||||
ResourceAccess = apps.get_model('core', 'ResourceAccess')
|
||||
|
||||
emails_with_duplicates = (
|
||||
User.objects.values('email')
|
||||
.annotate(count=Count('id'))
|
||||
.filter(count__gt=1)
|
||||
.values_list('email', flat=True)
|
||||
)
|
||||
|
||||
for email in emails_with_duplicates:
|
||||
# Keep the oldest user
|
||||
primary_user = User.objects.filter(email=email).order_by('created_at').first()
|
||||
duplicate_user_accounts = User.objects.filter(email=email).exclude(id=primary_user.id)
|
||||
|
||||
# Get IDs of duplicate accounts to be merged
|
||||
duplicate_account_ids = list(duplicate_user_accounts.values_list('id', flat=True))
|
||||
resource_accesses_to_transfer = ResourceAccess.objects.filter(user_id__in=duplicate_account_ids)
|
||||
|
||||
# Transfer resource access permissions to primary user
|
||||
# This process handles role hierarchy where:
|
||||
# OWNER > ADMIN > MEMBER
|
||||
for resource_access in resource_accesses_to_transfer:
|
||||
|
||||
# Determine if primary user already has access to this resource
|
||||
existing_primary_access = ResourceAccess.objects.filter(
|
||||
user_id=primary_user.id,
|
||||
resource_id=resource_access.resource.id
|
||||
).first()
|
||||
|
||||
if existing_primary_access:
|
||||
# Skip if primary user is already OWNER as it's the highest privilege level
|
||||
# No need to modify or downgrade owner access
|
||||
if existing_primary_access.role == RoleChoices.OWNER:
|
||||
continue
|
||||
|
||||
# Skip if primary user already has the exact same role
|
||||
# No need to update when roles match
|
||||
elif existing_primary_access.role == resource_access.role:
|
||||
continue
|
||||
|
||||
# Skip if new role is MEMBER since user already has base access
|
||||
# All existing access includes at least MEMBER privileges
|
||||
elif resource_access.role == RoleChoices.MEMBER:
|
||||
continue
|
||||
|
||||
# Update the role only if it represents a higher privilege level
|
||||
# Preserves existing access record while updating the role
|
||||
existing_primary_access.role = resource_access.role
|
||||
existing_primary_access.save()
|
||||
else:
|
||||
# Transfer access to primary user
|
||||
resource_access.user_id = primary_user.id
|
||||
resource_access.save()
|
||||
|
||||
# Delete duplicate accounts - CASCADE will automatically remove any untransferred
|
||||
# ResourceAccess records and other related data for these users
|
||||
duplicate_user_accounts.delete()
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('core', '0005_recording_recordingaccess_and_more'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RunPython(merge_duplicate_user_accounts, reverse_code=migrations.RunPython.noop),
|
||||
]
|
||||
@@ -1,18 +0,0 @@
|
||||
# Generated by Django 5.1.1 on 2024-11-12 10:59
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('core', '0006_merge_duplicate_users'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='recording',
|
||||
name='mode',
|
||||
field=models.CharField(choices=[('screen_recording', 'SCREEN_RECORDING'), ('transcript', 'TRANSCRIPT')], default='screen_recording', help_text='Defines the mode of recording being called.', max_length=20, verbose_name='Recording mode'),
|
||||
),
|
||||
]
|
||||
@@ -72,13 +72,6 @@ class RecordingStatusChoices(models.TextChoices):
|
||||
return status in {cls.ABORTED, cls.FAILED_TO_START, cls.FAILED_TO_STOP}
|
||||
|
||||
|
||||
class RecordingModeChoices(models.TextChoices):
|
||||
"""Recording mode choices."""
|
||||
|
||||
SCREEN_RECORDING = "screen_recording", _("SCREEN_RECORDING")
|
||||
TRANSCRIPT = "transcript", _("TRANSCRIPT")
|
||||
|
||||
|
||||
class BaseModel(models.Model):
|
||||
"""
|
||||
Serves as an abstract base model for other models, ensuring that records are validated
|
||||
@@ -489,13 +482,6 @@ class Recording(BaseModel):
|
||||
"This ID is retained even when the worker stops, allowing for easy tracking."
|
||||
),
|
||||
)
|
||||
mode = models.CharField(
|
||||
max_length=20,
|
||||
choices=RecordingModeChoices.choices,
|
||||
default=RecordingModeChoices.SCREEN_RECORDING,
|
||||
verbose_name=_("Recording mode"),
|
||||
help_text=_("Defines the mode of recording being called."),
|
||||
)
|
||||
|
||||
class Meta:
|
||||
db_table = "meet_recording"
|
||||
@@ -540,10 +526,10 @@ class Recording(BaseModel):
|
||||
def is_savable(self) -> bool:
|
||||
"""Determine if the recording can be saved based on its current status."""
|
||||
|
||||
return self.status in {
|
||||
RecordingStatusChoices.ACTIVE,
|
||||
RecordingStatusChoices.STOPPED,
|
||||
}
|
||||
is_unsuccessful = RecordingStatusChoices.is_unsuccessful(self.status)
|
||||
is_already_saved = self.status == RecordingStatusChoices.SAVED
|
||||
|
||||
return not is_unsuccessful and not is_already_saved
|
||||
|
||||
|
||||
class RecordingAccess(BaseAccess):
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
"""Meet event parser classes, authentication and exceptions."""
|
||||
@@ -1,96 +0,0 @@
|
||||
"""Authentication class for storage event token validation."""
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
|
||||
from django.conf import settings
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from rest_framework.authentication import BaseAuthentication
|
||||
from rest_framework.exceptions import AuthenticationFailed
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MachineUser:
|
||||
"""Represent a non-interactive system user for automated storage operations."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.pk = None
|
||||
self.username = "storage_event_user"
|
||||
self.is_active = True
|
||||
|
||||
@property
|
||||
def is_authenticated(self):
|
||||
"""Indicate if this machine user is authenticated."""
|
||||
return True
|
||||
|
||||
@property
|
||||
def is_anonymous(self) -> bool:
|
||||
"""Indicate if this is an anonymous user."""
|
||||
return False
|
||||
|
||||
def get_username(self) -> str:
|
||||
"""Return the machine user identifier."""
|
||||
return self.username
|
||||
|
||||
|
||||
class StorageEventAuthentication(BaseAuthentication):
|
||||
"""Authenticate requests using a Bearer token for storage event integration.
|
||||
This class validates Bearer tokens for storage events that don't map to database users.
|
||||
It's designed for S3-compatible storage integrations and similar use cases.
|
||||
Events are submitted when a webhook is configured on some bucket's events.
|
||||
"""
|
||||
|
||||
AUTH_HEADER = "Authorization"
|
||||
TOKEN_TYPE = "Bearer" # noqa S105
|
||||
|
||||
def authenticate(self, request):
|
||||
"""Validate the Bearer token from the Authorization header."""
|
||||
if not settings.RECORDING_ENABLE_STORAGE_EVENT_AUTH:
|
||||
return MachineUser(), None
|
||||
|
||||
if not settings.RECORDING_ENABLE_STORAGE_EVENT_AUTH:
|
||||
return MachineUser(), None
|
||||
|
||||
required_token = settings.RECORDING_STORAGE_EVENT_TOKEN
|
||||
if not required_token:
|
||||
if settings.RECORDING_ENABLE_STORAGE_EVENT_AUTH:
|
||||
raise AuthenticationFailed(
|
||||
_("Authentication is enabled but token is not configured.")
|
||||
)
|
||||
|
||||
return MachineUser(), None
|
||||
|
||||
auth_header = request.headers.get(self.AUTH_HEADER)
|
||||
|
||||
if not auth_header:
|
||||
logger.warning(
|
||||
"Authentication failed: Missing Authorization header (ip: %s)",
|
||||
request.META.get("REMOTE_ADDR"),
|
||||
)
|
||||
raise AuthenticationFailed(_("Authorization header is required"))
|
||||
|
||||
auth_parts = auth_header.split(" ")
|
||||
if len(auth_parts) != 2 or auth_parts[0] != self.TOKEN_TYPE:
|
||||
logger.warning(
|
||||
"Authentication failed: Invalid authorization header (ip: %s)",
|
||||
request.META.get("REMOTE_ADDR"),
|
||||
)
|
||||
raise AuthenticationFailed(_("Invalid authorization header."))
|
||||
|
||||
token = auth_parts[1]
|
||||
|
||||
# Use constant-time comparison to prevent timing attacks
|
||||
if not secrets.compare_digest(token.encode(), required_token.encode()):
|
||||
logger.warning(
|
||||
"Authentication failed: Invalid token (ip: %s)",
|
||||
request.META.get("REMOTE_ADDR"),
|
||||
)
|
||||
raise AuthenticationFailed(_("Invalid token"))
|
||||
|
||||
return MachineUser(), token
|
||||
|
||||
def authenticate_header(self, request):
|
||||
"""Return the WWW-Authenticate header value."""
|
||||
return f"{self.TOKEN_TYPE} realm='Storage event API'"
|
||||
@@ -1,17 +0,0 @@
|
||||
"""Storage parsers specific exceptions."""
|
||||
|
||||
|
||||
class ParsingEventDataError(Exception):
|
||||
"""Raised when the request data is malformed, incomplete, or missing."""
|
||||
|
||||
|
||||
class InvalidBucketError(Exception):
|
||||
"""Raised when the bucket name in the request does not match the expected one."""
|
||||
|
||||
|
||||
class InvalidFileTypeError(Exception):
|
||||
"""Raised when the file type in the request is not supported."""
|
||||
|
||||
|
||||
class InvalidFilepathError(Exception):
|
||||
"""Raised when the filepath in the request is invalid."""
|
||||
@@ -1,147 +0,0 @@
|
||||
"""Meet storage event parser classes."""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from typing import Any, Dict, Optional, Protocol
|
||||
|
||||
from django.conf import settings
|
||||
from django.utils.module_loading import import_string
|
||||
|
||||
from .exceptions import (
|
||||
InvalidBucketError,
|
||||
InvalidFilepathError,
|
||||
InvalidFileTypeError,
|
||||
ParsingEventDataError,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class StorageEvent:
|
||||
"""Represents a storage event with relevant metadata.
|
||||
Attributes:
|
||||
filepath: Identifier for the affected recording
|
||||
filetype: Type of storage event
|
||||
bucket_name: When the event occurred
|
||||
metadata: Additional event data
|
||||
"""
|
||||
|
||||
filepath: str
|
||||
filetype: str
|
||||
bucket_name: str
|
||||
metadata: Optional[Dict[str, Any]]
|
||||
|
||||
def __post_init__(self):
|
||||
if self.filepath is None:
|
||||
raise TypeError("filepath cannot be None")
|
||||
if self.filetype is None:
|
||||
raise TypeError("filetype cannot be None")
|
||||
if self.bucket_name is None:
|
||||
raise TypeError("bucket_name cannot be None")
|
||||
|
||||
|
||||
class EventParser(Protocol):
|
||||
"""Interface for parsing storage events."""
|
||||
|
||||
def __init__(self, bucket_name, allowed_filetypes=None):
|
||||
"""Initialize parser with bucket name and optional allowed filetypes."""
|
||||
|
||||
def parse(self, data: Dict) -> StorageEvent:
|
||||
"""Extract storage event data from raw dictionary input."""
|
||||
|
||||
def validate(self, data: StorageEvent) -> None:
|
||||
"""Verify storage event data meets all requirements."""
|
||||
|
||||
def get_recording_id(self, data: Dict) -> str:
|
||||
"""Extract recording ID from event dictionary."""
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def get_parser() -> EventParser:
|
||||
"""Return cached instance of configured event parser.
|
||||
Uses function memoization instead of a factory class since the only
|
||||
varying parameter is the parser class from settings. A factory class
|
||||
would add unnecessary complexity when a cached function provides the
|
||||
same singleton behavior with simpler code.
|
||||
"""
|
||||
|
||||
event_parser_cls = import_string(settings.RECORDING_EVENT_PARSER_CLASS)
|
||||
return event_parser_cls(bucket_name=settings.AWS_STORAGE_BUCKET_NAME)
|
||||
|
||||
|
||||
class MinioParser:
|
||||
"""Handle parsing and validation of Minio storage events."""
|
||||
|
||||
def __init__(self, bucket_name: str, allowed_filetypes=None):
|
||||
"""Initialize parser with target bucket name and accepted filetypes."""
|
||||
|
||||
if not bucket_name:
|
||||
raise ValueError("Bucket name cannot be None or empty")
|
||||
|
||||
self._bucket_name = bucket_name
|
||||
self._allowed_filetypes = allowed_filetypes or {"audio/ogg", "video/mp4"}
|
||||
|
||||
# pylint: disable=line-too-long
|
||||
self._filepath_regex = re.compile(
|
||||
r"(?P<url_encoded_folder_path>(?:[^%]+%2F)*)?(?P<recording_id>[0-9a-fA-F\-]{36})\.(?P<extension>[a-zA-Z0-9]+)"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def parse(data):
|
||||
"""Convert raw Minio event dictionary to StorageEvent object."""
|
||||
|
||||
if not data:
|
||||
raise ParsingEventDataError("Received empty data.")
|
||||
|
||||
try:
|
||||
record = data["Records"][0]
|
||||
s3 = record["s3"]
|
||||
bucket_name = s3["bucket"]["name"]
|
||||
file_object = s3["object"]
|
||||
filepath = file_object["key"]
|
||||
filetype = file_object["contentType"]
|
||||
except (KeyError, IndexError) as e:
|
||||
raise ParsingEventDataError(f"Missing or malformed key: {e}.") from e
|
||||
try:
|
||||
return StorageEvent(
|
||||
filepath=filepath,
|
||||
filetype=filetype,
|
||||
bucket_name=bucket_name,
|
||||
metadata=None,
|
||||
)
|
||||
except TypeError as e:
|
||||
raise ParsingEventDataError(f"Missing essential data fields: {e}") from e
|
||||
|
||||
def validate(self, event_data: StorageEvent) -> str:
|
||||
"""Verify StorageEvent matches bucket, filetype and filepath requirements."""
|
||||
|
||||
if event_data.bucket_name != self._bucket_name:
|
||||
raise InvalidBucketError(
|
||||
f"Invalid bucket: expected {self._bucket_name}, got {event_data.bucket_name}"
|
||||
)
|
||||
|
||||
if not event_data.filetype in self._allowed_filetypes:
|
||||
raise InvalidFileTypeError(
|
||||
f"Invalid file type, expected {self._allowed_filetypes},"
|
||||
f"got '{event_data.filetype}'"
|
||||
)
|
||||
|
||||
match = self._filepath_regex.match(event_data.filepath)
|
||||
if not match:
|
||||
raise InvalidFilepathError(
|
||||
f"Invalid filepath structure: {event_data.filepath}"
|
||||
)
|
||||
|
||||
recording_id = match.group("recording_id")
|
||||
return recording_id
|
||||
|
||||
def get_recording_id(self, data):
|
||||
"""Extract recording ID from Minio event through parsing and validation."""
|
||||
|
||||
event_data = self.parse(data)
|
||||
recording_id = self.validate(event_data)
|
||||
|
||||
return recording_id
|
||||
@@ -1 +0,0 @@
|
||||
"""Meet worker services classes and exceptions."""
|
||||
@@ -1,21 +0,0 @@
|
||||
"""Recording and worker services specific exceptions."""
|
||||
|
||||
|
||||
class WorkerRequestError(Exception):
|
||||
"""Raised when there is an issue with the worker request"""
|
||||
|
||||
|
||||
class WorkerConnectionError(Exception):
|
||||
"""Raised when there is an issue connecting to the worker."""
|
||||
|
||||
|
||||
class WorkerResponseError(Exception):
|
||||
"""Raised when the worker's response is not as expected."""
|
||||
|
||||
|
||||
class RecordingStartError(Exception):
|
||||
"""Raised when there is an error starting the recording."""
|
||||
|
||||
|
||||
class RecordingStopError(Exception):
|
||||
"""Raised when there is an error stopping the recording."""
|
||||
@@ -1,75 +0,0 @@
|
||||
"""Factory, configurations and Protocol to create worker services"""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from typing import Any, ClassVar, Dict, Optional, Protocol, Type
|
||||
|
||||
from django.conf import settings
|
||||
from django.utils.module_loading import import_string
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkerServiceConfig:
|
||||
"""Declare Worker Service common configurations"""
|
||||
|
||||
output_folder: str
|
||||
server_configurations: Dict[str, Any]
|
||||
verify_ssl: Optional[bool]
|
||||
bucket_args: Optional[dict]
|
||||
|
||||
@classmethod
|
||||
@lru_cache
|
||||
def from_settings(cls) -> "WorkerServiceConfig":
|
||||
"""Load configuration from Django settings with caching for efficiency."""
|
||||
|
||||
logger.debug("Loading WorkerServiceConfig from settings.")
|
||||
return cls(
|
||||
output_folder=settings.RECORDING_OUTPUT_FOLDER,
|
||||
server_configurations=settings.LIVEKIT_CONFIGURATION,
|
||||
verify_ssl=settings.RECORDING_VERIFY_SSL,
|
||||
bucket_args={
|
||||
"endpoint": settings.AWS_S3_ENDPOINT_URL,
|
||||
"access_key": settings.AWS_S3_ACCESS_KEY_ID,
|
||||
"secret": settings.AWS_S3_SECRET_ACCESS_KEY,
|
||||
"region": settings.AWS_S3_REGION_NAME,
|
||||
"bucket": settings.AWS_STORAGE_BUCKET_NAME,
|
||||
"force_path_style": True,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class WorkerService(Protocol):
|
||||
"""Define the interface for interacting with a worker service."""
|
||||
|
||||
hrid: ClassVar[str]
|
||||
|
||||
def __init__(self, config: WorkerServiceConfig):
|
||||
"""Initialize the service with the given configuration."""
|
||||
|
||||
def start(self, room_id: str, recording_id: str) -> str:
|
||||
"""Start a recording for a specified room."""
|
||||
|
||||
def stop(self, worker_id: str) -> str:
|
||||
"""Stop recording for a specified worker."""
|
||||
|
||||
|
||||
def get_worker_service(mode: str) -> WorkerService:
|
||||
"""Instantiate a worker service by its mode."""
|
||||
|
||||
worker_registry: Dict[str, str] = settings.RECORDING_WORKER_CLASSES
|
||||
|
||||
try:
|
||||
worker_class_path = worker_registry[mode]
|
||||
except KeyError as e:
|
||||
raise ValueError(
|
||||
f"Recording mode '{mode}' not found in RECORDING_WORKER_CLASSES. "
|
||||
f"Available modes: {list(worker_registry.keys())}"
|
||||
) from e
|
||||
|
||||
worker_class: Type[WorkerService] = import_string(worker_class_path)
|
||||
|
||||
config = WorkerServiceConfig.from_settings()
|
||||
return worker_class(config=config)
|
||||
@@ -1,98 +0,0 @@
|
||||
"""Mediator between the worker service and recording instances in the Django ORM."""
|
||||
|
||||
import logging
|
||||
|
||||
from core.models import Recording, RecordingStatusChoices
|
||||
|
||||
from .exceptions import (
|
||||
RecordingStartError,
|
||||
RecordingStopError,
|
||||
WorkerConnectionError,
|
||||
WorkerRequestError,
|
||||
WorkerResponseError,
|
||||
)
|
||||
from .factories import WorkerService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class WorkerServiceMediator:
|
||||
"""Mediate interactions between a worker service and a recording instance.
|
||||
|
||||
A mediator class that decouples the worker from Django ORM, handles recording updates
|
||||
based on worker status, and transforms worker errors into user-friendly exceptions.
|
||||
Implements Mediator pattern.
|
||||
"""
|
||||
|
||||
def __init__(self, worker_service: WorkerService):
|
||||
"""Initialize the WorkerServiceMediator with the provided worker service."""
|
||||
|
||||
self._worker_service = worker_service
|
||||
|
||||
def start(self, recording: Recording):
|
||||
"""Start the recording process using the worker service.
|
||||
|
||||
If the operation is successful, the recording's status will
|
||||
transition from INITIATED to ACTIVE, else to FAILED_TO_START to keep track of errors.
|
||||
|
||||
Args:
|
||||
recording (Recording): The recording instance to start.
|
||||
Raises:
|
||||
RecordingStartError: If there is an error starting the recording.
|
||||
"""
|
||||
|
||||
if recording.status != RecordingStatusChoices.INITIATED:
|
||||
logger.error("Cannot start recording in %s status.", recording.status)
|
||||
raise RecordingStartError()
|
||||
|
||||
room_name = str(recording.room.id)
|
||||
try:
|
||||
worker_id = self._worker_service.start(room_name, recording.id)
|
||||
except (WorkerRequestError, WorkerConnectionError, WorkerResponseError) as e:
|
||||
logger.exception(
|
||||
"Failed to start recording for room %s: %s", recording.room.slug, e
|
||||
)
|
||||
recording.status = RecordingStatusChoices.FAILED_TO_START
|
||||
raise RecordingStartError() from e
|
||||
else:
|
||||
recording.worker_id = worker_id
|
||||
recording.status = RecordingStatusChoices.ACTIVE
|
||||
finally:
|
||||
recording.save()
|
||||
|
||||
logger.info(
|
||||
"Worker started for room %s (worker ID: %s)",
|
||||
recording.room,
|
||||
recording.worker_id,
|
||||
)
|
||||
|
||||
def stop(self, recording: Recording):
|
||||
"""Stop the recording process using the worker service.
|
||||
|
||||
If the operation is successful, the recording's status will transition
|
||||
from ACTIVE to STOPPED, else to FAILED_TO_STOP to keep track of errors.
|
||||
|
||||
Args:
|
||||
recording (Recording): The recording instance to stop.
|
||||
Raises:
|
||||
RecordingStopError: If there is an error stopping the recording.
|
||||
"""
|
||||
|
||||
if recording.status != RecordingStatusChoices.ACTIVE:
|
||||
logger.error("Cannot stop recording in %s status.", recording.status)
|
||||
raise RecordingStopError()
|
||||
|
||||
try:
|
||||
response = self._worker_service.stop(worker_id=recording.worker_id)
|
||||
except (WorkerConnectionError, WorkerResponseError) as e:
|
||||
logger.exception(
|
||||
"Failed to stop recording for room %s: %s", recording.room.slug, e
|
||||
)
|
||||
recording.status = RecordingStatusChoices.FAILED_TO_STOP
|
||||
raise RecordingStopError() from e
|
||||
else:
|
||||
recording.status = RecordingStatusChoices[response]
|
||||
finally:
|
||||
recording.save()
|
||||
|
||||
logger.info("Worker stopped for room %s", recording.room)
|
||||
@@ -1,140 +0,0 @@
|
||||
"""Worker services in charge of recording a room."""
|
||||
|
||||
# pylint: disable=no-member
|
||||
|
||||
import aiohttp
|
||||
from asgiref.sync import async_to_sync
|
||||
from livekit import api as livekit_api
|
||||
from livekit.api.egress_service import EgressService
|
||||
|
||||
from .exceptions import WorkerConnectionError, WorkerResponseError
|
||||
from .factories import WorkerServiceConfig
|
||||
|
||||
|
||||
class BaseEgressService:
|
||||
"""Base egress defining common methods to manage and interact with LiveKit egress processes."""
|
||||
|
||||
def __init__(self, config: WorkerServiceConfig):
|
||||
self._config = config
|
||||
self._s3 = livekit_api.S3Upload(**config.bucket_args)
|
||||
|
||||
def _get_filepath(self, filename: str, extension: str) -> str:
|
||||
"""Construct the file path for a given filename and extension.
|
||||
Unsecure method, doesn't handle paths robustly and securely.
|
||||
"""
|
||||
return f"{self._config.output_folder}/{filename}.{extension}"
|
||||
|
||||
@async_to_sync
|
||||
async def _handle_request(self, request, method_name: str):
|
||||
"""Handle making a request to the LiveKit API and returns the response."""
|
||||
|
||||
# Use HTTP connector for local development with Tilt,
|
||||
# where cluster communications are unsecure
|
||||
connector = aiohttp.TCPConnector(ssl=self._config.verify_ssl)
|
||||
|
||||
async with aiohttp.ClientSession(connector=connector) as session:
|
||||
client = EgressService(session, **self._config.server_configurations)
|
||||
method = getattr(client, method_name)
|
||||
try:
|
||||
response = await method(request)
|
||||
except livekit_api.TwirpError as e:
|
||||
raise WorkerConnectionError(
|
||||
f"LiveKit client connection error, {e.message}."
|
||||
) from e
|
||||
|
||||
return response
|
||||
|
||||
def stop(self, worker_id: str) -> str:
|
||||
"""Stop an ongoing egress worker.
|
||||
The StopEgressRequest is shared among all types of egress,
|
||||
so a single implementation in the base class should be sufficient.
|
||||
"""
|
||||
|
||||
request = livekit_api.StopEgressRequest(
|
||||
egress_id=worker_id,
|
||||
)
|
||||
|
||||
response = self._handle_request(request, "stop_egress")
|
||||
|
||||
if not response.status:
|
||||
raise WorkerResponseError(
|
||||
"LiveKit response is missing the recording status."
|
||||
)
|
||||
|
||||
# To avoid exposing EgressStatus values and coupling with LiveKit outside of this class,
|
||||
# the response status is mapped to simpler "ABORTED", "STOPPED" or "FAILED_TO_STOP" strings.
|
||||
if response.status == livekit_api.EgressStatus.EGRESS_ABORTED:
|
||||
return "ABORTED"
|
||||
|
||||
if response.status == livekit_api.EgressStatus.EGRESS_ENDING:
|
||||
return "STOPPED"
|
||||
|
||||
return "FAILED_TO_STOP"
|
||||
|
||||
def start(self, room_name, recording_id):
|
||||
"""Start the egress process for a recording (not implemented in the base class).
|
||||
Each derived class must implement this method, providing the necessary parameters for
|
||||
its specific egress type (e.g. audio_only, streaming output).
|
||||
"""
|
||||
raise NotImplementedError("Subclass must implement this method.")
|
||||
|
||||
|
||||
class VideoCompositeEgressService(BaseEgressService):
|
||||
"""Record multiple participant video and audio tracks into a single output '.mp4' file."""
|
||||
|
||||
hrid = "video-recording-composite-livekit-egress"
|
||||
|
||||
def start(self, room_name, recording_id):
|
||||
"""Start the video composite egress process for a recording."""
|
||||
|
||||
# Save room's recording as a mp4 video file.
|
||||
file_type = livekit_api.EncodedFileType.MP4
|
||||
filepath = self._get_filepath(filename=recording_id, extension="mp4")
|
||||
|
||||
file_output = livekit_api.EncodedFileOutput(
|
||||
file_type=file_type,
|
||||
filepath=filepath,
|
||||
s3=self._s3,
|
||||
)
|
||||
|
||||
request = livekit_api.RoomCompositeEgressRequest(
|
||||
room_name=room_name,
|
||||
file_outputs=[file_output],
|
||||
)
|
||||
|
||||
response = self._handle_request(request, "start_room_composite_egress")
|
||||
|
||||
if not response.egress_id:
|
||||
raise WorkerResponseError("Egress ID not found in the response.")
|
||||
|
||||
return response.egress_id
|
||||
|
||||
|
||||
class AudioCompositeEgressService(BaseEgressService):
|
||||
"""Record multiple participant audio tracks into a single output '.ogg' file."""
|
||||
|
||||
hrid = "audio-recording-composite-livekit-egress"
|
||||
|
||||
def start(self, room_name, recording_id):
|
||||
"""Start the audio composite egress process for a recording."""
|
||||
|
||||
# Save room's recording as an ogg audio file.
|
||||
file_type = livekit_api.EncodedFileType.OGG
|
||||
filepath = self._get_filepath(filename=recording_id, extension="ogg")
|
||||
|
||||
file_output = livekit_api.EncodedFileOutput(
|
||||
file_type=file_type,
|
||||
filepath=filepath,
|
||||
s3=self._s3,
|
||||
)
|
||||
|
||||
request = livekit_api.RoomCompositeEgressRequest(
|
||||
room_name=room_name, file_outputs=[file_output], audio_only=True
|
||||
)
|
||||
|
||||
response = self._handle_request(request, "start_room_composite_egress")
|
||||
|
||||
if not response.egress_id:
|
||||
raise WorkerResponseError("Egress ID not found in the response.")
|
||||
|
||||
return response.egress_id
|
||||
@@ -11,35 +11,32 @@ from core.factories import UserFactory
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_authentication_getter_existing_user(monkeypatch):
|
||||
def test_authentication_getter_existing_user_no_email(
|
||||
django_assert_num_queries, monkeypatch
|
||||
):
|
||||
"""
|
||||
If an existing user matches, the user should be returned.
|
||||
If an existing user matches the user's info sub, the user should be returned.
|
||||
"""
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(email="foo@mail.com")
|
||||
db_user = UserFactory()
|
||||
|
||||
def get_userinfo_mocked(*args):
|
||||
return {"sub": db_user.sub, "email": "some@mail.com"}
|
||||
|
||||
def get_existing_user(*args):
|
||||
return db_user
|
||||
return {"sub": db_user.sub}
|
||||
|
||||
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
|
||||
monkeypatch.setattr(
|
||||
OIDCAuthenticationBackend, "get_existing_user", get_existing_user
|
||||
)
|
||||
|
||||
user = klass.get_or_create_user(
|
||||
access_token="test-token", id_token=None, payload=None
|
||||
)
|
||||
with django_assert_num_queries(1):
|
||||
user = klass.get_or_create_user(
|
||||
access_token="test-token", id_token=None, payload=None
|
||||
)
|
||||
|
||||
assert user == db_user
|
||||
|
||||
|
||||
def test_authentication_getter_new_user_no_email(monkeypatch):
|
||||
"""
|
||||
If no user matches, a user should be created.
|
||||
If no user matches the user's info sub, a user should be created.
|
||||
User's info doesn't contain an email, created user's email should be empty.
|
||||
"""
|
||||
klass = OIDCAuthenticationBackend()
|
||||
@@ -61,7 +58,7 @@ def test_authentication_getter_new_user_no_email(monkeypatch):
|
||||
|
||||
def test_authentication_getter_new_user_with_email(monkeypatch):
|
||||
"""
|
||||
If no user matches, a user should be created.
|
||||
If no user matches the user's info sub, a user should be created.
|
||||
User's email and name should be set on the identity.
|
||||
The "email" field on the User model should not be set as it is reserved for staff users.
|
||||
"""
|
||||
@@ -105,183 +102,3 @@ def test_models_oidc_user_getter_invalid_token(django_assert_num_queries, monkey
|
||||
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
|
||||
|
||||
assert models.User.objects.exists() is False
|
||||
|
||||
|
||||
def test_models_oidc_user_getter_empty_sub(django_assert_num_queries, monkeypatch):
|
||||
"""The user's info contains a sub, but it's an empty string."""
|
||||
klass = OIDCAuthenticationBackend()
|
||||
|
||||
def get_userinfo_mocked(*args):
|
||||
return {"test": "123", "sub": ""}
|
||||
|
||||
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
|
||||
|
||||
with (
|
||||
django_assert_num_queries(0),
|
||||
pytest.raises(
|
||||
SuspiciousOperation,
|
||||
match="User info contained no recognizable user identification",
|
||||
),
|
||||
):
|
||||
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
|
||||
|
||||
assert models.User.objects.exists() is False
|
||||
|
||||
|
||||
def test_authentication_get_inactive_user(monkeypatch):
|
||||
"""Test an exception is raised when attempting to authenticate inactive user."""
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(is_active=False)
|
||||
|
||||
def get_userinfo_mocked(*args):
|
||||
return {"sub": db_user.sub}
|
||||
|
||||
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
|
||||
|
||||
with (
|
||||
pytest.raises(
|
||||
SuspiciousOperation,
|
||||
match="User account is disabled",
|
||||
),
|
||||
):
|
||||
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
|
||||
|
||||
|
||||
def test_finds_user_by_sub(django_assert_num_queries):
|
||||
"""Should return user when found by sub, and email is matching."""
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(email="foo@mail.com")
|
||||
|
||||
with django_assert_num_queries(1):
|
||||
user = klass.get_existing_user(db_user.sub, db_user.email)
|
||||
|
||||
assert user == db_user
|
||||
|
||||
|
||||
def test_finds_user_when_email_fallback_disabled(django_assert_num_queries, settings):
|
||||
"""Should not return a user when not found by sub, and email fallback is disabled."""
|
||||
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = False
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(email="foo@mail.com")
|
||||
|
||||
with django_assert_num_queries(1):
|
||||
user = klass.get_existing_user("wrong-sub", db_user.email)
|
||||
|
||||
assert user is None
|
||||
|
||||
|
||||
def test_finds_user_when_email_is_none(django_assert_num_queries, settings):
|
||||
"""Should not return a user when not found by sub, and email is empty."""
|
||||
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
UserFactory(email="foo@mail.com")
|
||||
|
||||
empty_email = ""
|
||||
|
||||
with django_assert_num_queries(1):
|
||||
user = klass.get_existing_user("wrong-sub", empty_email)
|
||||
|
||||
assert user is None
|
||||
|
||||
|
||||
def test_finds_user_by_email(django_assert_num_queries, settings):
|
||||
"""Should return user when found by email, and sub is not matching."""
|
||||
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(email="foo@mail.com")
|
||||
|
||||
with django_assert_num_queries(2):
|
||||
user = klass.get_existing_user("wrong-sub", db_user.email)
|
||||
|
||||
assert user == db_user
|
||||
|
||||
|
||||
def test_finds_user_case_insensitive_email(django_assert_num_queries, settings):
|
||||
"""Should match email case-insensitively when falling back to email."""
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(email="foo@mail.com")
|
||||
|
||||
with django_assert_num_queries(2):
|
||||
user = klass.get_existing_user("wrong-sub", "FOO@MAIL.COM")
|
||||
|
||||
assert user == db_user
|
||||
|
||||
|
||||
def test_finds_user_multiple_users_same_email(django_assert_num_queries, settings):
|
||||
"""Should handle multiple users with same email appropriately."""
|
||||
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
email = "foo@mail.com"
|
||||
UserFactory(email=email)
|
||||
UserFactory(email=email) # Second user with same email
|
||||
|
||||
with (
|
||||
django_assert_num_queries(2),
|
||||
pytest.raises(
|
||||
SuspiciousOperation,
|
||||
match="Multiple user accounts share a common email.",
|
||||
),
|
||||
):
|
||||
klass.get_existing_user("wrong-sub", email)
|
||||
|
||||
|
||||
def test_finds_user_whitespace_email(django_assert_num_queries, settings):
|
||||
"""Should not match emails with whitespace."""
|
||||
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True
|
||||
settings.OIDC_CREATE_USER = False
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
UserFactory(email="foo@mail.com")
|
||||
|
||||
with django_assert_num_queries(2):
|
||||
user = klass.get_existing_user("wrong-sub", " foo@mail.com ")
|
||||
|
||||
assert user is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"email",
|
||||
[
|
||||
"john.doe@example.com", # Fullwidth character in domain
|
||||
"john.doe@еxample.com", # Cyrillic 'е' in domain
|
||||
"JOHN.DOe@exam𝔭le.com", # Mixed Gothic '𝔭' in domain
|
||||
"john.doe@exаmple.com", # Cyrillic 'а' (a) in domain
|
||||
"john.doe@e𝓧𝓪𝓶𝓹𝓵𝓮.com", # Mixed fullwidth and cursive in domain
|
||||
],
|
||||
)
|
||||
def test_authentication_getter_existing_user_email_tricky(email, monkeypatch, settings):
|
||||
"""Test email matching security against visually similar but non-ASCII domains.
|
||||
|
||||
Validates that emails with Unicode characters that visually resemble ASCII
|
||||
(homoglyphs) are treated as distinct from their ASCII counterparts for security,
|
||||
per RFC compliance requirements for hostnames.
|
||||
"""
|
||||
|
||||
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
db_user = UserFactory(email="john.doe@example.com")
|
||||
|
||||
def get_userinfo_mocked(*args):
|
||||
return {"sub": "123", "email": email}
|
||||
|
||||
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
|
||||
|
||||
user = klass.get_or_create_user(
|
||||
access_token="test-token", id_token=None, payload=None
|
||||
)
|
||||
|
||||
assert user != db_user
|
||||
|
||||
@@ -1,145 +0,0 @@
|
||||
"""
|
||||
Test event authentication.
|
||||
"""
|
||||
|
||||
# pylint: disable=E1128
|
||||
|
||||
from django.test import RequestFactory
|
||||
|
||||
import pytest
|
||||
from rest_framework.exceptions import AuthenticationFailed
|
||||
|
||||
from core.recording.event.authentication import (
|
||||
MachineUser,
|
||||
StorageEventAuthentication,
|
||||
)
|
||||
|
||||
|
||||
def test_successful_authentication(settings):
|
||||
"""Test successful authentication with valid token."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "Bearer valid-test-token"}
|
||||
|
||||
user, token = StorageEventAuthentication().authenticate(request)
|
||||
assert token == "valid-test-token"
|
||||
assert isinstance(user, MachineUser)
|
||||
|
||||
|
||||
def test_disabled_authentication_with_header(settings):
|
||||
"""Authentication should pass when no auth is configured, and header is present."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = None
|
||||
settings.RECORDING_ENABLE_STORAGE_EVENT_AUTH = False
|
||||
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "Bearer some-token"}
|
||||
|
||||
user, token = StorageEventAuthentication().authenticate(request)
|
||||
assert token is None
|
||||
assert isinstance(user, MachineUser)
|
||||
|
||||
|
||||
def test_disabled_authentication_without_header(settings):
|
||||
"""Authentication should pass when no auth is configured, and no header is present."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = None
|
||||
settings.RECORDING_ENABLE_STORAGE_EVENT_AUTH = False
|
||||
|
||||
request = RequestFactory().get("/")
|
||||
|
||||
user, token = StorageEventAuthentication().authenticate(request)
|
||||
assert token is None
|
||||
assert isinstance(user, MachineUser)
|
||||
|
||||
|
||||
def test_authentication_when_disabled(settings):
|
||||
"""Authentication should pass when disabled, regardless of token configuration."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "some-token"
|
||||
settings.RECORDING_ENABLE_STORAGE_EVENT_AUTH = False
|
||||
|
||||
request = RequestFactory().get("/")
|
||||
|
||||
user, token = StorageEventAuthentication().authenticate(request)
|
||||
assert token is None
|
||||
assert isinstance(user, MachineUser)
|
||||
|
||||
|
||||
def test_authentication_fails_when_token_not_configured(settings):
|
||||
"""Authentication should fail when authentication is enabled but no token is configured."""
|
||||
|
||||
# By default RECORDING_ENABLE_STORAGE_EVENT_AUTH should be True
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = None
|
||||
|
||||
request = RequestFactory().get("/")
|
||||
|
||||
with pytest.raises(
|
||||
AuthenticationFailed,
|
||||
match="Authentication is enabled but token is not configured",
|
||||
):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
|
||||
|
||||
def test_missing_auth_header(settings):
|
||||
"""Test failure when Authorization header is missing."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {}
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Authorization header is required"):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
|
||||
|
||||
def test_invalid_auth_header_format(settings):
|
||||
"""Test failure when Authorization header has invalid format."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "InvalidFormat"}
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
|
||||
|
||||
def test_invalid_token_type(settings):
|
||||
"""Test failure when token type is not Bearer."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "Basic some-token"}
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
|
||||
|
||||
def test_invalid_token(settings):
|
||||
"""Test failure when token is invalid."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "Bearer wrong-token"}
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid token"):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
|
||||
|
||||
def test_malformed_auth_header(settings):
|
||||
"""Test failure when Authorization header is malformed."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "Bearer"} # Missing token part
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
|
||||
|
||||
def test_authenticate_header():
|
||||
"""Test the WWW-Authenticate header value."""
|
||||
request = RequestFactory().get("/")
|
||||
header = StorageEventAuthentication().authenticate_header(request)
|
||||
assert header == "Bearer realm='Storage event API'"
|
||||
|
||||
|
||||
def test_multiple_spaces_in_auth_header(settings):
|
||||
"""Test failure when Authorization header contains multiple spaces."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "valid-test-token"
|
||||
request = RequestFactory().get("/")
|
||||
request.headers = {"Authorization": "Bearer extra-spaces-token"}
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
|
||||
StorageEventAuthentication().authenticate(request)
|
||||
@@ -1,310 +0,0 @@
|
||||
"""
|
||||
Test event parsers.
|
||||
"""
|
||||
|
||||
# pylint: disable=W0212,W0621,W0613
|
||||
|
||||
from unittest import mock
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
import pytest
|
||||
|
||||
from core.recording.event.exceptions import (
|
||||
InvalidBucketError,
|
||||
InvalidFilepathError,
|
||||
InvalidFileTypeError,
|
||||
ParsingEventDataError,
|
||||
)
|
||||
from core.recording.event.parsers import (
|
||||
MinioParser,
|
||||
StorageEvent,
|
||||
get_parser,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_minio_event():
|
||||
"""Mock a valid Minio event."""
|
||||
return {
|
||||
"Records": [
|
||||
{
|
||||
"s3": {
|
||||
"bucket": {"name": "test-bucket"},
|
||||
"object": {
|
||||
"key": "recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
"contentType": "audio/ogg",
|
||||
},
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def minio_parser():
|
||||
"""Mock a Minio parser."""
|
||||
return MinioParser(bucket_name="test-bucket")
|
||||
|
||||
|
||||
def test_parse_valid_event(minio_parser, valid_minio_event):
|
||||
"""Test parsing a valid Minio event."""
|
||||
event = minio_parser.parse(valid_minio_event)
|
||||
assert isinstance(event, StorageEvent)
|
||||
assert event.filepath == "recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg"
|
||||
assert event.filetype == "audio/ogg"
|
||||
assert event.bucket_name == "test-bucket"
|
||||
assert event.metadata is None
|
||||
|
||||
|
||||
def test_parse_empty_data(minio_parser):
|
||||
"""Test parsing empty event data raises error."""
|
||||
with pytest.raises(ParsingEventDataError, match="Received empty data."):
|
||||
minio_parser.parse({})
|
||||
|
||||
|
||||
def test_parse_missing_keys(minio_parser):
|
||||
"""Test parsing event with missing key."""
|
||||
|
||||
invalid_minio_event = {
|
||||
"Records": [
|
||||
{
|
||||
"s3": {
|
||||
"bucket": {"name": None},
|
||||
# Missing 'object' key
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
with pytest.raises(ParsingEventDataError, match="Missing or malformed key"):
|
||||
minio_parser.parse(invalid_minio_event)
|
||||
|
||||
|
||||
def test_parse_none_key(minio_parser):
|
||||
"""Test parsing event with None field."""
|
||||
|
||||
invalid_minio_event = {
|
||||
"Records": [
|
||||
{
|
||||
"s3": {
|
||||
"bucket": {"name": "test-bucket"},
|
||||
"object": {
|
||||
"key": "recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
"contentType": None, # 'contentType' should not be None
|
||||
},
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
with pytest.raises(ParsingEventDataError, match="Missing essential data fields"):
|
||||
minio_parser.parse(invalid_minio_event)
|
||||
|
||||
|
||||
def test_validate_invalid_bucket(minio_parser):
|
||||
"""Test validation with wrong bucket name."""
|
||||
event = StorageEvent(
|
||||
filepath="recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
filetype="audio/ogg",
|
||||
bucket_name="wrong-bucket",
|
||||
metadata=None,
|
||||
)
|
||||
with pytest.raises(InvalidBucketError):
|
||||
minio_parser.validate(event)
|
||||
|
||||
|
||||
def test_validate_invalid_filetype(minio_parser):
|
||||
"""Test validation with unsupported file type."""
|
||||
event = StorageEvent(
|
||||
filepath="recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.txt",
|
||||
filetype="text/plain", # Not included in the default allowed filetypes
|
||||
bucket_name="test-bucket",
|
||||
metadata=None,
|
||||
)
|
||||
with pytest.raises(InvalidFileTypeError):
|
||||
minio_parser.validate(event)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"invalid_filepath",
|
||||
[
|
||||
"invalid_filepath",
|
||||
"recording/46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
"recording%2F46d1a1212426484d8fb309b5d886f7a8.ogg",
|
||||
],
|
||||
)
|
||||
def test_validate_invalid_filepath(invalid_filepath, minio_parser):
|
||||
"""Test validation with malformed filepath."""
|
||||
event = StorageEvent(
|
||||
filepath=invalid_filepath,
|
||||
filetype="audio/ogg",
|
||||
bucket_name="test-bucket",
|
||||
metadata=None,
|
||||
)
|
||||
with pytest.raises(InvalidFilepathError):
|
||||
minio_parser.validate(event)
|
||||
|
||||
|
||||
def test_validate_valid_event(minio_parser):
|
||||
"""Test validation with valid event data."""
|
||||
event = StorageEvent(
|
||||
filepath="recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
filetype="audio/ogg",
|
||||
bucket_name="test-bucket",
|
||||
metadata=None,
|
||||
)
|
||||
recording_id = minio_parser.validate(event)
|
||||
assert recording_id == "46d1a121-2426-484d-8fb3-09b5d886f7a8"
|
||||
|
||||
|
||||
def test_get_recording_id_success(minio_parser, valid_minio_event):
|
||||
"""Test successful extraction of recording ID."""
|
||||
recording_id = minio_parser.get_recording_id(valid_minio_event)
|
||||
assert recording_id == "46d1a121-2426-484d-8fb3-09b5d886f7a8"
|
||||
|
||||
|
||||
def test_validate_filepath_with_folder(minio_parser):
|
||||
"""Test validation of filepath with folder structure."""
|
||||
event = StorageEvent(
|
||||
filepath="parent_folder%2Ffolder%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
filetype="audio/ogg",
|
||||
bucket_name="test-bucket",
|
||||
metadata=None,
|
||||
)
|
||||
recording_id = minio_parser.validate(event)
|
||||
assert recording_id == "46d1a121-2426-484d-8fb3-09b5d886f7a8"
|
||||
|
||||
|
||||
def test_parse_with_video_type(minio_parser):
|
||||
"""Test parsing event with video file type."""
|
||||
video_event = {
|
||||
"Records": [
|
||||
{
|
||||
"s3": {
|
||||
"bucket": {"name": "test-bucket"},
|
||||
"object": {
|
||||
"key": "46d1a121-2426-484d-8fb3-09b5d886f7a8.mp4",
|
||||
"contentType": "video/mp4",
|
||||
},
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
event = minio_parser.parse(video_event)
|
||||
assert event.filetype == "video/mp4"
|
||||
assert event.filepath.endswith(".mp4")
|
||||
|
||||
|
||||
def test_empty_allowed_filetypes():
|
||||
"""Test MinioParser with empty allowed_filetypes."""
|
||||
empty_types = set()
|
||||
parser = MinioParser(bucket_name="test-bucket", allowed_filetypes=empty_types)
|
||||
assert parser._allowed_filetypes == {"audio/ogg", "video/mp4"}
|
||||
|
||||
|
||||
def test_custom_allowed_filetypes():
|
||||
"""Test MinioParser with empty allowed_filetypes."""
|
||||
custom_types = {"audio/mp3", "video/mov"}
|
||||
parser = MinioParser(bucket_name="test-bucket", allowed_filetypes=custom_types)
|
||||
assert parser._allowed_filetypes == {"audio/mp3", "video/mov"}
|
||||
|
||||
|
||||
def test_validate_custom_filetypes():
|
||||
"""Test validation of filepath with folder structure."""
|
||||
|
||||
parser = MinioParser(bucket_name="test-bucket", allowed_filetypes={"audio/mp3"})
|
||||
|
||||
event = StorageEvent(
|
||||
filepath="parent_folder%2Ffolder%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
|
||||
filetype="audio/mp3",
|
||||
bucket_name="test-bucket",
|
||||
metadata=None,
|
||||
)
|
||||
parser.validate(event)
|
||||
|
||||
|
||||
def test_constructor_none_bucket():
|
||||
"""Test MinioParser constructor with None bucket name."""
|
||||
with pytest.raises(ValueError, match="Bucket name cannot be None or empty"):
|
||||
MinioParser(bucket_name=None)
|
||||
|
||||
|
||||
def test_constructor_empty_bucket():
|
||||
"""Test MinioParser constructor with empty bucket name."""
|
||||
with pytest.raises(ValueError, match="Bucket name cannot be None or empty"):
|
||||
MinioParser(bucket_name="")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clear_lru_cache():
|
||||
"""Fixture to clear the LRU cache between tests."""
|
||||
get_parser.cache_clear()
|
||||
yield
|
||||
get_parser.cache_clear()
|
||||
|
||||
|
||||
def test_returns_correct_instance(clear_lru_cache):
|
||||
"""Test if get_parser returns the correct parser instance."""
|
||||
settings.AWS_STORAGE_BUCKET_NAME = "test-bucket"
|
||||
parser = get_parser()
|
||||
assert isinstance(parser, MinioParser)
|
||||
assert parser._bucket_name == "test-bucket"
|
||||
|
||||
|
||||
def test_caching_behavior(clear_lru_cache):
|
||||
"""Test if the function properly caches the parser instance."""
|
||||
settings.AWS_STORAGE_BUCKET_NAME = "test-bucket"
|
||||
parser1 = get_parser()
|
||||
parser2 = get_parser()
|
||||
assert parser1 is parser2 # Check object identity
|
||||
|
||||
|
||||
def test_different_settings_new_instance():
|
||||
"""Test if changing settings creates a new instance."""
|
||||
settings.AWS_STORAGE_BUCKET_NAME = "different-bucket"
|
||||
parser = get_parser()
|
||||
assert parser._bucket_name == "different-bucket"
|
||||
|
||||
|
||||
def test_import_error_handling(clear_lru_cache):
|
||||
"""Test handling of import errors for invalid parser class."""
|
||||
settings.RECORDING_EVENT_PARSER_CLASS = "invalid.parser.path"
|
||||
with pytest.raises(ImportError):
|
||||
get_parser()
|
||||
|
||||
|
||||
@mock.patch("core.recording.event.parsers.import_string")
|
||||
def test_parser_instantiation_called_once(mock_import_string, clear_lru_cache):
|
||||
"""Test that parser class is instantiated only once due to caching."""
|
||||
mock_parser_cls = type(
|
||||
"MockParser",
|
||||
(),
|
||||
{
|
||||
"__init__": lambda self, bucket_name: setattr(
|
||||
self, "_bucket_name", bucket_name
|
||||
)
|
||||
},
|
||||
)
|
||||
mock_import_string.return_value = mock_parser_cls
|
||||
|
||||
# First call
|
||||
parser1 = get_parser()
|
||||
# Second call
|
||||
parser2 = get_parser()
|
||||
|
||||
# Verify import_string was called only once
|
||||
mock_import_string.assert_called_once_with(settings.RECORDING_EVENT_PARSER_CLASS)
|
||||
assert parser1 is parser2
|
||||
|
||||
|
||||
def test_cache_clear_behavior(clear_lru_cache, settings):
|
||||
"""Test that cache clearing creates new instance."""
|
||||
|
||||
settings.RECORDING_EVENT_PARSER_CLASS = "core.recording.event.parsers.MinioParser"
|
||||
|
||||
parser1 = get_parser()
|
||||
get_parser.cache_clear()
|
||||
parser2 = get_parser()
|
||||
|
||||
assert parser1 is not parser2 # Should be different instances after cache clear
|
||||
@@ -1,203 +0,0 @@
|
||||
"""
|
||||
Test recordings API endpoints in the Meet core app: save recording.
|
||||
"""
|
||||
|
||||
# pylint: disable=W0621,W0613
|
||||
|
||||
import uuid
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from ...factories import RecordingFactory
|
||||
from ...models import Recording, RecordingStatusChoices
|
||||
from ...recording.event.exceptions import (
|
||||
InvalidBucketError,
|
||||
InvalidFileTypeError,
|
||||
ParsingEventDataError,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def recording_settings(settings):
|
||||
"""Configure recording-related and storage event Django settings."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "testAuthToken"
|
||||
settings.RECORDING_STORAGE_EVENT_ENABLE = True
|
||||
return settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_get_parser():
|
||||
"""Mock 'get_parser' factory function."""
|
||||
with mock.patch("core.api.viewsets.get_parser") as mock_parser:
|
||||
yield mock_parser
|
||||
|
||||
|
||||
def test_save_recording_anonymous(settings, client):
|
||||
"""Anonymous users should not be allowed to save room recordings."""
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "testAuthToken"
|
||||
|
||||
RecordingFactory(status="active")
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert Recording.objects.count() == 1
|
||||
|
||||
|
||||
def test_save_recording_wrong_bearer(settings, client):
|
||||
"""Requests with incorrect bearer token should be rejected when auth is required."""
|
||||
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "testAuthToken"
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer wrongAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_save_recording_permission_needed(settings, client):
|
||||
"""Recordings should not be saved when feature is disabled."""
|
||||
|
||||
settings.RECORDING_STORAGE_EVENT_TOKEN = "testAuthToken"
|
||||
settings.RECORDING_STORAGE_EVENT_ENABLE = False
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_save_recording_parsing_error(recording_settings, mock_get_parser, client):
|
||||
"""Test handling of parsing errors in recording event data."""
|
||||
mock_parser = mock.Mock()
|
||||
mock_parser.get_recording_id.side_effect = ParsingEventDataError("Error message")
|
||||
mock_get_parser.return_value = mock_parser
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "Invalid request data: Error message"}
|
||||
|
||||
|
||||
def test_save_recording_bucket_error(recording_settings, mock_get_parser, client):
|
||||
"""Test handling of invalid storage bucket errors in recording event data."""
|
||||
|
||||
mock_parser = mock.Mock()
|
||||
mock_parser.get_recording_id.side_effect = InvalidBucketError("Error message")
|
||||
mock_get_parser.return_value = mock_parser
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "Invalid bucket specified"}
|
||||
|
||||
|
||||
def test_save_recording_filetype_error(recording_settings, mock_get_parser):
|
||||
"""Test handling of unsupported file types in recording event data."""
|
||||
|
||||
mock_parser = mock.Mock()
|
||||
mock_parser.get_recording_id.side_effect = InvalidFileTypeError(
|
||||
"unsupported '.json'"
|
||||
)
|
||||
mock_get_parser.return_value = mock_parser
|
||||
|
||||
client = APIClient()
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"message": "Ignore this file type, unsupported '.json'"}
|
||||
|
||||
|
||||
def test_save_recording_unknown_recording(recording_settings, mock_get_parser, client):
|
||||
"""Test handling of events for non-existent recordings."""
|
||||
|
||||
RecordingFactory(status="active")
|
||||
|
||||
mock_parser = mock.Mock()
|
||||
mock_parser.get_recording_id.return_value = uuid.uuid4()
|
||||
mock_get_parser.return_value = mock_parser
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert response.json() == {"detail": "No recording found for this event."}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status", ["failed_to_start", "aborted", "failed_to_stop", "saved", "initiated"]
|
||||
)
|
||||
def test_save_recording_non_savable_recording(
|
||||
recording_settings, mock_get_parser, client, status
|
||||
):
|
||||
"""Test that recordings in non-savable states cannot be saved."""
|
||||
|
||||
recording = RecordingFactory(status=status)
|
||||
|
||||
mock_parser = mock.Mock()
|
||||
mock_parser.get_recording_id.return_value = recording.id
|
||||
mock_get_parser.return_value = mock_parser
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {
|
||||
"detail": f"Recording with ID {recording.id} cannot be saved because it is either,"
|
||||
" in an error state or has already been saved."
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", ["active", "stopped"])
|
||||
def test_save_recording_success(recording_settings, mock_get_parser, client, status):
|
||||
"""Test successful saving of recordings in valid states."""
|
||||
|
||||
recording = RecordingFactory(status=status)
|
||||
|
||||
mock_parser = mock.Mock()
|
||||
mock_parser.get_recording_id.return_value = recording.id
|
||||
mock_get_parser.return_value = mock_parser
|
||||
|
||||
response = client.post(
|
||||
"/api/v1.0/recordings/storage-hook/",
|
||||
{"recording_data": "valid-data"},
|
||||
HTTP_AUTHORIZATION="Bearer testAuthToken",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"message": "Event processed."}
|
||||
|
||||
recording.refresh_from_db()
|
||||
assert recording.status == RecordingStatusChoices.SAVED
|
||||
@@ -1,171 +0,0 @@
|
||||
"""
|
||||
Test worker service factories.
|
||||
"""
|
||||
|
||||
# pylint: disable=W0212,W0621,W0613
|
||||
|
||||
from dataclasses import FrozenInstanceError
|
||||
from unittest.mock import Mock
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
|
||||
from core.recording.worker.factories import (
|
||||
WorkerService,
|
||||
WorkerServiceConfig,
|
||||
get_worker_service,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clear_lru_cache():
|
||||
"""Clear the lru_cache before and after each test"""
|
||||
WorkerServiceConfig.from_settings.cache_clear()
|
||||
yield
|
||||
WorkerServiceConfig.from_settings.cache_clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_settings():
|
||||
"""Fixture to provide test Django settings"""
|
||||
mocked_settings = {
|
||||
"RECORDING_OUTPUT_FOLDER": "/test/output",
|
||||
"LIVEKIT_CONFIGURATION": {"server": "test.example.com"},
|
||||
"RECORDING_VERIFY_SSL": True,
|
||||
"AWS_S3_ENDPOINT_URL": "https://s3.test.com",
|
||||
"AWS_S3_ACCESS_KEY_ID": "test_key",
|
||||
"AWS_S3_SECRET_ACCESS_KEY": "test_secret",
|
||||
"AWS_S3_REGION_NAME": "test-region",
|
||||
"AWS_STORAGE_BUCKET_NAME": "test-bucket",
|
||||
}
|
||||
|
||||
# Use override_settings to properly patch Django settings
|
||||
with override_settings(**mocked_settings):
|
||||
yield test_settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def default_config(test_settings):
|
||||
"""Fixture to provide a WorkerServiceConfig instance"""
|
||||
return WorkerServiceConfig.from_settings()
|
||||
|
||||
|
||||
# Tests
|
||||
def test_config_initialization(default_config):
|
||||
"""Test that WorkerServiceConfig is properly initialized from settings"""
|
||||
assert default_config.output_folder == "/test/output"
|
||||
assert default_config.server_configurations == {"server": "test.example.com"}
|
||||
assert default_config.verify_ssl is True
|
||||
assert default_config.bucket_args == {
|
||||
"endpoint": "https://s3.test.com",
|
||||
"access_key": "test_key",
|
||||
"secret": "test_secret",
|
||||
"region": "test-region",
|
||||
"bucket": "test-bucket",
|
||||
"force_path_style": True,
|
||||
}
|
||||
|
||||
|
||||
def test_config_immutability(default_config):
|
||||
"""Test that config instances are immutable after creation"""
|
||||
with pytest.raises(FrozenInstanceError):
|
||||
default_config.output_folder = "new/path"
|
||||
|
||||
|
||||
@override_settings(
|
||||
RECORDING_OUTPUT_FOLDER="/test/output",
|
||||
LIVEKIT_CONFIGURATION={"server": "test.example.com"},
|
||||
RECORDING_VERIFY_SSL=True,
|
||||
AWS_S3_ENDPOINT_URL="https://s3.test.com",
|
||||
AWS_S3_ACCESS_KEY_ID="test_key",
|
||||
AWS_S3_SECRET_ACCESS_KEY="test_secret",
|
||||
AWS_S3_REGION_NAME="test-region",
|
||||
AWS_STORAGE_BUCKET_NAME="test-bucket",
|
||||
)
|
||||
def test_config_caching():
|
||||
"""Test that from_settings method caches its result"""
|
||||
# Clear cache before testing caching behavior
|
||||
WorkerServiceConfig.from_settings.cache_clear()
|
||||
|
||||
config1 = WorkerServiceConfig.from_settings()
|
||||
config2 = WorkerServiceConfig.from_settings()
|
||||
assert config1 is config2
|
||||
|
||||
|
||||
class MockWorkerService(WorkerService):
|
||||
"""Mock worker service for testing."""
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_import_string(monkeypatch):
|
||||
"""Fixture to mock import_string function."""
|
||||
mock = Mock(return_value=MockWorkerService)
|
||||
monkeypatch.setattr("core.recording.worker.factories.import_string", mock)
|
||||
return mock
|
||||
|
||||
|
||||
def test_factory_valid_mode(mock_import_string, settings, default_config):
|
||||
"""Test getting worker service with valid mode."""
|
||||
|
||||
settings.RECORDING_WORKER_CLASSES = {
|
||||
"test_mode": "path.to.MockWorkerService",
|
||||
"another_mode": "path.to.AnotherWorkerService",
|
||||
}
|
||||
|
||||
worker = get_worker_service("test_mode")
|
||||
|
||||
mock_import_string.assert_called_once_with("path.to.MockWorkerService")
|
||||
assert isinstance(worker, MockWorkerService)
|
||||
assert worker.config == default_config
|
||||
|
||||
|
||||
def test_factory_invalid_mode(settings, mock_import_string, default_config):
|
||||
"""Test getting worker service with invalid mode raises ValueError."""
|
||||
|
||||
settings.RECORDING_WORKER_CLASSES = {
|
||||
"test_mode": "path.to.MockWorkerService",
|
||||
"another_mode": "path.to.AnotherWorkerService",
|
||||
}
|
||||
|
||||
worker = get_worker_service("test_mode")
|
||||
|
||||
mock_import_string.assert_called_once_with("path.to.MockWorkerService")
|
||||
assert isinstance(worker, MockWorkerService)
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
get_worker_service("invalid_mode")
|
||||
mock_import_string.assert_not_called()
|
||||
|
||||
assert "Recording mode 'invalid_mode' not found" in str(exc_info.value)
|
||||
assert "Available modes: ['test_mode', 'another_mode']" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_factory_import_error(mock_import_string, settings):
|
||||
"""Test handling of import errors."""
|
||||
|
||||
mock_import_string.side_effect = ImportError("Module not found")
|
||||
|
||||
settings.RECORDING_WORKER_CLASSES = {
|
||||
"test_mode": "path.to.MockWorkerService",
|
||||
"another_mode": "path.to.AnotherWorkerService",
|
||||
}
|
||||
|
||||
with pytest.raises(ImportError) as exc_info:
|
||||
get_worker_service("test_mode")
|
||||
|
||||
assert "Module not found" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_factory_empty_registry(settings):
|
||||
"""Test behavior when worker registry is empty."""
|
||||
|
||||
settings.RECORDING_WORKER_CLASSES = {}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
get_worker_service("any_mode")
|
||||
|
||||
assert "Available modes: []" in str(exc_info.value)
|
||||
@@ -1,161 +0,0 @@
|
||||
"""Test WorkerServiceMediator class."""
|
||||
|
||||
# pylint: disable=W0621,W0613
|
||||
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from core.factories import RecordingFactory
|
||||
from core.models import RecordingStatusChoices
|
||||
from core.recording.worker.exceptions import (
|
||||
RecordingStartError,
|
||||
RecordingStopError,
|
||||
WorkerConnectionError,
|
||||
WorkerRequestError,
|
||||
WorkerResponseError,
|
||||
)
|
||||
from core.recording.worker.factories import WorkerService
|
||||
from core.recording.worker.mediator import WorkerServiceMediator
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_service():
|
||||
"""Fixture for mock worker service"""
|
||||
return Mock(spec=WorkerService)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mediator(mock_worker_service):
|
||||
"""Fixture for WorkerServiceMediator"""
|
||||
return WorkerServiceMediator(mock_worker_service)
|
||||
|
||||
|
||||
def test_start_recording_success(mediator, mock_worker_service):
|
||||
"""Test successful recording start"""
|
||||
# Setup
|
||||
worker_id = "test-worker-123"
|
||||
mock_worker_service.start.return_value = worker_id
|
||||
|
||||
mock_recording = RecordingFactory(
|
||||
status=RecordingStatusChoices.INITIATED, worker_id=None
|
||||
)
|
||||
mediator.start(mock_recording)
|
||||
|
||||
# Verify worker service call
|
||||
expected_room_name = str(mock_recording.room.id)
|
||||
mock_worker_service.start.assert_called_once_with(
|
||||
expected_room_name, mock_recording.id
|
||||
)
|
||||
|
||||
# Verify recording updates
|
||||
mock_recording.refresh_from_db()
|
||||
assert mock_recording.worker_id == worker_id
|
||||
assert mock_recording.status == RecordingStatusChoices.ACTIVE
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error_class", [WorkerRequestError, WorkerConnectionError, WorkerResponseError]
|
||||
)
|
||||
def test_mediator_start_recording_worker_errors(
|
||||
mediator, mock_worker_service, error_class
|
||||
):
|
||||
"""Test handling of various worker errors during start"""
|
||||
# Setup
|
||||
mock_worker_service.start.side_effect = error_class("Test error")
|
||||
mock_recording = RecordingFactory(
|
||||
status=RecordingStatusChoices.INITIATED, worker_id=None
|
||||
)
|
||||
|
||||
# Execute and verify
|
||||
with pytest.raises(RecordingStartError):
|
||||
mediator.start(mock_recording)
|
||||
|
||||
# Verify recording updates
|
||||
mock_recording.refresh_from_db()
|
||||
assert mock_recording.status == RecordingStatusChoices.FAILED_TO_START
|
||||
assert mock_recording.worker_id is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status",
|
||||
[
|
||||
RecordingStatusChoices.ACTIVE,
|
||||
RecordingStatusChoices.FAILED_TO_START,
|
||||
RecordingStatusChoices.FAILED_TO_STOP,
|
||||
RecordingStatusChoices.STOPPED,
|
||||
RecordingStatusChoices.SAVED,
|
||||
RecordingStatusChoices.ABORTED,
|
||||
],
|
||||
)
|
||||
def test_mediator_start_recording_from_forbidden_status(
|
||||
mediator, mock_worker_service, status
|
||||
):
|
||||
"""Test handling of various worker errors during start"""
|
||||
# Setup
|
||||
mock_recording = RecordingFactory(status=status)
|
||||
|
||||
# Execute and verify
|
||||
with pytest.raises(RecordingStartError):
|
||||
mediator.start(mock_recording)
|
||||
|
||||
# Verify recording was not updated
|
||||
mock_recording.refresh_from_db()
|
||||
assert mock_recording.status == status
|
||||
|
||||
|
||||
def test_mediator_stop_recording_success(mediator, mock_worker_service):
|
||||
"""Test successful recording stop"""
|
||||
# Setup
|
||||
mock_recording = RecordingFactory(
|
||||
status=RecordingStatusChoices.ACTIVE, worker_id="test-worker-123"
|
||||
)
|
||||
mock_worker_service.stop.return_value = "STOPPED"
|
||||
|
||||
# Execute
|
||||
mediator.stop(mock_recording)
|
||||
|
||||
# Verify worker service call
|
||||
mock_worker_service.stop.assert_called_once_with(worker_id=mock_recording.worker_id)
|
||||
|
||||
# Verify recording updates
|
||||
mock_recording.refresh_from_db()
|
||||
assert mock_recording.status == RecordingStatusChoices.STOPPED
|
||||
|
||||
|
||||
def test_mediator_stop_recording_aborted(mediator, mock_worker_service):
|
||||
"""Test recording stop when worker returns ABORTED"""
|
||||
# Setup
|
||||
mock_recording = RecordingFactory(
|
||||
status=RecordingStatusChoices.ACTIVE, worker_id="test-worker-123"
|
||||
)
|
||||
mock_worker_service.stop.return_value = "ABORTED"
|
||||
|
||||
# Execute
|
||||
mediator.stop(mock_recording)
|
||||
|
||||
# Verify recording updates
|
||||
mock_recording.refresh_from_db()
|
||||
assert mock_recording.status == RecordingStatusChoices.ABORTED
|
||||
|
||||
|
||||
@pytest.mark.parametrize("error_class", [WorkerConnectionError, WorkerResponseError])
|
||||
def test_mediator_stop_recording_worker_errors(
|
||||
mediator, mock_worker_service, error_class
|
||||
):
|
||||
"""Test handling of worker errors during stop"""
|
||||
# Setup
|
||||
mock_recording = RecordingFactory(
|
||||
status=RecordingStatusChoices.ACTIVE, worker_id="test-worker-123"
|
||||
)
|
||||
mock_worker_service.stop.side_effect = error_class("Test error")
|
||||
|
||||
# Execute and verify
|
||||
with pytest.raises(RecordingStopError):
|
||||
mediator.stop(mock_recording)
|
||||
|
||||
# Verify recording updates
|
||||
mock_recording.refresh_from_db()
|
||||
assert mock_recording.status == RecordingStatusChoices.FAILED_TO_STOP
|
||||
@@ -1,368 +0,0 @@
|
||||
"""
|
||||
Test worker service classes.
|
||||
"""
|
||||
|
||||
# pylint: disable=W0212,W0621,W0613,E1101
|
||||
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import aiohttp
|
||||
import pytest
|
||||
|
||||
from core.recording.worker.exceptions import WorkerConnectionError, WorkerResponseError
|
||||
from core.recording.worker.factories import WorkerServiceConfig
|
||||
from core.recording.worker.services import (
|
||||
AudioCompositeEgressService,
|
||||
BaseEgressService,
|
||||
VideoCompositeEgressService,
|
||||
livekit_api,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config():
|
||||
"""Fixture to provide a WorkerServiceConfig instance"""
|
||||
return WorkerServiceConfig(
|
||||
output_folder="/test/output",
|
||||
server_configurations={
|
||||
"host": "test.livekit.io",
|
||||
"api_key": "test_key",
|
||||
"api_secret": "test_secret",
|
||||
},
|
||||
verify_ssl=True,
|
||||
bucket_args={
|
||||
"endpoint": "https://s3.test.com",
|
||||
"access_key": "test_key",
|
||||
"secret": "test_secret",
|
||||
"region": "test-region",
|
||||
"bucket": "test-bucket",
|
||||
"force_path_style": True,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_s3_upload():
|
||||
"""Fixture for mocked S3Upload"""
|
||||
with patch("core.recording.worker.services.livekit_api.S3Upload") as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_egress_service():
|
||||
"""Fixture for mocked EgressService"""
|
||||
with patch("core.recording.worker.services.EgressService") as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service(config, mock_s3_upload):
|
||||
"""Fixture for BaseEgressService instance"""
|
||||
return BaseEgressService(config)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_client_session():
|
||||
"""Fixture for mocked aiohttp.ClientSession"""
|
||||
with patch("aiohttp.ClientSession") as mock:
|
||||
mock.return_value.__aenter__ = AsyncMock()
|
||||
mock.return_value.__aexit__ = AsyncMock()
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_tcp_connector():
|
||||
"""Fixture for TCPConnector"""
|
||||
with patch("aiohttp.TCPConnector") as mock_connector:
|
||||
mock_connector_instance = Mock()
|
||||
mock_connector.return_value = mock_connector_instance
|
||||
yield mock_connector
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def video_service(config):
|
||||
"""Fixture for VideoCompositeEgressService"""
|
||||
service = VideoCompositeEgressService(config)
|
||||
service._handle_request = Mock() # Mock the request handler
|
||||
return service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def audio_service(config):
|
||||
"""Fixture for AudioCompositeEgressService"""
|
||||
service = AudioCompositeEgressService(config)
|
||||
service._handle_request = Mock() # Mock the request handler
|
||||
return service
|
||||
|
||||
|
||||
def test_base_egress_initialization(config, mock_s3_upload):
|
||||
"""Test service initialization"""
|
||||
|
||||
service = BaseEgressService(config)
|
||||
assert service._config == config
|
||||
|
||||
mock_s3_upload.assert_called_once_with(
|
||||
endpoint="https://s3.test.com",
|
||||
access_key="test_key",
|
||||
secret="test_secret",
|
||||
region="test-region",
|
||||
bucket="test-bucket",
|
||||
force_path_style=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"filename,extension,expected",
|
||||
[
|
||||
("test", "mp4", "/test/output/test.mp4"),
|
||||
("recording123", "ogg", "/test/output/recording123.ogg"),
|
||||
("live_stream", "m3u8", "/test/output/live_stream.m3u8"),
|
||||
],
|
||||
)
|
||||
def test_base_egress_filepath_construction(service, filename, extension, expected):
|
||||
"""Test filepath construction with various inputs"""
|
||||
result = service._get_filepath(filename, extension)
|
||||
assert result == expected
|
||||
assert result.startswith(service._config.output_folder)
|
||||
assert result.endswith(f"{filename}.{extension}")
|
||||
|
||||
|
||||
def test_base_egress_handle_request_success(
|
||||
config, service, mock_client_session, mock_egress_service, mock_tcp_connector
|
||||
):
|
||||
"""Test successful request handling"""
|
||||
# Setup mock response
|
||||
mock_response = Mock()
|
||||
mock_method = AsyncMock(return_value=mock_response)
|
||||
mock_egress_instance = Mock()
|
||||
mock_egress_instance.test_method = mock_method
|
||||
mock_egress_service.return_value = mock_egress_instance
|
||||
|
||||
# Create test request
|
||||
test_request = Mock()
|
||||
|
||||
response = service._handle_request(test_request, "test_method")
|
||||
|
||||
mock_client_session.assert_called_once_with(
|
||||
connector=mock_tcp_connector.return_value
|
||||
)
|
||||
|
||||
# Verify EgressService initialization
|
||||
mock_egress_service.assert_called_once_with(
|
||||
mock_client_session.return_value.__aenter__.return_value,
|
||||
**service._config.server_configurations,
|
||||
)
|
||||
|
||||
# Verify method call and response
|
||||
mock_method.assert_called_once_with(test_request)
|
||||
assert response == mock_response
|
||||
|
||||
|
||||
def test_base_egress_handle_request_connection_error(service, mock_egress_service):
|
||||
"""Test handling of connection errors"""
|
||||
# Setup mock error
|
||||
mock_method = AsyncMock(
|
||||
side_effect=livekit_api.TwirpError(msg="Connection failed", code=500)
|
||||
)
|
||||
mock_egress_instance = Mock()
|
||||
mock_egress_instance.test_method = mock_method
|
||||
mock_egress_service.return_value = mock_egress_instance
|
||||
|
||||
# Create test request
|
||||
test_request = Mock()
|
||||
|
||||
# Verify error handling
|
||||
with pytest.raises(WorkerConnectionError) as exc:
|
||||
service._handle_request(test_request, "test_method")
|
||||
|
||||
assert "LiveKit client connection error" in str(exc.value)
|
||||
assert "Connection failed" in str(exc.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"response_status,expected_result",
|
||||
[
|
||||
(livekit_api.EgressStatus.EGRESS_ABORTED, "ABORTED"),
|
||||
(livekit_api.EgressStatus.EGRESS_COMPLETE, "FAILED_TO_STOP"),
|
||||
(livekit_api.EgressStatus.EGRESS_ENDING, "STOPPED"),
|
||||
(livekit_api.EgressStatus.EGRESS_FAILED, "FAILED_TO_STOP"),
|
||||
],
|
||||
)
|
||||
def test_base_egress_stop_with_status(service, response_status, expected_result):
|
||||
"""Test stop method with different response statuses"""
|
||||
# Mock _handle_request
|
||||
mock_response = Mock(status=response_status)
|
||||
service._handle_request = Mock(return_value=mock_response)
|
||||
|
||||
# Execute stop
|
||||
result = service.stop("test_worker_id")
|
||||
|
||||
# Verify request and response handling
|
||||
service._handle_request.assert_called_once_with(
|
||||
livekit_api.StopEgressRequest(egress_id="test_worker_id"), "stop_egress"
|
||||
)
|
||||
assert result == expected_result
|
||||
|
||||
|
||||
def test_base_egress_stop_missing_status(service):
|
||||
"""Test stop method when response is missing status"""
|
||||
# Mock _handle_request with missing status
|
||||
mock_response = Mock(status=None)
|
||||
service._handle_request = Mock(return_value=mock_response)
|
||||
|
||||
# Verify error handling
|
||||
with pytest.raises(WorkerResponseError) as exc:
|
||||
service.stop("test_worker_id")
|
||||
|
||||
assert "missing the recording status" in str(exc.value)
|
||||
|
||||
|
||||
def test_base_egress_start_not_implemented(service):
|
||||
"""Test that start method raises NotImplementedError"""
|
||||
with pytest.raises(NotImplementedError) as exc:
|
||||
service.start("test_room", "test_recording")
|
||||
assert "Subclass must implement this method" in str(exc.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("verify_ssl", [True, False])
|
||||
def test_base_egress_ssl_verification_config(verify_ssl):
|
||||
"""Test SSL verification configuration"""
|
||||
config = WorkerServiceConfig(
|
||||
output_folder="/test/output",
|
||||
server_configurations={
|
||||
"host": "test.livekit.io",
|
||||
"api_key": "test_key",
|
||||
"api_secret": "test_secret",
|
||||
},
|
||||
verify_ssl=verify_ssl,
|
||||
bucket_args={
|
||||
"endpoint": "https://s3.test.com",
|
||||
"access_key": "test_key",
|
||||
"secret": "test_secret",
|
||||
"region": "test-region",
|
||||
"bucket": "test-bucket",
|
||||
"force_path_style": True,
|
||||
},
|
||||
)
|
||||
|
||||
service = BaseEgressService(config)
|
||||
|
||||
# Mock ClientSession to capture connector configuration
|
||||
with patch("aiohttp.ClientSession") as mock_session:
|
||||
mock_session.return_value.__aenter__ = AsyncMock()
|
||||
mock_session.return_value.__aexit__ = AsyncMock()
|
||||
|
||||
# Trigger request to verify connector configuration
|
||||
service._handle_request(Mock(), "test_method")
|
||||
|
||||
# Verify SSL configuration
|
||||
connector = mock_session.call_args[1]["connector"]
|
||||
assert isinstance(connector, aiohttp.TCPConnector)
|
||||
assert connector._ssl == verify_ssl
|
||||
|
||||
|
||||
def test_video_composite_egress_hrid(video_service):
|
||||
"""Test HRID is correct"""
|
||||
assert video_service.hrid == "video-recording-composite-livekit-egress"
|
||||
|
||||
|
||||
def test_video_composite_egress_start_success(video_service):
|
||||
"""Test successful start of video composite egress"""
|
||||
# Setup mock response
|
||||
egress_id = "test-egress-123"
|
||||
video_service._handle_request.return_value = Mock(egress_id=egress_id)
|
||||
|
||||
# Test parameters
|
||||
room_name = "test-room"
|
||||
recording_id = "rec-123"
|
||||
|
||||
# Call start
|
||||
result = video_service.start(room_name, recording_id)
|
||||
|
||||
# Verify result
|
||||
assert result == egress_id
|
||||
|
||||
# Verify request construction
|
||||
video_service._handle_request.assert_called_once()
|
||||
request = video_service._handle_request.call_args[0][0]
|
||||
method = video_service._handle_request.call_args[0][1]
|
||||
|
||||
# Verify request properties
|
||||
assert isinstance(request, livekit_api.RoomCompositeEgressRequest)
|
||||
assert request.room_name == room_name
|
||||
assert len(request.file_outputs) == 1
|
||||
|
||||
assert not request.audio_only # Video service shouldn't set audio_only
|
||||
|
||||
# Verify file output configuration
|
||||
file_output = request.file_outputs[0]
|
||||
assert file_output.file_type == livekit_api.EncodedFileType.MP4
|
||||
assert file_output.filepath == f"/test/output/{recording_id}.mp4"
|
||||
assert file_output.s3.bucket == "test-bucket"
|
||||
|
||||
# Verify method name
|
||||
assert method == "start_room_composite_egress"
|
||||
|
||||
|
||||
def test_video_composite_egress_start_missing_egress_id(video_service):
|
||||
"""Test handling of missing egress ID in response"""
|
||||
# Setup mock response without egress_id
|
||||
video_service._handle_request.return_value = Mock(egress_id=None)
|
||||
|
||||
with pytest.raises(WorkerResponseError) as exc_info:
|
||||
video_service.start("test-room", "rec-123")
|
||||
|
||||
assert "Egress ID not found" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_audio_composite_egress_hrid(audio_service):
|
||||
"""Test HRID is correct"""
|
||||
assert audio_service.hrid == "audio-recording-composite-livekit-egress"
|
||||
|
||||
|
||||
def test_audio_composite_egress_start_success(audio_service):
|
||||
"""Test successful start of audio composite egress"""
|
||||
# Setup mock response
|
||||
egress_id = "test-egress-123"
|
||||
audio_service._handle_request.return_value = Mock(egress_id=egress_id)
|
||||
|
||||
# Test parameters
|
||||
room_name = "test-room"
|
||||
recording_id = "rec-123"
|
||||
|
||||
# Call start
|
||||
result = audio_service.start(room_name, recording_id)
|
||||
|
||||
# Verify result
|
||||
assert result == egress_id
|
||||
|
||||
# Verify request construction
|
||||
audio_service._handle_request.assert_called_once()
|
||||
request = audio_service._handle_request.call_args[0][0]
|
||||
method = audio_service._handle_request.call_args[0][1]
|
||||
|
||||
# Verify request properties
|
||||
assert isinstance(request, livekit_api.RoomCompositeEgressRequest)
|
||||
assert request.room_name == room_name
|
||||
assert len(request.file_outputs) == 1
|
||||
assert request.audio_only is True # Audio service should set audio_only
|
||||
|
||||
# Verify file output configuration
|
||||
file_output = request.file_outputs[0]
|
||||
assert file_output.file_type == livekit_api.EncodedFileType.OGG
|
||||
assert file_output.filepath == f"/test/output/{recording_id}.ogg"
|
||||
assert file_output.s3.bucket == "test-bucket"
|
||||
|
||||
# Verify method name
|
||||
assert method == "start_room_composite_egress"
|
||||
|
||||
|
||||
def test_audio_composite_egress_start_missing_egress_id(audio_service):
|
||||
"""Test handling of missing egress ID in response"""
|
||||
# Setup mock response without egress_id
|
||||
audio_service._handle_request.return_value = Mock(egress_id=None)
|
||||
|
||||
with pytest.raises(WorkerResponseError) as exc_info:
|
||||
audio_service.start("test-room", "rec-123")
|
||||
|
||||
assert "Egress ID not found" in str(exc_info.value)
|
||||
@@ -38,7 +38,7 @@ def test_api_rooms_retrieve_anonymous_private_pk():
|
||||
def test_api_rooms_retrieve_anonymous_private_pk_no_dashes():
|
||||
"""It should be possible to get a room by its id stripped of its dashes."""
|
||||
room = RoomFactory(is_public=False)
|
||||
id_no_dashes = str(room.id)
|
||||
id_no_dashes = str(room.id).replace("-", "")
|
||||
|
||||
client = APIClient()
|
||||
response = client.get(f"/api/v1.0/rooms/{id_no_dashes:s}/")
|
||||
@@ -178,7 +178,7 @@ def test_api_rooms_retrieve_anonymous_public(mock_token):
|
||||
response = client.get(f"/api/v1.0/rooms/{room.id!s}/")
|
||||
|
||||
assert response.status_code == 200
|
||||
expected_name = f"{room.id!s}"
|
||||
expected_name = f"{room.id!s}".replace("-", "")
|
||||
assert response.json() == {
|
||||
"id": str(room.id),
|
||||
"is_administrable": False,
|
||||
@@ -220,7 +220,7 @@ def test_api_rooms_retrieve_authenticated_public(mock_token):
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
expected_name = f"{room.id!s}"
|
||||
expected_name = f"{room.id!s}".replace("-", "")
|
||||
assert response.json() == {
|
||||
"id": str(room.id),
|
||||
"is_administrable": False,
|
||||
@@ -318,7 +318,7 @@ def test_api_rooms_retrieve_members(mock_token, django_assert_num_queries):
|
||||
key=lambda x: x["id"],
|
||||
)
|
||||
|
||||
expected_name = str(room.id)
|
||||
expected_name = str(room.id).replace("-", "")
|
||||
assert content_dict == {
|
||||
"id": str(room.id),
|
||||
"is_administrable": False,
|
||||
@@ -390,7 +390,7 @@ def test_api_rooms_retrieve_administrators(mock_token, django_assert_num_queries
|
||||
],
|
||||
key=lambda x: x["id"],
|
||||
)
|
||||
expected_name = str(room.id)
|
||||
expected_name = str(room.id).replace("-", "")
|
||||
assert content_dict == {
|
||||
"id": str(room.id),
|
||||
"is_administrable": True,
|
||||
|
||||
@@ -1,200 +0,0 @@
|
||||
"""
|
||||
Test rooms API endpoints in the Meet core app: start recording.
|
||||
"""
|
||||
|
||||
# pylint: disable=W0621,W0613
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from ...factories import RoomFactory, UserFactory
|
||||
from ...models import Recording
|
||||
from ...recording.worker.exceptions import RecordingStartError
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_service():
|
||||
"""Mock worker service."""
|
||||
return mock.Mock()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_service_factory(mock_worker_service):
|
||||
"""Mock worker service factory."""
|
||||
with mock.patch(
|
||||
"core.api.viewsets.get_worker_service",
|
||||
return_value=mock_worker_service,
|
||||
) as mock_worker_service_factory:
|
||||
yield mock_worker_service_factory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_manager(mock_worker_service):
|
||||
"""Mock worker service mediator."""
|
||||
with mock.patch("core.api.viewsets.WorkerServiceMediator") as mock_mediator_class:
|
||||
mock_mediator = mock.Mock()
|
||||
mock_mediator_class.return_value = mock_mediator
|
||||
yield mock_mediator
|
||||
|
||||
|
||||
def test_start_recording_anonymous():
|
||||
"""Anonymous users should not be allowed to start room recordings."""
|
||||
room = RoomFactory()
|
||||
client = APIClient()
|
||||
|
||||
response = client.post(
|
||||
f"/api/v1.0/rooms/{room.id}/start-recording/",
|
||||
{"mode": "screen_recording"},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert Recording.objects.count() == 0
|
||||
|
||||
|
||||
def test_start_recording_non_owner_and_non_administrator():
|
||||
"""Non-owner and Non-Administrator users should not be allowed to start room recordings."""
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(
|
||||
f"/api/v1.0/rooms/{room.id}/start-recording/",
|
||||
{"mode": "screen_recording"},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert Recording.objects.count() == 0
|
||||
|
||||
|
||||
def test_start_recording_recording_disabled(settings):
|
||||
"""Should fail if recording is disabled for the room."""
|
||||
settings.RECORDING_ENABLE = False
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(
|
||||
f"/api/v1.0/rooms/{room.id}/start-recording/",
|
||||
{"mode": "screen_recording"},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "Access denied, recording is disabled."}
|
||||
assert Recording.objects.count() == 0
|
||||
|
||||
|
||||
def test_start_recording_missing_mode(settings):
|
||||
"""Should fail if recording mode is not provided."""
|
||||
settings.RECORDING_ENABLE = True
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/start-recording/", {})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Invalid request."}
|
||||
assert Recording.objects.count() == 0
|
||||
|
||||
|
||||
def test_start_recording_worker_error(
|
||||
mock_worker_service_factory, mock_worker_manager, settings
|
||||
):
|
||||
"""Should handle worker service errors appropriately."""
|
||||
settings.RECORDING_ENABLE = True
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
# Configure mock mediator to raise error
|
||||
mock_start = mock.Mock()
|
||||
mock_start.side_effect = RecordingStartError("Failed to connect to worker")
|
||||
|
||||
mock_worker_manager.start = mock_start
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(
|
||||
f"/api/v1.0/rooms/{room.id}/start-recording/",
|
||||
{"mode": "screen_recording"},
|
||||
)
|
||||
|
||||
mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
|
||||
|
||||
assert response.status_code == 500
|
||||
assert response.json() == {
|
||||
"error": f"Recording failed to start for room {room.slug}"
|
||||
}
|
||||
|
||||
# Recording object should be created even if worker fails
|
||||
assert Recording.objects.count() == 1
|
||||
recording = Recording.objects.first()
|
||||
assert recording.room == room
|
||||
assert recording.mode == "screen_recording"
|
||||
|
||||
# Verify recording access details
|
||||
assert recording.accesses.count() == 1
|
||||
access = recording.accesses.first()
|
||||
assert access.user == user
|
||||
assert access.role == "owner"
|
||||
|
||||
|
||||
def test_start_recording_success(
|
||||
mock_worker_service_factory, mock_worker_manager, settings
|
||||
):
|
||||
"""Should successfully start recording when everything is configured correctly."""
|
||||
settings.RECORDING_ENABLE = True
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
mock_start = mock.Mock()
|
||||
mock_worker_manager.start = mock_start
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(
|
||||
f"/api/v1.0/rooms/{room.id}/start-recording/",
|
||||
{"mode": "screen_recording"},
|
||||
)
|
||||
|
||||
mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.json() == {
|
||||
"message": f"Recording successfully started for room {room.slug}"
|
||||
}
|
||||
|
||||
# Verify the mediator was called with the recording
|
||||
recording = Recording.objects.first()
|
||||
mock_start.assert_called_once_with(recording)
|
||||
|
||||
assert recording.room == room
|
||||
assert recording.mode == "screen_recording"
|
||||
|
||||
# Verify recording access details
|
||||
assert recording.accesses.count() == 1
|
||||
access = recording.accesses.first()
|
||||
assert access.user == user
|
||||
assert access.role == "owner"
|
||||
@@ -1,182 +0,0 @@
|
||||
"""
|
||||
Test rooms API endpoints in the Meet core app: stop recording.
|
||||
"""
|
||||
|
||||
# pylint: disable=W0621,W0613
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from ...factories import RecordingFactory, RoomFactory, UserFactory
|
||||
from ...models import Recording, RecordingStatusChoices
|
||||
from ...recording.worker.exceptions import RecordingStopError
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_service():
|
||||
"""Mock worker service."""
|
||||
return mock.Mock()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_service_factory(mock_worker_service):
|
||||
"""Mock worker service factory."""
|
||||
with mock.patch(
|
||||
"core.api.viewsets.get_worker_service",
|
||||
return_value=mock_worker_service,
|
||||
) as mock_worker_service_factory:
|
||||
yield mock_worker_service_factory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_worker_manager(mock_worker_service):
|
||||
"""Mock worker service mediator."""
|
||||
with mock.patch("core.api.viewsets.WorkerServiceMediator") as mock_mediator_class:
|
||||
mock_mediator = mock.Mock()
|
||||
mock_mediator_class.return_value = mock_mediator
|
||||
yield mock_mediator
|
||||
|
||||
|
||||
def test_stop_recording_anonymous():
|
||||
"""Anonymous users should not be allowed to stop room recordings."""
|
||||
room = RoomFactory()
|
||||
RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
|
||||
client = APIClient()
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
|
||||
|
||||
assert response.status_code == 401
|
||||
# Verify recording status hasn't changed
|
||||
assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
|
||||
|
||||
|
||||
def test_stop_recording_non_owner_and_non_administrator():
|
||||
"""Non-owner and Non-Administrator users should not be allowed to stop room recordings."""
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
|
||||
|
||||
assert response.status_code == 403
|
||||
# Verify recording status hasn't changed
|
||||
assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
|
||||
|
||||
|
||||
def test_stop_recording_recording_disabled(settings):
|
||||
"""Should fail if recording is disabled for the room."""
|
||||
|
||||
settings.RECORDING_ENABLE = False
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "Access denied, recording is disabled."}
|
||||
# Verify no recording exists
|
||||
assert Recording.objects.count() == 0
|
||||
|
||||
|
||||
def test_stop_recording_no_active_recording(settings):
|
||||
"""Should fail when there is no active recording for the room."""
|
||||
|
||||
settings.RECORDING_ENABLE = True
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
|
||||
|
||||
assert response.status_code == 404
|
||||
assert response.json() == {"detail": "No active recording found for this room."}
|
||||
|
||||
|
||||
def test_stop_recording_worker_error(
|
||||
mock_worker_service_factory, mock_worker_manager, settings
|
||||
):
|
||||
"""Should handle worker service errors appropriately."""
|
||||
|
||||
settings.RECORDING_ENABLE = True
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
recording = RecordingFactory(
|
||||
room=room,
|
||||
status=RecordingStatusChoices.ACTIVE,
|
||||
mode="screen_recording",
|
||||
)
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
# Configure mock mediator to raise error
|
||||
mock_stop = mock.Mock()
|
||||
mock_stop.side_effect = RecordingStopError("Failed to connect to worker")
|
||||
mock_worker_manager.stop = mock_stop
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
|
||||
|
||||
mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
|
||||
mock_stop.assert_called_once_with(recording)
|
||||
|
||||
assert response.status_code == 500
|
||||
assert response.json() == {
|
||||
"error": f"Recording failed to stop for room {room.slug}"
|
||||
}
|
||||
# Verify recording status hasn't changed
|
||||
assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
|
||||
|
||||
|
||||
def test_stop_recording_success(
|
||||
mock_worker_service_factory, mock_worker_manager, settings
|
||||
):
|
||||
"""Should successfully stop recording when everything is configured correctly."""
|
||||
|
||||
settings.RECORDING_ENABLE = True
|
||||
|
||||
room = RoomFactory()
|
||||
user = UserFactory()
|
||||
recording = RecordingFactory(
|
||||
room=room,
|
||||
status=RecordingStatusChoices.ACTIVE,
|
||||
mode="screen_recording",
|
||||
)
|
||||
# Make user the room owner
|
||||
room.accesses.create(user=user, role="owner")
|
||||
|
||||
mock_stop = mock.Mock()
|
||||
mock_worker_manager.stop = mock_stop
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
|
||||
|
||||
mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
|
||||
mock_stop.assert_called_once_with(recording)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"message": f"Recording stopped for room {room.slug}."}
|
||||
|
||||
# Verify the recording still exists
|
||||
assert Recording.objects.count() == 1
|
||||
@@ -173,12 +173,6 @@ def test_models_recording_is_savable_already_saved():
|
||||
assert recording.is_savable() is False
|
||||
|
||||
|
||||
def test_models_recording_is_savable_only_initiated():
|
||||
"""Test is_savable for only initiated recording."""
|
||||
recording = RecordingFactory(status=RecordingStatusChoices.INITIATED)
|
||||
assert recording.is_savable() is False
|
||||
|
||||
|
||||
# Test few corner cases
|
||||
|
||||
|
||||
|
||||
@@ -327,10 +327,6 @@ class Base(Configuration):
|
||||
default=True,
|
||||
environ_name="OIDC_CREATE_USER",
|
||||
)
|
||||
OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = values.BooleanValue(
|
||||
default=False,
|
||||
environ_name="OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION",
|
||||
)
|
||||
OIDC_RP_SIGN_ALGO = values.Value(
|
||||
"RS256", environ_name="OIDC_RP_SIGN_ALGO", environ_prefix=None
|
||||
)
|
||||
@@ -406,39 +402,6 @@ class Base(Configuration):
|
||||
True, environ_name="ALLOW_UNREGISTERED_ROOMS", environ_prefix=None
|
||||
)
|
||||
|
||||
# Recording settings
|
||||
RECORDING_ENABLE = values.BooleanValue(
|
||||
False, environ_name="RECORDING_ENABLE", environ_prefix=None
|
||||
)
|
||||
RECORDING_OUTPUT_FOLDER = values.Value(
|
||||
"recordings", environ_name="RECORDING_OUTPUT_FOLDER", environ_prefix=None
|
||||
)
|
||||
RECORDING_VERIFY_SSL = values.BooleanValue(
|
||||
True, environ_name="RECORDING_VERIFY_SSL", environ_prefix=None
|
||||
)
|
||||
RECORDING_WORKER_CLASSES = values.DictValue(
|
||||
{
|
||||
"screen_recording": "core.recording.worker.services.VideoCompositeEgressService",
|
||||
"transcript": "core.recording.worker.services.AudioCompositeEgressService",
|
||||
},
|
||||
environ_name="RECORDING_WORKER_CLASSES",
|
||||
environ_prefix=None,
|
||||
)
|
||||
RECORDING_EVENT_PARSER_CLASS = values.Value(
|
||||
"core.recording.event.parsers.MinioParser",
|
||||
environ_name="RECORDING_EVENT_PARSER_CLASS",
|
||||
environ_prefix=None,
|
||||
)
|
||||
RECORDING_ENABLE_STORAGE_EVENT_AUTH = values.BooleanValue(
|
||||
True, environ_name="RECORDING_ENABLE_STORAGE_EVENT_AUTH", environ_prefix=None
|
||||
)
|
||||
RECORDING_STORAGE_EVENT_ENABLE = values.BooleanValue(
|
||||
False, environ_name="RECORDING_STORAGE_EVENT_ENABLE", environ_prefix=None
|
||||
)
|
||||
RECORDING_STORAGE_EVENT_TOKEN = values.Value(
|
||||
None, environ_name="RECORDING_STORAGE_HOOK_TOKEN", environ_prefix=None
|
||||
)
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
@property
|
||||
def ENVIRONMENT(self):
|
||||
|
||||
@@ -57,7 +57,6 @@ dependencies = [
|
||||
"whitenoise==6.7.0",
|
||||
"mozilla-django-oidc==4.0.1",
|
||||
"livekit-api==0.7.0",
|
||||
"aiohttp==3.10.10",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
|
||||
@@ -2,12 +2,6 @@ import { fetchApi } from './fetchApi'
|
||||
import { keys } from './queryKeys'
|
||||
import { useQuery } from '@tanstack/react-query'
|
||||
|
||||
// todo - refactor it in a proper place
|
||||
export enum RecordingMode {
|
||||
Transcript = 'transcript',
|
||||
ScreenRecording = 'screen_recording',
|
||||
}
|
||||
|
||||
export interface ApiConfig {
|
||||
analytics?: {
|
||||
id: string
|
||||
@@ -17,10 +11,6 @@ export interface ApiConfig {
|
||||
id: string
|
||||
}
|
||||
silence_livekit_debug_logs?: boolean
|
||||
recording?: {
|
||||
is_enabled?: boolean
|
||||
available_modes?: RecordingMode[]
|
||||
}
|
||||
}
|
||||
|
||||
const fetchConfig = (): Promise<ApiConfig> => {
|
||||
|
||||
@@ -89,13 +89,6 @@ export const Conference = ({
|
||||
)
|
||||
}
|
||||
|
||||
// Some clients (like DINUM) operate in bandwidth-constrained environments
|
||||
// These settings help ensure successful connections in poor network conditions
|
||||
const connectOptions = {
|
||||
maxRetries: 5, // Default: 1. Only for unreachable server scenarios
|
||||
peerConnectionTimeout: 60, // Default: 15s. Extended for slow TURN/TLS negotiation
|
||||
}
|
||||
|
||||
return (
|
||||
<QueryAware status={isFetchError ? createStatus : fetchStatus}>
|
||||
<Screen header={false}>
|
||||
@@ -106,7 +99,6 @@ export const Conference = ({
|
||||
connect={true}
|
||||
audio={userConfig.audioEnabled}
|
||||
video={userConfig.videoEnabled}
|
||||
connectOptions={connectOptions}
|
||||
>
|
||||
<VideoConference />
|
||||
{showInviteDialog && (
|
||||
|
||||
Reference in New Issue
Block a user