quaik8/sdb2-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add VLC player API endpoints and associated tests

Browse Source 
- Implemented VLC player API endpoints for playing and stopping sounds.
- Added tests for successful playback, error handling, and authentication scenarios.
- Created utility function to get sound file paths based on sound properties.
- Refactored player service to utilize shared sound path utility.
- Enhanced test coverage for sound file path utility with various sound types.
- Introduced tests for VLC player service, including subprocess handling and play count tracking.
This commit is contained in:
JSC
2025-07-30 20:46:49 +02:00
parent 1b0d291ad3
commit dd10ef5d41
9 changed files with 1413 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

511
tests/services/test_vlc_player.py Normal file
View File

@@ -0,0 +1,511 @@
"""Tests for VLC player service."""
import asyncio
from pathlib import Path
from unittest.mock import AsyncMock, Mock, patch
import pytest
from app.models.sound import Sound
from app.models.sound_played import SoundPlayed
from app.models.user import User
from app.services.vlc_player import VLCPlayerService, get_vlc_player_service
from app.utils.audio import get_sound_file_path
class TestVLCPlayerService:
"""Test VLC player service."""
@pytest.fixture
def vlc_service(self):
"""Create a VLC service instance."""
with patch("app.services.vlc_player.subprocess.run") as mock_run:
# Mock VLC executable detection
mock_run.return_value.returncode = 0
return VLCPlayerService()
@pytest.fixture
def vlc_service_with_db(self):
"""Create a VLC service instance with database session factory."""
with patch("app.services.vlc_player.subprocess.run") as mock_run:
# Mock VLC executable detection
mock_run.return_value.returncode = 0
mock_session_factory = Mock()
return VLCPlayerService(mock_session_factory)
@pytest.fixture
def sample_sound(self):
"""Create a sample sound for testing."""
return Sound(
id=1,
type="SDB",
name="Test Sound",
filename="test_audio.mp3",
duration=5000,
size=1024,
hash="test_hash",
is_normalized=False,
normalized_filename=None,
)
@pytest.fixture
def normalized_sound(self):
"""Create a normalized sound for testing."""
return Sound(
id=2,
type="TTS",
name="Normalized Sound",
filename="original.wav",
duration=7500,
size=2048,
hash="normalized_hash",
is_normalized=True,
normalized_filename="normalized.mp3",
)
def test_init(self, vlc_service):
"""Test VLC service initialization."""
assert vlc_service.vlc_executable is not None
assert isinstance(vlc_service.vlc_executable, str)
@patch("app.services.vlc_player.subprocess.run")
def test_find_vlc_executable_found_in_path(self, mock_run):
"""Test VLC executable detection when found in PATH."""
mock_run.return_value.returncode = 0
service = VLCPlayerService()
assert service.vlc_executable == "vlc"
@patch("app.services.vlc_player.subprocess.run")
def test_find_vlc_executable_found_by_path(self, mock_run):
"""Test VLC executable detection when found by absolute path."""
mock_run.return_value.returncode = 1 # which command fails
# Mock Path to return True for the first absolute path
with patch("app.services.vlc_player.Path") as mock_path:
def path_side_effect(path_str):
mock_instance = Mock()
mock_instance.exists.return_value = str(path_str) == "/usr/bin/vlc"
return mock_instance
mock_path.side_effect = path_side_effect
service = VLCPlayerService()
assert service.vlc_executable == "/usr/bin/vlc"
@patch("app.services.vlc_player.subprocess.run")
@patch("app.services.vlc_player.Path")
def test_find_vlc_executable_fallback(self, mock_path, mock_run):
"""Test VLC executable detection fallback to default."""
# Mock all paths as non-existent
mock_path_instance = Mock()
mock_path_instance.exists.return_value = False
mock_path.return_value = mock_path_instance
# Mock which command as failing
mock_run.return_value.returncode = 1
service = VLCPlayerService()
assert service.vlc_executable == "vlc"
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_play_sound_success(
self, mock_subprocess, vlc_service, sample_sound
):
"""Test successful sound playback."""
# Mock subprocess
mock_process = Mock()
mock_process.pid = 12345
mock_subprocess.return_value = mock_process
# Mock the file path utility to avoid Path issues
with patch("app.services.vlc_player.get_sound_file_path") as mock_get_path:
mock_path = Mock()
mock_path.exists.return_value = True
mock_get_path.return_value = mock_path
result = await vlc_service.play_sound(sample_sound)
assert result is True
mock_subprocess.assert_called_once()
args = mock_subprocess.call_args
# Check command arguments
cmd_args = args[1] # keyword arguments
assert "--play-and-exit" in args[0]
assert "--intf" in args[0]
assert "dummy" in args[0]
assert "--no-video" in args[0]
assert "--no-repeat" in args[0]
assert "--no-loop" in args[0]
assert cmd_args["stdout"] == asyncio.subprocess.DEVNULL
assert cmd_args["stderr"] == asyncio.subprocess.DEVNULL
@pytest.mark.asyncio
async def test_play_sound_file_not_found(
self, vlc_service, sample_sound
):
"""Test sound playback when file doesn't exist."""
# Mock the file path utility to return a non-existent path
with patch("app.services.vlc_player.get_sound_file_path") as mock_get_path:
mock_path = Mock()
mock_path.exists.return_value = False
mock_get_path.return_value = mock_path
result = await vlc_service.play_sound(sample_sound)
assert result is False
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_play_sound_subprocess_error(
self, mock_subprocess, vlc_service, sample_sound
):
"""Test sound playback when subprocess fails."""
# Mock the file path utility to return an existing path
with patch("app.services.vlc_player.get_sound_file_path") as mock_get_path:
mock_path = Mock()
mock_path.exists.return_value = True
mock_get_path.return_value = mock_path
# Mock subprocess exception
mock_subprocess.side_effect = Exception("Subprocess failed")
result = await vlc_service.play_sound(sample_sound)
assert result is False
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_stop_all_vlc_instances_success(self, mock_subprocess, vlc_service):
"""Test successful stopping of all VLC instances."""
# Mock pgrep process (find VLC processes)
mock_find_process = Mock()
mock_find_process.returncode = 0
mock_find_process.communicate = AsyncMock(
return_value=(b"12345\n67890\n", b"")
)
# Mock pkill process (kill VLC processes)
mock_kill_process = Mock()
mock_kill_process.communicate = AsyncMock(return_value=(b"", b""))
# Mock verify process (check remaining processes)
mock_verify_process = Mock()
mock_verify_process.returncode = 1 # No processes found
mock_verify_process.communicate = AsyncMock(return_value=(b"", b""))
# Set up subprocess mock to return different processes for each call
mock_subprocess.side_effect = [
mock_find_process,
mock_kill_process,
mock_verify_process,
]
result = await vlc_service.stop_all_vlc_instances()
assert result["success"] is True
assert result["processes_found"] == 2
assert result["processes_killed"] == 2
assert result["processes_remaining"] == 0
assert "Killed 2 VLC processes" in result["message"]
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_stop_all_vlc_instances_no_processes(
self, mock_subprocess, vlc_service
):
"""Test stopping VLC instances when none are running."""
# Mock pgrep process (no VLC processes found)
mock_find_process = Mock()
mock_find_process.returncode = 1 # No processes found
mock_find_process.communicate = AsyncMock(return_value=(b"", b""))
mock_subprocess.return_value = mock_find_process
result = await vlc_service.stop_all_vlc_instances()
assert result["success"] is True
assert result["processes_found"] == 0
assert result["processes_killed"] == 0
assert result["message"] == "No VLC processes found"
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_stop_all_vlc_instances_partial_kill(
self, mock_subprocess, vlc_service
):
"""Test stopping VLC instances when some processes remain."""
# Mock pgrep process (find VLC processes)
mock_find_process = Mock()
mock_find_process.returncode = 0
mock_find_process.communicate = AsyncMock(
return_value=(b"12345\n67890\n11111\n", b"")
)
# Mock pkill process (kill VLC processes)
mock_kill_process = Mock()
mock_kill_process.communicate = AsyncMock(return_value=(b"", b""))
# Mock verify process (one process remains)
mock_verify_process = Mock()
mock_verify_process.returncode = 0
mock_verify_process.communicate = AsyncMock(return_value=(b"11111\n", b""))
mock_subprocess.side_effect = [
mock_find_process,
mock_kill_process,
mock_verify_process,
]
result = await vlc_service.stop_all_vlc_instances()
assert result["success"] is True
assert result["processes_found"] == 3
assert result["processes_killed"] == 2
assert result["processes_remaining"] == 1
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_stop_all_vlc_instances_error(self, mock_subprocess, vlc_service):
"""Test stopping VLC instances when an error occurs."""
# Mock subprocess exception
mock_subprocess.side_effect = Exception("Command failed")
result = await vlc_service.stop_all_vlc_instances()
assert result["success"] is False
assert result["processes_found"] == 0
assert result["processes_killed"] == 0
assert "error" in result
assert result["message"] == "Failed to stop VLC processes"
def test_get_vlc_player_service_singleton(self):
"""Test that get_vlc_player_service returns the same instance."""
with patch("app.services.vlc_player.VLCPlayerService") as mock_service_class:
mock_instance = Mock()
mock_service_class.return_value = mock_instance
# Clear the global instance
import app.services.vlc_player
app.services.vlc_player.vlc_player_service = None
# First call should create new instance
service1 = get_vlc_player_service()
assert service1 == mock_instance
mock_service_class.assert_called_once()
# Second call should return same instance
service2 = get_vlc_player_service()
assert service2 == mock_instance
assert service1 is service2
# Constructor should not be called again
mock_service_class.assert_called_once()
@pytest.mark.asyncio
@patch("app.services.vlc_player.asyncio.create_subprocess_exec")
async def test_play_sound_with_play_count_tracking(
self, mock_subprocess, vlc_service_with_db, sample_sound
):
"""Test sound playback with play count tracking."""
# Mock subprocess
mock_process = Mock()
mock_process.pid = 12345
mock_subprocess.return_value = mock_process
# Mock session and repositories
mock_session = AsyncMock()
vlc_service_with_db.db_session_factory.return_value = mock_session
# Mock repositories
mock_sound_repo = AsyncMock()
mock_user_repo = AsyncMock()
with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo):
with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo):
with patch("app.services.vlc_player.socket_manager") as mock_socket:
with patch("app.services.vlc_player.select") as mock_select:
# Mock the file path utility
with patch("app.services.vlc_player.get_sound_file_path") as mock_get_path:
mock_path = Mock()
mock_path.exists.return_value = True
mock_get_path.return_value = mock_path
# Mock sound repository responses
updated_sound = Sound(
id=1,
type="SDB",
name="Test Sound",
filename="test.mp3",
duration=5000,
size=1024,
hash="test_hash",
play_count=1, # Updated count
)
mock_sound_repo.get_by_id.return_value = sample_sound
mock_sound_repo.update.return_value = updated_sound
# Mock admin user
admin_user = User(
id=1,
email="admin@test.com",
name="Admin User",
role="admin",
)
mock_user_repo.get_by_id.return_value = admin_user
# Mock socket broadcast
mock_socket.broadcast_to_all = AsyncMock()
result = await vlc_service_with_db.play_sound(sample_sound)
# Wait a bit for the async task to complete
await asyncio.sleep(0.1)
assert result is True
# Verify subprocess was called
mock_subprocess.assert_called_once()
# Note: The async task runs in the background, so we can't easily
# verify the database operations in this test without more complex
# mocking or using a real async test framework setup
@pytest.mark.asyncio
async def test_record_play_count_success(self, vlc_service_with_db):
"""Test successful play count recording."""
# Mock session and repositories
mock_session = AsyncMock()
vlc_service_with_db.db_session_factory.return_value = mock_session
mock_sound_repo = AsyncMock()
mock_user_repo = AsyncMock()
# Create test sound and user
test_sound = Sound(
id=1,
type="SDB",
name="Test Sound",
filename="test.mp3",
duration=5000,
size=1024,
hash="test_hash",
play_count=0,
)
admin_user = User(
id=1,
email="admin@test.com",
name="Admin User",
role="admin",
)
with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo):
with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo):
with patch("app.services.vlc_player.socket_manager") as mock_socket:
with patch("app.services.vlc_player.select") as mock_select:
# Setup mocks
mock_sound_repo.get_by_id.return_value = test_sound
mock_user_repo.get_by_id.return_value = admin_user
# Mock socket broadcast
mock_socket.broadcast_to_all = AsyncMock()
await vlc_service_with_db._record_play_count(1, "Test Sound")
# Verify sound repository calls
mock_sound_repo.get_by_id.assert_called_once_with(1)
mock_sound_repo.update.assert_called_once_with(
test_sound, {"play_count": 1}
)
# Verify user repository calls
mock_user_repo.get_by_id.assert_called_once_with(1)
# Verify session operations
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()
mock_session.close.assert_called_once()
# Verify socket broadcast
mock_socket.broadcast_to_all.assert_called_once_with(
"sound_played",
{
"sound_id": 1,
"sound_name": "Test Sound",
"user_id": 1,
"play_count": 1,
},
)
@pytest.mark.asyncio
async def test_record_play_count_no_session_factory(self, vlc_service):
"""Test play count recording when no session factory is available."""
# This should not raise an error and should log a warning
await vlc_service._record_play_count(1, "Test Sound")
# The method should return early without doing anything
@pytest.mark.asyncio
async def test_record_play_count_always_creates_record(self, vlc_service_with_db):
"""Test play count recording always creates a new SoundPlayed record."""
# Mock session and repositories
mock_session = AsyncMock()
vlc_service_with_db.db_session_factory.return_value = mock_session
mock_sound_repo = AsyncMock()
mock_user_repo = AsyncMock()
# Create test sound and user
test_sound = Sound(
id=1,
type="SDB",
name="Test Sound",
filename="test.mp3",
duration=5000,
size=1024,
hash="test_hash",
play_count=5,
)
admin_user = User(
id=1,
email="admin@test.com",
name="Admin User",
role="admin",
)
with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo):
with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo):
with patch("app.services.vlc_player.socket_manager") as mock_socket:
# Setup mocks
mock_sound_repo.get_by_id.return_value = test_sound
mock_user_repo.get_by_id.return_value = admin_user
# Mock socket broadcast
mock_socket.broadcast_to_all = AsyncMock()
await vlc_service_with_db._record_play_count(1, "Test Sound")
# Verify sound play count was updated
mock_sound_repo.update.assert_called_once_with(
test_sound, {"play_count": 6}
)
# Verify new SoundPlayed record was always added
mock_session.add.assert_called_once()
# Verify commit happened
mock_session.commit.assert_called_once()
def test_uses_shared_sound_path_utility(self, vlc_service, sample_sound):
"""Test that VLC service uses the shared sound path utility."""
with patch("app.services.vlc_player.get_sound_file_path") as mock_path:
mock_file_path = Mock(spec=Path)
mock_file_path.exists.return_value = False # File doesn't exist
mock_path.return_value = mock_file_path
# This should fail because file doesn't exist
result = asyncio.run(vlc_service.play_sound(sample_sound))
# Verify the utility was called and returned False
mock_path.assert_called_once_with(sample_sound)
assert result is False