"""Tests for extraction service.""" import tempfile from pathlib import Path from unittest.mock import AsyncMock, Mock, patch import pytest from sqlmodel.ext.asyncio.session import AsyncSession from app.models.extraction import Extraction from app.models.sound import Sound from app.services.extraction import ExtractionService class TestExtractionService: """Test extraction service.""" @pytest.fixture def mock_session(self): """Create a mock session.""" return Mock(spec=AsyncSession) @pytest.fixture def extraction_service(self, mock_session): """Create an extraction service with mock session.""" with patch("app.services.extraction.Path.mkdir"): return ExtractionService(mock_session) def test_init(self, extraction_service) -> None: """Test service initialization.""" assert extraction_service.session is not None assert extraction_service.extraction_repo is not None assert extraction_service.sound_repo is not None def test_sanitize_filename(self, extraction_service) -> None: """Test filename sanitization.""" test_cases = [ ("Hello World", "Hello World"), ("Test<>Video", "Test__Video"), ("Bad/File\\Name", "Bad_File_Name"), (" Spaces ", "Spaces"), ( "Very long filename that exceeds the maximum length limit and should be truncated to 100 characters maximum", "Very long filename that exceeds the maximum length limit and should be truncated to 100 characters m", ), ("", "untitled"), ] for input_name, expected in test_cases: result = extraction_service._sanitize_filename(input_name) assert result == expected @patch("app.services.extraction.yt_dlp.YoutubeDL") @pytest.mark.asyncio async def test_detect_service_info_youtube( self, mock_ydl_class, extraction_service, ) -> None: """Test service detection for YouTube.""" mock_ydl = Mock() mock_ydl_class.return_value.__enter__.return_value = mock_ydl mock_ydl.extract_info.return_value = { "extractor": "youtube", "id": "test123", "title": "Test Video", "duration": 240, "uploader": "Test Channel", } result = await extraction_service._detect_service_info( "https://www.youtube.com/watch?v=test123", ) assert result is not None assert result["service"] == "youtube" assert result["service_id"] == "test123" assert result["title"] == "Test Video" @patch("app.services.extraction.yt_dlp.YoutubeDL") @pytest.mark.asyncio async def test_detect_service_info_failure( self, mock_ydl_class, extraction_service, ) -> None: """Test service detection failure.""" mock_ydl = Mock() mock_ydl_class.return_value.__enter__.return_value = mock_ydl mock_ydl.extract_info.side_effect = Exception("Network error") result = await extraction_service._detect_service_info("https://invalid.url") assert result is None @pytest.mark.asyncio async def test_create_extraction_success(self, extraction_service) -> None: """Test successful extraction creation.""" url = "https://www.youtube.com/watch?v=test123" user_id = 1 # Mock user for user_name retrieval mock_user = Mock() mock_user.name = "Test User" extraction_service.user_repo.get_by_id = AsyncMock(return_value=mock_user) # Mock repository call - no service detection happens during creation mock_extraction = Extraction( id=1, url=url, user_id=user_id, service=None, # Service detection deferred to processing service_id=None, # Service detection deferred to processing title=None, # Service detection deferred to processing status="pending", ) extraction_service.extraction_repo.create = AsyncMock( return_value=mock_extraction, ) result = await extraction_service.create_extraction(url, user_id) assert result["id"] == 1 assert result["service"] is None # Not detected during creation assert result["service_id"] is None # Not detected during creation assert result["title"] is None # Not detected during creation assert result["status"] == "pending" assert result["user_name"] == "Test User" @pytest.mark.asyncio async def test_create_extraction_basic(self, extraction_service) -> None: """Test basic extraction creation without validation.""" url = "https://www.youtube.com/watch?v=test123" user_id = 1 # Mock user for user_name retrieval mock_user = Mock() mock_user.name = "Test User" extraction_service.user_repo.get_by_id = AsyncMock(return_value=mock_user) # Mock repository call - creation always succeeds now mock_extraction = Extraction( id=2, url=url, user_id=user_id, service=None, service_id=None, title=None, status="pending", ) extraction_service.extraction_repo.create = AsyncMock( return_value=mock_extraction, ) result = await extraction_service.create_extraction(url, user_id) assert result["id"] == 2 assert result["url"] == url assert result["status"] == "pending" assert result["user_name"] == "Test User" @pytest.mark.asyncio async def test_create_extraction_any_url(self, extraction_service) -> None: """Test extraction creation accepts any URL.""" url = "https://invalid.url" user_id = 1 # Mock user for user_name retrieval mock_user = Mock() mock_user.name = "Test User" extraction_service.user_repo.get_by_id = AsyncMock(return_value=mock_user) # Mock repository call - even invalid URLs are accepted during creation mock_extraction = Extraction( id=3, url=url, user_id=user_id, service=None, service_id=None, title=None, status="pending", ) extraction_service.extraction_repo.create = AsyncMock( return_value=mock_extraction, ) result = await extraction_service.create_extraction(url, user_id) assert result["id"] == 3 assert result["url"] == url assert result["status"] == "pending" assert result["user_name"] == "Test User" @pytest.mark.asyncio async def test_process_extraction_with_service_detection( self, extraction_service, ) -> None: """Test extraction processing with service detection.""" extraction_id = 1 # Mock extraction without service info mock_extraction = Extraction( id=extraction_id, url="https://www.youtube.com/watch?v=test123", user_id=1, service=None, service_id=None, title=None, status="pending", ) extraction_service.extraction_repo.get_by_id = AsyncMock( return_value=mock_extraction, ) extraction_service.extraction_repo.update = AsyncMock() extraction_service.extraction_repo.get_by_service_and_id = AsyncMock( return_value=None, ) # Mock user repository from app.models.user import User mock_user = User(id=1, name="Test User") extraction_service.user_repo.get_by_id = AsyncMock(return_value=mock_user) # Mock service detection service_info = { "service": "youtube", "service_id": "test123", "title": "Test Video", } with ( patch.object( extraction_service, "_detect_service_info", return_value=service_info, ), patch.object(extraction_service, "_extract_media") as mock_extract, patch.object( extraction_service, "_move_files_to_final_location", ) as mock_move, patch.object( extraction_service, "_create_sound_record", ) as mock_create_sound, patch.object(extraction_service, "_normalize_sound"), patch.object(extraction_service, "_add_to_main_playlist"), ): mock_sound = Sound(id=42, type="EXT", name="Test", filename="test.mp3") mock_extract.return_value = (Path("/fake/audio.mp3"), None) mock_move.return_value = (Path("/final/audio.mp3"), None) mock_create_sound.return_value = mock_sound result = await extraction_service.process_extraction(extraction_id) # Verify service detection was called extraction_service._detect_service_info.assert_called_once_with( "https://www.youtube.com/watch?v=test123", ) # Verify extraction was updated with service info extraction_service.extraction_repo.update.assert_called() assert result["status"] == "completed" assert result["service"] == "youtube" assert result["service_id"] == "test123" assert result["title"] == "Test Video" def test_ensure_unique_filename(self, extraction_service) -> None: """Test unique filename generation.""" with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # Create original file original_file = temp_path / "test.mp3" original_file.touch() # Test unique filename generation result = extraction_service._ensure_unique_filename(original_file) expected = temp_path / "test_1.mp3" assert result == expected # Create the first duplicate and test again expected.touch() result = extraction_service._ensure_unique_filename(original_file) expected_2 = temp_path / "test_2.mp3" assert result == expected_2 @pytest.mark.asyncio async def test_create_sound_record(self, extraction_service) -> None: """Test sound record creation.""" # Create temporary audio file with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: audio_path = Path(f.name) f.write(b"fake audio data") try: extraction = Extraction( id=1, service="youtube", service_id="test123", title="Test Video", url="https://www.youtube.com/watch?v=test123", user_id=1, status="processing", ) mock_sound = Sound( id=1, type="EXT", name="Test Video", filename=audio_path.name, duration=240000, size=1024, hash="test_hash", is_deletable=True, is_music=True, is_normalized=False, play_count=0, ) with ( patch( "app.services.extraction.get_audio_duration", return_value=240000, ), patch("app.services.extraction.get_file_size", return_value=1024), patch( "app.services.extraction.get_file_hash", return_value="test_hash", ), ): extraction_service.sound_repo.create = AsyncMock( return_value=mock_sound, ) result = await extraction_service._create_sound_record( audio_path, None, # thumbnail_path extraction.title, extraction.service, extraction.service_id, ) assert result.type == "EXT" assert result.name == "Test Video" assert result.is_deletable is True assert result.is_music is True assert result.is_normalized is False finally: audio_path.unlink() @pytest.mark.asyncio async def test_normalize_sound_success(self, extraction_service) -> None: """Test sound normalization.""" sound = Sound( id=1, type="EXT", name="Test Sound", filename="test.mp3", duration=240000, size=1024, hash="test_hash", is_normalized=False, ) sound_id = 1 # Mock sound repository to return the sound extraction_service.sound_repo.get_by_id = AsyncMock(return_value=sound) mock_normalizer = Mock() mock_normalizer.normalize_sound = AsyncMock( return_value={"status": "normalized"}, ) with patch( "app.services.extraction.SoundNormalizerService", return_value=mock_normalizer, ): # Should not raise exception await extraction_service._normalize_sound(sound_id) extraction_service.sound_repo.get_by_id.assert_called_once_with(sound_id) mock_normalizer.normalize_sound.assert_called_once_with(sound) @pytest.mark.asyncio async def test_normalize_sound_failure(self, extraction_service) -> None: """Test sound normalization failure.""" sound = Sound( id=1, type="EXT", name="Test Sound", filename="test.mp3", duration=240000, size=1024, hash="test_hash", is_normalized=False, ) sound_id = 1 # Mock sound repository to return the sound extraction_service.sound_repo.get_by_id = AsyncMock(return_value=sound) mock_normalizer = Mock() mock_normalizer.normalize_sound = AsyncMock( return_value={"status": "error", "error": "Test error"}, ) with patch( "app.services.extraction.SoundNormalizerService", return_value=mock_normalizer, ): # Should not raise exception even on failure await extraction_service._normalize_sound(sound_id) extraction_service.sound_repo.get_by_id.assert_called_once_with(sound_id) mock_normalizer.normalize_sound.assert_called_once_with(sound) @pytest.mark.asyncio async def test_get_extraction_by_id(self, extraction_service) -> None: """Test getting extraction by ID.""" extraction = Extraction( id=1, service="youtube", service_id="test123", url="https://www.youtube.com/watch?v=test123", user_id=1, title="Test Video", status="completed", sound_id=42, ) # Mock user for user_name retrieval mock_user = Mock() mock_user.name = "Test User" extraction_service.extraction_repo.get_by_id = AsyncMock( return_value=extraction, ) extraction_service.user_repo.get_by_id = AsyncMock( return_value=mock_user, ) result = await extraction_service.get_extraction_by_id(1) assert result is not None assert result["id"] == 1 assert result["service"] == "youtube" assert result["service_id"] == "test123" assert result["title"] == "Test Video" assert result["status"] == "completed" assert result["sound_id"] == 42 assert result["user_name"] == "Test User" @pytest.mark.asyncio async def test_get_extraction_by_id_not_found(self, extraction_service) -> None: """Test getting extraction by ID when not found.""" extraction_service.extraction_repo.get_by_id = AsyncMock(return_value=None) result = await extraction_service.get_extraction_by_id(999) assert result is None @pytest.mark.asyncio async def test_get_user_extractions(self, extraction_service) -> None: """Test getting user extractions.""" from app.models.user import User user = User(id=1, name="Test User", email="test@example.com") extractions = [ ( Extraction( id=1, service="youtube", service_id="test123", url="https://www.youtube.com/watch?v=test123", user_id=1, title="Test Video 1", status="completed", sound_id=42, ), user, ), ( Extraction( id=2, service="youtube", service_id="test456", url="https://www.youtube.com/watch?v=test456", user_id=1, title="Test Video 2", status="pending", ), user, ), ] extraction_service.extraction_repo.get_user_extractions_filtered = AsyncMock( return_value=(extractions, 2), ) result = await extraction_service.get_user_extractions(1) assert result["total"] == 2 assert len(result["extractions"]) == 2 assert result["extractions"][0]["id"] == 1 assert result["extractions"][0]["title"] == "Test Video 1" assert result["extractions"][0]["user_name"] == "Test User" assert result["extractions"][1]["id"] == 2 assert result["extractions"][1]["title"] == "Test Video 2" assert result["extractions"][1]["user_name"] == "Test User" @pytest.mark.asyncio async def test_get_pending_extractions(self, extraction_service) -> None: """Test getting pending extractions.""" from app.models.user import User user = User(id=1, name="Test User", email="test@example.com") pending_extractions = [ ( Extraction( id=1, service="youtube", service_id="test123", url="https://www.youtube.com/watch?v=test123", user_id=1, title="Pending Video", status="pending", ), user, ), ] extraction_service.extraction_repo.get_pending_extractions = AsyncMock( return_value=pending_extractions, ) result = await extraction_service.get_pending_extractions() assert len(result) == 1 assert result[0]["id"] == 1 assert result[0]["status"] == "pending" assert result[0]["user_name"] == "Test User" @pytest.mark.asyncio async def test_delete_extraction_with_sound(self, extraction_service, test_user): """Test deleting extraction with associated sound and files.""" import tempfile from pathlib import Path # Create temporary directories for testing with tempfile.TemporaryDirectory() as temp_dir: temp_dir_path = Path(temp_dir) # Set up temporary directories original_dir = temp_dir_path / "originals" / "extracted" normalized_dir = temp_dir_path / "normalized" / "extracted" thumbnail_dir = temp_dir_path / "thumbnails" original_dir.mkdir(parents=True) normalized_dir.mkdir(parents=True) thumbnail_dir.mkdir(parents=True) # Create test files audio_file = original_dir / "test_audio.mp3" normalized_file = normalized_dir / "test_audio.mp3" thumbnail_file = thumbnail_dir / "test_thumb.jpg" audio_file.write_text("audio content") normalized_file.write_text("normalized content") thumbnail_file.write_text("thumbnail content") # Create extraction and sound records extraction = Extraction( id=1, url="https://example.com/video", user_id=test_user.id, status="completed", sound_id=1, ) sound = Sound( id=1, type="EXT", name="Test Audio", filename="test_audio.mp3", duration=60000, size=2048, hash="test_hash", is_normalized=True, normalized_filename="test_audio.mp3", thumbnail="test_thumb.jpg", is_deletable=True, is_music=True, ) # Mock repository methods extraction_service.extraction_repo.get_by_id = AsyncMock(return_value=extraction) extraction_service.sound_repo.get_by_id = AsyncMock(return_value=sound) extraction_service.extraction_repo.delete = AsyncMock() extraction_service.sound_repo.delete = AsyncMock() extraction_service.session.commit = AsyncMock() extraction_service.session.rollback = AsyncMock() # Monkey patch the paths in the service method import app.services.extraction original_path_class = app.services.extraction.Path def mock_path(*args: str): path_str = str(args[0]) if path_str == "sounds/originals/extracted": return original_dir if path_str == "sounds/normalized/extracted": return normalized_dir if path_str.endswith("thumbnails"): return thumbnail_dir return original_path_class(*args) # Mock the Path constructor and settings with patch("app.services.extraction.Path", side_effect=mock_path), \ patch("app.services.extraction.settings") as mock_settings: mock_settings.EXTRACTION_THUMBNAILS_DIR = str(thumbnail_dir) # Test deletion result = await extraction_service.delete_extraction(1, test_user.id) assert result is True # Verify repository calls extraction_service.extraction_repo.get_by_id.assert_called_once_with(1) extraction_service.sound_repo.get_by_id.assert_called_once_with(1) extraction_service.extraction_repo.delete.assert_called_once_with(extraction) extraction_service.sound_repo.delete.assert_called_once_with(sound) extraction_service.session.commit.assert_called_once() # Verify files were deleted assert not audio_file.exists() assert not normalized_file.exists() assert not thumbnail_file.exists() @pytest.mark.asyncio async def test_delete_extraction_not_found(self, extraction_service, test_user): """Test deleting non-existent extraction.""" extraction_service.extraction_repo.get_by_id = AsyncMock(return_value=None) result = await extraction_service.delete_extraction(999, test_user.id) assert result is False @pytest.mark.asyncio async def test_delete_extraction_permission_denied(self, extraction_service, test_user): """Test deleting extraction owned by another user.""" extraction = Extraction( id=1, url="https://example.com/video", user_id=999, # Different user ID status="completed", ) extraction_service.extraction_repo.get_by_id = AsyncMock(return_value=extraction) with pytest.raises(ValueError, match="You don't have permission"): await extraction_service.delete_extraction(1, test_user.id) @pytest.mark.asyncio async def test_delete_extraction_admin(self, extraction_service, test_user): """Test admin deleting any extraction.""" extraction = Extraction( id=1, url="https://example.com/video", user_id=999, # Different user ID status="completed", ) extraction_service.extraction_repo.get_by_id = AsyncMock(return_value=extraction) extraction_service.extraction_repo.delete = AsyncMock() extraction_service.session.commit = AsyncMock() # Admin deletion (user_id=None) result = await extraction_service.delete_extraction(1, None) assert result is True extraction_service.extraction_repo.delete.assert_called_once_with(extraction)