mirror of
https://github.com/Monadical-SAS/reflector.git
synced 2026-02-07 03:06:46 +00:00
fix: prevent presence race condition during WebRTC handshake
Add /joining, /joined, and /leave endpoints to track user join intent and trigger presence updates. Backend: - Add pending_joins Redis module with 30s TTL - Add /joining endpoint (before WebRTC handshake) - Add /joined endpoint (after connection, triggers presence poll) - Add /leave endpoint (on tab close, triggers presence poll) - Check for pending joins before deactivating meetings in worker Frontend: - Generate unique connectionId per browser tab - Call /joining before Daily.co join, /joined after connection - Add beforeunload handler calling /leave via sendBeacon
This commit is contained in:
Igor Loskutov
367
server/tests/test_joining_endpoint.py
Normal file
367
server/tests/test_joining_endpoint.py
Normal file
@@ -0,0 +1,367 @@
|
||||
"""Integration tests for /joining and /joined endpoints.
|
||||
|
||||
Tests for the join intent tracking to prevent race conditions during
|
||||
WebRTC handshake when users join meetings.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from reflector.db.meetings import Meeting
|
||||
from reflector.presence.pending_joins import PENDING_JOIN_PREFIX
|
||||
|
||||
TEST_CONNECTION_ID = "test-connection-uuid-12345"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_room():
|
||||
"""Mock room object."""
|
||||
from reflector.db.rooms import Room
|
||||
|
||||
return Room(
|
||||
id="room-123",
|
||||
name="test-room",
|
||||
user_id="owner-user",
|
||||
created_at=datetime.now(timezone.utc),
|
||||
zulip_auto_post=False,
|
||||
zulip_stream="",
|
||||
zulip_topic="",
|
||||
is_locked=False,
|
||||
room_mode="normal",
|
||||
recording_type="cloud",
|
||||
recording_trigger="automatic",
|
||||
is_shared=True,
|
||||
platform="daily",
|
||||
skip_consent=False,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_meeting():
|
||||
"""Mock meeting object."""
|
||||
now = datetime.now(timezone.utc)
|
||||
return Meeting(
|
||||
id="meeting-456",
|
||||
room_id="room-123",
|
||||
room_name="test-room-20251118120000",
|
||||
room_url="https://daily.co/test-room-20251118120000",
|
||||
host_room_url="https://daily.co/test-room-20251118120000?t=host",
|
||||
platform="daily",
|
||||
num_clients=0,
|
||||
is_active=True,
|
||||
start_date=now,
|
||||
end_date=now + timedelta(hours=1),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
@patch("reflector.views.rooms.get_async_redis_client")
|
||||
async def test_joining_endpoint_creates_pending_join(
|
||||
mock_get_redis,
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that /joining endpoint creates pending join in Redis."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = mock_meeting
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.setex = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
response = await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
# Verify Redis setex was called with correct key pattern
|
||||
mock_redis.setex.assert_called_once()
|
||||
call_args = mock_redis.setex.call_args[0]
|
||||
assert call_args[0].startswith(f"{PENDING_JOIN_PREFIX}:{mock_meeting.id}:")
|
||||
assert TEST_CONNECTION_ID in call_args[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.poll_daily_room_presence_task")
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
@patch("reflector.views.rooms.get_async_redis_client")
|
||||
async def test_joined_endpoint_deletes_pending_join(
|
||||
mock_get_redis,
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_poll_task,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that /joined endpoint deletes pending join from Redis."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = mock_meeting
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.delete = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
response = await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joined",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
# Verify Redis delete was called with correct key pattern
|
||||
mock_redis.delete.assert_called_once()
|
||||
call_args = mock_redis.delete.call_args[0]
|
||||
assert call_args[0].startswith(f"{PENDING_JOIN_PREFIX}:{mock_meeting.id}:")
|
||||
assert TEST_CONNECTION_ID in call_args[0]
|
||||
|
||||
# Verify presence poll was triggered for Daily meetings
|
||||
mock_poll_task.delay.assert_called_once_with(mock_meeting.id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
async def test_joining_endpoint_room_not_found(
|
||||
mock_get_room,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that /joining returns 404 when room not found."""
|
||||
mock_get_room.return_value = None
|
||||
|
||||
response = await client.post(
|
||||
"/rooms/nonexistent-room/meetings/meeting-123/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert response.json()["detail"] == "Room not found"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
async def test_joining_endpoint_meeting_not_found(
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that /joining returns 404 when meeting not found."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = None
|
||||
|
||||
response = await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/nonexistent-meeting/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert response.json()["detail"] == "Meeting not found"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
async def test_joining_endpoint_meeting_not_active(
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that /joining returns 400 when meeting is not active."""
|
||||
mock_get_room.return_value = mock_room
|
||||
inactive_meeting = mock_meeting.model_copy(update={"is_active": False})
|
||||
mock_get_meeting.return_value = inactive_meeting
|
||||
|
||||
response = await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["detail"] == "Meeting is not active"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
@patch("reflector.views.rooms.get_async_redis_client")
|
||||
async def test_joining_endpoint_anonymous_user(
|
||||
mock_get_redis,
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
):
|
||||
"""Test that /joining works for anonymous users with unique connection_id."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = mock_meeting
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.setex = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
response = await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
# Verify Redis setex was called with "anon:" prefix and connection_id
|
||||
call_args = mock_redis.setex.call_args[0]
|
||||
assert ":anon:" in call_args[0]
|
||||
assert TEST_CONNECTION_ID in call_args[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
@patch("reflector.views.rooms.get_async_redis_client")
|
||||
async def test_joining_endpoint_redis_closed_on_success(
|
||||
mock_get_redis,
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that Redis connection is closed after successful operation."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = mock_meeting
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.setex = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
mock_redis.aclose.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
@patch("reflector.views.rooms.get_async_redis_client")
|
||||
async def test_joining_endpoint_redis_closed_on_error(
|
||||
mock_get_redis,
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
authenticated_client,
|
||||
):
|
||||
"""Test that Redis connection is closed even when operation fails."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = mock_meeting
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.setex = AsyncMock(side_effect=Exception("Redis error"))
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
with pytest.raises(Exception):
|
||||
await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": TEST_CONNECTION_ID},
|
||||
)
|
||||
|
||||
mock_redis.aclose.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_joining_endpoint_requires_connection_id(
|
||||
client,
|
||||
):
|
||||
"""Test that /joining returns 422 when connection_id is missing."""
|
||||
response = await client.post(
|
||||
"/rooms/test-room/meetings/meeting-123/joining",
|
||||
json={},
|
||||
)
|
||||
|
||||
assert response.status_code == 422 # Validation error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_joining_endpoint_rejects_empty_connection_id(
|
||||
client,
|
||||
):
|
||||
"""Test that /joining returns 422 when connection_id is empty string."""
|
||||
response = await client.post(
|
||||
"/rooms/test-room/meetings/meeting-123/joining",
|
||||
json={"connection_id": ""},
|
||||
)
|
||||
|
||||
assert response.status_code == 422 # Validation error (NonEmptyString)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.views.rooms.rooms_controller.get_by_name")
|
||||
@patch("reflector.views.rooms.meetings_controller.get_by_id")
|
||||
@patch("reflector.views.rooms.get_async_redis_client")
|
||||
async def test_different_connection_ids_create_different_keys(
|
||||
mock_get_redis,
|
||||
mock_get_meeting,
|
||||
mock_get_room,
|
||||
mock_room,
|
||||
mock_meeting,
|
||||
client,
|
||||
):
|
||||
"""Test that different connection_ids create different Redis keys."""
|
||||
mock_get_room.return_value = mock_room
|
||||
mock_get_meeting.return_value = mock_meeting
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.setex = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
# First connection
|
||||
await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": "connection-1"},
|
||||
)
|
||||
key1 = mock_redis.setex.call_args[0][0]
|
||||
|
||||
mock_redis.setex.reset_mock()
|
||||
|
||||
# Second connection (different tab)
|
||||
await client.post(
|
||||
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
|
||||
json={"connection_id": "connection-2"},
|
||||
)
|
||||
key2 = mock_redis.setex.call_args[0][0]
|
||||
|
||||
# Keys should be different
|
||||
assert key1 != key2
|
||||
assert "connection-1" in key1
|
||||
assert "connection-2" in key2
|
||||
153
server/tests/test_pending_joins.py
Normal file
153
server/tests/test_pending_joins.py
Normal file
@@ -0,0 +1,153 @@
|
||||
"""Tests for pending joins Redis helper functions.
|
||||
|
||||
TDD tests for tracking join intent to prevent race conditions during
|
||||
WebRTC handshake when users join meetings.
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from reflector.presence.pending_joins import (
|
||||
PENDING_JOIN_PREFIX,
|
||||
PENDING_JOIN_TTL,
|
||||
create_pending_join,
|
||||
delete_pending_join,
|
||||
has_pending_joins,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_redis():
|
||||
"""Mock async Redis client."""
|
||||
redis = AsyncMock()
|
||||
redis.setex = AsyncMock()
|
||||
redis.delete = AsyncMock()
|
||||
redis.scan = AsyncMock(return_value=(0, []))
|
||||
return redis
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_pending_join_sets_key_with_ttl(mock_redis):
|
||||
"""Test that create_pending_join stores key with correct TTL."""
|
||||
meeting_id = "meeting-123"
|
||||
user_id = "user-456"
|
||||
|
||||
await create_pending_join(mock_redis, meeting_id, user_id)
|
||||
|
||||
expected_key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
|
||||
mock_redis.setex.assert_called_once()
|
||||
call_args = mock_redis.setex.call_args
|
||||
assert call_args[0][0] == expected_key
|
||||
assert call_args[0][1] == PENDING_JOIN_TTL
|
||||
# Value should be a timestamp string
|
||||
assert call_args[0][2] is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_pending_join_removes_key(mock_redis):
|
||||
"""Test that delete_pending_join removes the key."""
|
||||
meeting_id = "meeting-123"
|
||||
user_id = "user-456"
|
||||
|
||||
await delete_pending_join(mock_redis, meeting_id, user_id)
|
||||
|
||||
expected_key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
|
||||
mock_redis.delete.assert_called_once_with(expected_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_has_pending_joins_returns_false_when_no_keys(mock_redis):
|
||||
"""Test has_pending_joins returns False when no matching keys."""
|
||||
mock_redis.scan.return_value = (0, [])
|
||||
|
||||
result = await has_pending_joins(mock_redis, "meeting-123")
|
||||
|
||||
assert result is False
|
||||
mock_redis.scan.assert_called_once()
|
||||
call_kwargs = mock_redis.scan.call_args.kwargs
|
||||
assert call_kwargs["match"] == f"{PENDING_JOIN_PREFIX}:meeting-123:*"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_has_pending_joins_returns_true_when_keys_exist(mock_redis):
|
||||
"""Test has_pending_joins returns True when matching keys found."""
|
||||
mock_redis.scan.return_value = (0, [b"pending_join:meeting-123:user-1"])
|
||||
|
||||
result = await has_pending_joins(mock_redis, "meeting-123")
|
||||
|
||||
assert result is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_has_pending_joins_scans_with_correct_pattern(mock_redis):
|
||||
"""Test has_pending_joins uses correct scan pattern."""
|
||||
meeting_id = "meeting-abc-def"
|
||||
mock_redis.scan.return_value = (0, [])
|
||||
|
||||
await has_pending_joins(mock_redis, meeting_id)
|
||||
|
||||
expected_pattern = f"{PENDING_JOIN_PREFIX}:{meeting_id}:*"
|
||||
mock_redis.scan.assert_called_once()
|
||||
call_kwargs = mock_redis.scan.call_args.kwargs
|
||||
assert call_kwargs["match"] == expected_pattern
|
||||
assert call_kwargs["count"] == 100
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_users_pending_joins(mock_redis):
|
||||
"""Test that multiple users can have pending joins for same meeting."""
|
||||
meeting_id = "meeting-123"
|
||||
# Simulate two pending joins
|
||||
mock_redis.scan.return_value = (
|
||||
0,
|
||||
[b"pending_join:meeting-123:user-1", b"pending_join:meeting-123:user-2"],
|
||||
)
|
||||
|
||||
result = await has_pending_joins(mock_redis, meeting_id)
|
||||
|
||||
assert result is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pending_join_ttl_value():
|
||||
"""Test that PENDING_JOIN_TTL has expected value."""
|
||||
# 30 seconds should be enough for WebRTC handshake but not too long
|
||||
assert PENDING_JOIN_TTL == 30
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pending_join_prefix_value():
|
||||
"""Test that PENDING_JOIN_PREFIX has expected value."""
|
||||
assert PENDING_JOIN_PREFIX == "pending_join"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_has_pending_joins_multi_iteration_scan_no_keys(mock_redis):
|
||||
"""Test has_pending_joins iterates until cursor returns 0."""
|
||||
# Simulate multi-iteration scan: cursor 100 -> cursor 50 -> cursor 0
|
||||
mock_redis.scan.side_effect = [
|
||||
(100, []), # First iteration, no keys, continue
|
||||
(50, []), # Second iteration, no keys, continue
|
||||
(0, []), # Third iteration, cursor 0, done
|
||||
]
|
||||
|
||||
result = await has_pending_joins(mock_redis, "meeting-123")
|
||||
|
||||
assert result is False
|
||||
assert mock_redis.scan.call_count == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_has_pending_joins_multi_iteration_finds_key_later(mock_redis):
|
||||
"""Test has_pending_joins finds key on second iteration."""
|
||||
# Simulate finding key on second scan iteration
|
||||
mock_redis.scan.side_effect = [
|
||||
(100, []), # First iteration, no keys
|
||||
(0, [b"pending_join:meeting-123:user-1"]), # Second iteration, found key
|
||||
]
|
||||
|
||||
result = await has_pending_joins(mock_redis, "meeting-123")
|
||||
|
||||
assert result is True
|
||||
assert mock_redis.scan.call_count == 2
|
||||
241
server/tests/test_process_meetings_pending_joins.py
Normal file
241
server/tests/test_process_meetings_pending_joins.py
Normal file
@@ -0,0 +1,241 @@
|
||||
"""Tests for process_meetings pending joins check.
|
||||
|
||||
Tests that process_meetings correctly skips deactivation when
|
||||
pending joins exist for a meeting.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from reflector.db.meetings import Meeting
|
||||
|
||||
|
||||
def _get_process_meetings_fn():
|
||||
"""Get the underlying async function without Celery/asynctask decorators."""
|
||||
from reflector.worker import process
|
||||
|
||||
fn = process.process_meetings
|
||||
# Get through both decorator layers (@shared_task and @asynctask)
|
||||
if hasattr(fn, "__wrapped__"):
|
||||
fn = fn.__wrapped__
|
||||
if hasattr(fn, "__wrapped__"):
|
||||
fn = fn.__wrapped__
|
||||
return fn
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_active_meeting():
|
||||
"""Mock an active meeting that should be considered for deactivation."""
|
||||
now = datetime.now(timezone.utc)
|
||||
return Meeting(
|
||||
id="meeting-123",
|
||||
room_id="room-456",
|
||||
room_name="test-room-20251118120000",
|
||||
room_url="https://daily.co/test-room-20251118120000",
|
||||
host_room_url="https://daily.co/test-room-20251118120000?t=host",
|
||||
platform="daily",
|
||||
num_clients=0,
|
||||
is_active=True,
|
||||
start_date=now - timedelta(hours=1),
|
||||
end_date=now - timedelta(minutes=30), # Already ended
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.worker.process.meetings_controller.get_all_active")
|
||||
@patch("reflector.worker.process.RedisAsyncLock")
|
||||
@patch("reflector.worker.process.create_platform_client")
|
||||
@patch("reflector.worker.process.get_async_redis_client")
|
||||
@patch("reflector.worker.process.has_pending_joins")
|
||||
@patch("reflector.worker.process.meetings_controller.update_meeting")
|
||||
async def test_process_meetings_skips_deactivation_with_pending_joins(
|
||||
mock_update_meeting,
|
||||
mock_has_pending_joins,
|
||||
mock_get_redis,
|
||||
mock_create_client,
|
||||
mock_redis_lock_class,
|
||||
mock_get_all_active,
|
||||
mock_active_meeting,
|
||||
):
|
||||
"""Test that process_meetings skips deactivation when pending joins exist."""
|
||||
process_meetings = _get_process_meetings_fn()
|
||||
|
||||
mock_get_all_active.return_value = [mock_active_meeting]
|
||||
|
||||
# Mock lock acquired
|
||||
mock_lock_instance = AsyncMock()
|
||||
mock_lock_instance.acquired = True
|
||||
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
|
||||
mock_lock_instance.__aexit__ = AsyncMock()
|
||||
mock_redis_lock_class.return_value = mock_lock_instance
|
||||
|
||||
# Mock platform client - no active sessions, but had sessions (triggers deactivation)
|
||||
mock_daily_client = AsyncMock()
|
||||
mock_session = AsyncMock()
|
||||
mock_session.ended_at = datetime.now(timezone.utc) # Session ended
|
||||
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
|
||||
mock_create_client.return_value = mock_daily_client
|
||||
|
||||
# Mock Redis client
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
# Mock pending joins exist
|
||||
mock_has_pending_joins.return_value = True
|
||||
|
||||
await process_meetings()
|
||||
|
||||
# Verify has_pending_joins was called
|
||||
mock_has_pending_joins.assert_called_once_with(mock_redis, mock_active_meeting.id)
|
||||
|
||||
# Verify meeting was NOT deactivated
|
||||
mock_update_meeting.assert_not_called()
|
||||
|
||||
# Verify Redis was closed
|
||||
mock_redis.aclose.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.worker.process.meetings_controller.get_all_active")
|
||||
@patch("reflector.worker.process.RedisAsyncLock")
|
||||
@patch("reflector.worker.process.create_platform_client")
|
||||
@patch("reflector.worker.process.get_async_redis_client")
|
||||
@patch("reflector.worker.process.has_pending_joins")
|
||||
@patch("reflector.worker.process.meetings_controller.update_meeting")
|
||||
async def test_process_meetings_deactivates_without_pending_joins(
|
||||
mock_update_meeting,
|
||||
mock_has_pending_joins,
|
||||
mock_get_redis,
|
||||
mock_create_client,
|
||||
mock_redis_lock_class,
|
||||
mock_get_all_active,
|
||||
mock_active_meeting,
|
||||
):
|
||||
"""Test that process_meetings deactivates when no pending joins."""
|
||||
process_meetings = _get_process_meetings_fn()
|
||||
|
||||
mock_get_all_active.return_value = [mock_active_meeting]
|
||||
|
||||
# Mock lock acquired
|
||||
mock_lock_instance = AsyncMock()
|
||||
mock_lock_instance.acquired = True
|
||||
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
|
||||
mock_lock_instance.__aexit__ = AsyncMock()
|
||||
mock_redis_lock_class.return_value = mock_lock_instance
|
||||
|
||||
# Mock platform client - no active sessions, but had sessions
|
||||
mock_daily_client = AsyncMock()
|
||||
mock_session = AsyncMock()
|
||||
mock_session.ended_at = datetime.now(timezone.utc)
|
||||
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
|
||||
mock_create_client.return_value = mock_daily_client
|
||||
|
||||
# Mock Redis client
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
# Mock no pending joins
|
||||
mock_has_pending_joins.return_value = False
|
||||
|
||||
await process_meetings()
|
||||
|
||||
# Verify meeting was deactivated
|
||||
mock_update_meeting.assert_called_once_with(mock_active_meeting.id, is_active=False)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.worker.process.meetings_controller.get_all_active")
|
||||
@patch("reflector.worker.process.RedisAsyncLock")
|
||||
@patch("reflector.worker.process.create_platform_client")
|
||||
async def test_process_meetings_no_check_when_active_sessions(
|
||||
mock_create_client,
|
||||
mock_redis_lock_class,
|
||||
mock_get_all_active,
|
||||
mock_active_meeting,
|
||||
):
|
||||
"""Test that pending joins check is skipped when there are active sessions."""
|
||||
process_meetings = _get_process_meetings_fn()
|
||||
|
||||
mock_get_all_active.return_value = [mock_active_meeting]
|
||||
|
||||
# Mock lock acquired
|
||||
mock_lock_instance = AsyncMock()
|
||||
mock_lock_instance.acquired = True
|
||||
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
|
||||
mock_lock_instance.__aexit__ = AsyncMock()
|
||||
mock_redis_lock_class.return_value = mock_lock_instance
|
||||
|
||||
# Mock platform client - has active session
|
||||
mock_daily_client = AsyncMock()
|
||||
mock_session = AsyncMock()
|
||||
mock_session.ended_at = None # Still active
|
||||
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
|
||||
mock_create_client.return_value = mock_daily_client
|
||||
|
||||
with (
|
||||
patch("reflector.worker.process.get_async_redis_client") as mock_get_redis,
|
||||
patch("reflector.worker.process.has_pending_joins") as mock_has_pending_joins,
|
||||
patch(
|
||||
"reflector.worker.process.meetings_controller.update_meeting"
|
||||
) as mock_update_meeting,
|
||||
):
|
||||
await process_meetings()
|
||||
|
||||
# Verify pending joins check was NOT called (no need - active sessions exist)
|
||||
mock_has_pending_joins.assert_not_called()
|
||||
|
||||
# Verify meeting was NOT deactivated
|
||||
mock_update_meeting.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("reflector.worker.process.meetings_controller.get_all_active")
|
||||
@patch("reflector.worker.process.RedisAsyncLock")
|
||||
@patch("reflector.worker.process.create_platform_client")
|
||||
@patch("reflector.worker.process.get_async_redis_client")
|
||||
@patch("reflector.worker.process.has_pending_joins")
|
||||
@patch("reflector.worker.process.meetings_controller.update_meeting")
|
||||
async def test_process_meetings_closes_redis_even_on_continue(
|
||||
mock_update_meeting,
|
||||
mock_has_pending_joins,
|
||||
mock_get_redis,
|
||||
mock_create_client,
|
||||
mock_redis_lock_class,
|
||||
mock_get_all_active,
|
||||
mock_active_meeting,
|
||||
):
|
||||
"""Test that Redis connection is always closed, even when skipping deactivation."""
|
||||
process_meetings = _get_process_meetings_fn()
|
||||
|
||||
mock_get_all_active.return_value = [mock_active_meeting]
|
||||
|
||||
# Mock lock acquired
|
||||
mock_lock_instance = AsyncMock()
|
||||
mock_lock_instance.acquired = True
|
||||
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
|
||||
mock_lock_instance.__aexit__ = AsyncMock()
|
||||
mock_redis_lock_class.return_value = mock_lock_instance
|
||||
|
||||
# Mock platform client - no active sessions
|
||||
mock_daily_client = AsyncMock()
|
||||
mock_session = AsyncMock()
|
||||
mock_session.ended_at = datetime.now(timezone.utc)
|
||||
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
|
||||
mock_create_client.return_value = mock_daily_client
|
||||
|
||||
# Mock Redis client
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.aclose = AsyncMock()
|
||||
mock_get_redis.return_value = mock_redis
|
||||
|
||||
# Mock pending joins exist (will trigger continue)
|
||||
mock_has_pending_joins.return_value = True
|
||||
|
||||
await process_meetings()
|
||||
|
||||
# Verify Redis was closed
|
||||
mock_redis.aclose.assert_called_once()
|
||||
Reference in New Issue
Block a user