diff --git a/app/api/v1/__init__.py b/app/api/v1/__init__.py index 8d2638e..3933b1d 100644 --- a/app/api/v1/__init__.py +++ b/app/api/v1/__init__.py @@ -2,7 +2,7 @@ from fastapi import APIRouter -from app.api.v1 import auth, main, socket, sound_normalizer, sounds +from app.api.v1 import auth, main, socket, sounds # V1 API router with v1 prefix api_router = APIRouter(prefix="/v1") @@ -12,4 +12,3 @@ api_router.include_router(main.router, tags=["main"]) api_router.include_router(auth.router, prefix="/auth", tags=["authentication"]) api_router.include_router(socket.router, tags=["socket"]) api_router.include_router(sounds.router, tags=["sounds"]) -api_router.include_router(sound_normalizer.router, tags=["sound-normalization"]) diff --git a/app/api/v1/sound_normalizer.py b/app/api/v1/sound_normalizer.py deleted file mode 100644 index 17f98bb..0000000 --- a/app/api/v1/sound_normalizer.py +++ /dev/null @@ -1,166 +0,0 @@ -"""Sound normalization API endpoints.""" - -from typing import Annotated - -from fastapi import APIRouter, Depends, HTTPException, Query, status -from sqlmodel.ext.asyncio.session import AsyncSession - -from app.core.database import get_db -from app.core.dependencies import get_current_active_user_flexible -from app.models.user import User -from app.services.sound_normalizer import NormalizationResults, SoundNormalizerService - -router = APIRouter(prefix="/sounds/normalize", tags=["sound-normalization"]) - - -async def get_sound_normalizer_service( - session: Annotated[AsyncSession, Depends(get_db)], -) -> SoundNormalizerService: - """Get the sound normalizer service.""" - return SoundNormalizerService(session) - - -@router.post("/all") -async def normalize_all_sounds( - current_user: Annotated[User, Depends(get_current_active_user_flexible)], - normalizer_service: Annotated[ - SoundNormalizerService, Depends(get_sound_normalizer_service) - ], - force: bool = Query( - False, description="Force normalization of already normalized sounds" - ), - one_pass: bool | None = Query( - None, description="Use one-pass normalization (overrides config)" - ), -) -> dict[str, NormalizationResults | str]: - """Normalize all unnormalized sounds.""" - # Only allow admins to normalize sounds - if current_user.role not in ["admin", "superadmin"]: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Only administrators can normalize sounds", - ) - - try: - results = await normalizer_service.normalize_all_sounds( - force=force, - one_pass=one_pass, - ) - return { - "message": "Sound normalization completed", - "results": results, - } - except Exception as e: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to normalize sounds: {e!s}", - ) from e - - -@router.post("/type/{sound_type}") -async def normalize_sounds_by_type( - sound_type: str, - current_user: Annotated[User, Depends(get_current_active_user_flexible)], - normalizer_service: Annotated[ - SoundNormalizerService, Depends(get_sound_normalizer_service) - ], - force: bool = Query( - False, description="Force normalization of already normalized sounds" - ), - one_pass: bool | None = Query( - None, description="Use one-pass normalization (overrides config)" - ), -) -> dict[str, NormalizationResults | str]: - """Normalize all sounds of a specific type (SDB, TTS, EXT).""" - # Only allow admins to normalize sounds - if current_user.role not in ["admin", "superadmin"]: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Only administrators can normalize sounds", - ) - - # Validate sound type - valid_types = ["SDB", "TTS", "EXT"] - if sound_type not in valid_types: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Invalid sound type. Must be one of: {', '.join(valid_types)}", - ) - - try: - results = await normalizer_service.normalize_sounds_by_type( - sound_type=sound_type, - force=force, - one_pass=one_pass, - ) - return { - "message": f"Normalization of {sound_type} sounds completed", - "results": results, - } - except Exception as e: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to normalize {sound_type} sounds: {e!s}", - ) from e - - -@router.post("/{sound_id}") -async def normalize_sound_by_id( - sound_id: int, - current_user: Annotated[User, Depends(get_current_active_user_flexible)], - normalizer_service: Annotated[ - SoundNormalizerService, Depends(get_sound_normalizer_service) - ], - force: bool = Query( - False, description="Force normalization of already normalized sound" - ), - one_pass: bool | None = Query( - None, description="Use one-pass normalization (overrides config)" - ), -) -> dict[str, str]: - """Normalize a specific sound by ID.""" - # Only allow admins to normalize sounds - if current_user.role not in ["admin", "superadmin"]: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Only administrators can normalize sounds", - ) - - try: - # Get the sound - sound = await normalizer_service.sound_repo.get_by_id(sound_id) - if not sound: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"Sound with ID {sound_id} not found", - ) - - # Normalize the sound - result = await normalizer_service.normalize_sound( - sound=sound, - force=force, - one_pass=one_pass, - ) - - # Check result status - if result["status"] == "error": - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to normalize sound: {result['error']}", - ) - - return { - "message": f"Sound normalization {result['status']}: {sound.filename}", - "status": result["status"], - "reason": result["reason"] or "", - "normalized_filename": result["normalized_filename"] or "", - } - - except HTTPException: - # Re-raise HTTPExceptions without wrapping them - raise - except Exception as e: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to normalize sound: {e!s}", - ) from e diff --git a/app/api/v1/sounds.py b/app/api/v1/sounds.py index 5c25157..fe301fe 100644 --- a/app/api/v1/sounds.py +++ b/app/api/v1/sounds.py @@ -2,12 +2,13 @@ from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlmodel.ext.asyncio.session import AsyncSession from app.core.database import get_db from app.core.dependencies import get_current_active_user_flexible from app.models.user import User +from app.services.sound_normalizer import NormalizationResults, SoundNormalizerService from app.services.sound_scanner import ScanResults, SoundScannerService router = APIRouter(prefix="/sounds", tags=["sounds"]) @@ -20,6 +21,14 @@ async def get_sound_scanner_service( return SoundScannerService(session) +async def get_sound_normalizer_service( + session: Annotated[AsyncSession, Depends(get_db)], +) -> SoundNormalizerService: + """Get the sound normalizer service.""" + return SoundNormalizerService(session) + + +# SCAN @router.post("/scan") async def scan_sounds( current_user: Annotated[User, Depends(get_current_active_user_flexible)], @@ -77,3 +86,150 @@ async def scan_custom_directory( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to sync directory: {e!s}", ) from e + + +# NORMALIZE +@router.post("/normalize/all") +async def normalize_all_sounds( + current_user: Annotated[User, Depends(get_current_active_user_flexible)], + normalizer_service: Annotated[ + SoundNormalizerService, Depends(get_sound_normalizer_service) + ], + force: bool = Query( + False, description="Force normalization of already normalized sounds" + ), + one_pass: bool | None = Query( + None, description="Use one-pass normalization (overrides config)" + ), +) -> dict[str, NormalizationResults | str]: + """Normalize all unnormalized sounds.""" + # Only allow admins to normalize sounds + if current_user.role not in ["admin", "superadmin"]: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only administrators can normalize sounds", + ) + + try: + results = await normalizer_service.normalize_all_sounds( + force=force, + one_pass=one_pass, + ) + return { + "message": "Sound normalization completed", + "results": results, + } + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to normalize sounds: {e!s}", + ) from e + + +@router.post("/normalize/type/{sound_type}") +async def normalize_sounds_by_type( + sound_type: str, + current_user: Annotated[User, Depends(get_current_active_user_flexible)], + normalizer_service: Annotated[ + SoundNormalizerService, Depends(get_sound_normalizer_service) + ], + force: bool = Query( + False, description="Force normalization of already normalized sounds" + ), + one_pass: bool | None = Query( + None, description="Use one-pass normalization (overrides config)" + ), +) -> dict[str, NormalizationResults | str]: + """Normalize all sounds of a specific type (SDB, TTS, EXT).""" + # Only allow admins to normalize sounds + if current_user.role not in ["admin", "superadmin"]: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only administrators can normalize sounds", + ) + + # Validate sound type + valid_types = ["SDB", "TTS", "EXT"] + if sound_type not in valid_types: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid sound type. Must be one of: {', '.join(valid_types)}", + ) + + try: + results = await normalizer_service.normalize_sounds_by_type( + sound_type=sound_type, + force=force, + one_pass=one_pass, + ) + return { + "message": f"Normalization of {sound_type} sounds completed", + "results": results, + } + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to normalize {sound_type} sounds: {e!s}", + ) from e + + +@router.post("/normalize/{sound_id}") +async def normalize_sound_by_id( + sound_id: int, + current_user: Annotated[User, Depends(get_current_active_user_flexible)], + normalizer_service: Annotated[ + SoundNormalizerService, Depends(get_sound_normalizer_service) + ], + force: bool = Query( + False, description="Force normalization of already normalized sound" + ), + one_pass: bool | None = Query( + None, description="Use one-pass normalization (overrides config)" + ), +) -> dict[str, str]: + """Normalize a specific sound by ID.""" + # Only allow admins to normalize sounds + if current_user.role not in ["admin", "superadmin"]: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only administrators can normalize sounds", + ) + + try: + # Get the sound + sound = await normalizer_service.sound_repo.get_by_id(sound_id) + if not sound: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Sound with ID {sound_id} not found", + ) + + # Normalize the sound + result = await normalizer_service.normalize_sound( + sound=sound, + force=force, + one_pass=one_pass, + ) + + # Check result status + if result["status"] == "error": + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to normalize sound: {result['error']}", + ) + + return { + "message": f"Sound normalization {result['status']}: {sound.filename}", + "status": result["status"], + "reason": result["reason"] or "", + "normalized_filename": result["normalized_filename"] or "", + } + + except HTTPException: + # Re-raise HTTPExceptions without wrapping them + raise + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to normalize sound: {e!s}", + ) from e diff --git a/tests/api/v1/test_sound_endpoints.py b/tests/api/v1/test_sound_endpoints.py index 2c37cc3..bef91d0 100644 --- a/tests/api/v1/test_sound_endpoints.py +++ b/tests/api/v1/test_sound_endpoints.py @@ -6,6 +6,7 @@ import pytest from httpx import ASGITransport, AsyncClient from app.models.user import User +from app.services.sound_normalizer import NormalizationResults from app.services.sound_scanner import ScanResults @@ -512,3 +513,605 @@ class TestSoundEndpoints: assert isinstance(results[field], list) else: assert isinstance(results[field], int) + + @pytest.mark.asyncio + async def test_normalize_all_sounds_success( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test successful normalization of all sounds.""" + mock_results: NormalizationResults = { + "processed": 3, + "normalized": 2, + "skipped": 1, + "errors": 0, + "files": [ + { + "filename": "test1.mp3", + "status": "normalized", + "reason": None, + "original_path": "/fake/test1.mp3", + "normalized_path": "/fake/test1_normalized.mp3", + "normalized_filename": "test1_normalized.mp3", + "normalized_duration": 5000, + "normalized_size": 1024, + "normalized_hash": "norm_hash1", + "id": 1, + "error": None, + }, + { + "filename": "test2.wav", + "status": "normalized", + "reason": None, + "original_path": "/fake/test2.wav", + "normalized_path": "/fake/test2_normalized.mp3", + "normalized_filename": "test2_normalized.mp3", + "normalized_duration": 7000, + "normalized_size": 2048, + "normalized_hash": "norm_hash2", + "id": 2, + "error": None, + }, + { + "filename": "test3.mp3", + "status": "skipped", + "reason": "already normalized", + "original_path": None, + "normalized_path": None, + "normalized_filename": None, + "normalized_duration": None, + "normalized_size": None, + "normalized_hash": None, + "id": 3, + "error": None, + }, + ], + } + + with patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" + ) as mock_normalize: + mock_normalize.return_value = mock_results + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/all" + ) + + assert response.status_code == 200 + data = response.json() + + assert "message" in data + assert "Sound normalization completed" in data["message"] + assert "results" in data + + results = data["results"] + assert results["processed"] == 3 + assert results["normalized"] == 2 + assert results["skipped"] == 1 + assert results["errors"] == 0 + assert len(results["files"]) == 3 + + @pytest.mark.asyncio + async def test_normalize_all_sounds_with_force( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization with force parameter.""" + mock_results: NormalizationResults = { + "processed": 1, + "normalized": 1, + "skipped": 0, + "errors": 0, + "files": [], + } + + with patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" + ) as mock_normalize: + mock_normalize.return_value = mock_results + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/all", params={"force": True} + ) + + assert response.status_code == 200 + + # Verify force parameter was passed + mock_normalize.assert_called_once_with(force=True, one_pass=None) + + @pytest.mark.asyncio + async def test_normalize_all_sounds_with_one_pass( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization with one_pass parameter.""" + mock_results: NormalizationResults = { + "processed": 1, + "normalized": 1, + "skipped": 0, + "errors": 0, + "files": [], + } + + with patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" + ) as mock_normalize: + mock_normalize.return_value = mock_results + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/all", params={"one_pass": True} + ) + + assert response.status_code == 200 + + # Verify one_pass parameter was passed + mock_normalize.assert_called_once_with(force=False, one_pass=True) + + @pytest.mark.asyncio + async def test_normalize_all_sounds_unauthenticated(self, client: AsyncClient): + """Test normalizing sounds without authentication.""" + response = await client.post("/api/v1/sounds/normalize/all") + + assert response.status_code == 401 + data = response.json() + assert "Could not validate credentials" in data["detail"] + + @pytest.mark.asyncio + async def test_normalize_all_sounds_non_admin( + self, + test_app, + test_user: User, + ): + """Test normalizing sounds with non-admin user.""" + from app.core.dependencies import get_current_active_user_flexible + + # Override the dependency to return regular user + async def override_get_current_user(): + test_user.role = "user" + return test_user + + test_app.dependency_overrides[get_current_active_user_flexible] = ( + override_get_current_user + ) + + headers = {"API-TOKEN": "test_api_token"} + + async with AsyncClient( + transport=ASGITransport(app=test_app), + base_url="http://test", + ) as client: + response = await client.post( + "/api/v1/sounds/normalize/all", headers=headers + ) + + assert response.status_code == 403 + data = response.json() + assert "Only administrators can normalize sounds" in data["detail"] + + # Clean up override + test_app.dependency_overrides.pop(get_current_active_user_flexible, None) + + @pytest.mark.asyncio + async def test_normalize_all_sounds_service_error( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization when service raises an error.""" + with patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" + ) as mock_normalize: + mock_normalize.side_effect = Exception("Normalization service failed") + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/all" + ) + + assert response.status_code == 500 + data = response.json() + assert "Failed to normalize sounds" in data["detail"] + assert "Normalization service failed" in data["detail"] + + @pytest.mark.asyncio + async def test_normalize_sounds_by_type_success( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test successful normalization by sound type.""" + mock_results: NormalizationResults = { + "processed": 2, + "normalized": 2, + "skipped": 0, + "errors": 0, + "files": [ + { + "filename": "sdb1.mp3", + "status": "normalized", + "reason": None, + "original_path": "/fake/sdb1.mp3", + "normalized_path": "/fake/sdb1_normalized.mp3", + "normalized_filename": "sdb1_normalized.mp3", + "normalized_duration": 4000, + "normalized_size": 800, + "normalized_hash": "sdb_hash1", + "id": 10, + "error": None, + }, + { + "filename": "sdb2.wav", + "status": "normalized", + "reason": None, + "original_path": "/fake/sdb2.wav", + "normalized_path": "/fake/sdb2_normalized.mp3", + "normalized_filename": "sdb2_normalized.mp3", + "normalized_duration": 6000, + "normalized_size": 1200, + "normalized_hash": "sdb_hash2", + "id": 11, + "error": None, + }, + ], + } + + with patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_sounds_by_type" + ) as mock_normalize: + mock_normalize.return_value = mock_results + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/type/SDB" + ) + + assert response.status_code == 200 + data = response.json() + + assert "Normalization of SDB sounds completed" in data["message"] + assert "results" in data + + results = data["results"] + assert results["processed"] == 2 + assert results["normalized"] == 2 + assert len(results["files"]) == 2 + + # Verify the service was called with correct type + mock_normalize.assert_called_once_with( + sound_type="SDB", force=False, one_pass=None + ) + + @pytest.mark.asyncio + async def test_normalize_sounds_by_type_invalid_type( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization with invalid sound type.""" + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/type/INVALID" + ) + + assert response.status_code == 400 + data = response.json() + assert "Invalid sound type" in data["detail"] + assert "Must be one of: SDB, TTS, EXT" in data["detail"] + + @pytest.mark.asyncio + async def test_normalize_sounds_by_type_with_params( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization by type with force and one_pass parameters.""" + mock_results: NormalizationResults = { + "processed": 1, + "normalized": 1, + "skipped": 0, + "errors": 0, + "files": [], + } + + with patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_sounds_by_type" + ) as mock_normalize: + mock_normalize.return_value = mock_results + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/type/TTS", + params={"force": True, "one_pass": False}, + ) + + assert response.status_code == 200 + + # Verify parameters were passed correctly + mock_normalize.assert_called_once_with( + sound_type="TTS", force=True, one_pass=False + ) + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_success( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test successful normalization of a specific sound.""" + # Mock the sound + mock_sound = type( + "Sound", + (), + { + "id": 42, + "filename": "specific_sound.mp3", + "type": "SDB", + "name": "Specific Sound", + }, + )() + + # Mock normalization result + mock_result = { + "filename": "specific_sound.mp3", + "status": "normalized", + "reason": None, + "original_path": "/fake/specific_sound.mp3", + "normalized_path": "/fake/specific_sound_normalized.mp3", + "normalized_filename": "specific_sound_normalized.mp3", + "normalized_duration": 8000, + "normalized_size": 1600, + "normalized_hash": "specific_hash", + "id": 42, + "error": None, + } + + with ( + patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" + ) as mock_normalize_sound, + patch("app.repositories.sound.SoundRepository.get_by_id") as mock_get_sound, + ): + + mock_get_sound.return_value = mock_sound + mock_normalize_sound.return_value = mock_result + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/42" + ) + + assert response.status_code == 200 + data = response.json() + + assert "Sound normalization normalized" in data["message"] + assert "specific_sound.mp3" in data["message"] + assert data["status"] == "normalized" + assert data["normalized_filename"] == "specific_sound_normalized.mp3" + + # Verify sound was retrieved and normalized + mock_get_sound.assert_called_once_with(42) + mock_normalize_sound.assert_called_once() + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_not_found( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization of non-existent sound.""" + with patch( + "app.repositories.sound.SoundRepository.get_by_id" + ) as mock_get_sound: + mock_get_sound.return_value = None + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/999" + ) + + assert response.status_code == 404 + data = response.json() + assert "Sound with ID 999 not found" in data["detail"] + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_normalization_error( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization when the sound normalization fails.""" + # Mock the sound + mock_sound = type( + "Sound", + (), + { + "id": 42, + "filename": "error_sound.mp3", + "type": "SDB", + "name": "Error Sound", + }, + )() + + # Mock normalization error result + mock_result = { + "filename": "error_sound.mp3", + "status": "error", + "reason": None, + "original_path": "/fake/error_sound.mp3", + "normalized_path": None, + "normalized_filename": None, + "normalized_duration": None, + "normalized_size": None, + "normalized_hash": None, + "id": 42, + "error": "File format not supported", + } + + with ( + patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" + ) as mock_normalize_sound, + patch("app.repositories.sound.SoundRepository.get_by_id") as mock_get_sound, + ): + + mock_get_sound.return_value = mock_sound + mock_normalize_sound.return_value = mock_result + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/42" + ) + + assert response.status_code == 500 + data = response.json() + assert "Failed to normalize sound" in data["detail"] + assert "File format not supported" in data["detail"] + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_with_params( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test sound normalization with force and one_pass parameters.""" + # Mock the sound + mock_sound = type( + "Sound", + (), + { + "id": 42, + "filename": "param_sound.mp3", + "type": "SDB", + "name": "Param Sound", + }, + )() + + # Mock normalization result + mock_result = { + "filename": "param_sound.mp3", + "status": "normalized", + "reason": None, + "original_path": "/fake/param_sound.mp3", + "normalized_path": "/fake/param_sound_normalized.mp3", + "normalized_filename": "param_sound_normalized.mp3", + "normalized_duration": 5000, + "normalized_size": 1000, + "normalized_hash": "param_hash", + "id": 42, + "error": None, + } + + with ( + patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" + ) as mock_normalize_sound, + patch("app.repositories.sound.SoundRepository.get_by_id") as mock_get_sound, + ): + + mock_get_sound.return_value = mock_sound + mock_normalize_sound.return_value = mock_result + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/42", + params={"force": True, "one_pass": True}, + ) + + assert response.status_code == 200 + + # Verify parameters were passed to normalize_sound + call_args = mock_normalize_sound.call_args + assert call_args[1]["force"] == True + assert call_args[1]["one_pass"] == True + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_skipped( + self, + authenticated_admin_client: AsyncClient, + admin_user: User, + ): + """Test normalization when sound is already normalized and not forced.""" + # Mock the sound + mock_sound = type( + "Sound", + (), + { + "id": 42, + "filename": "already_normalized.mp3", + "type": "SDB", + "name": "Already Normalized", + }, + )() + + # Mock skipped result + mock_result = { + "filename": "already_normalized.mp3", + "status": "skipped", + "reason": "already normalized", + "original_path": None, + "normalized_path": None, + "normalized_filename": None, + "normalized_duration": None, + "normalized_size": None, + "normalized_hash": None, + "id": 42, + "error": None, + } + + with ( + patch( + "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" + ) as mock_normalize_sound, + patch("app.repositories.sound.SoundRepository.get_by_id") as mock_get_sound, + ): + + mock_get_sound.return_value = mock_sound + mock_normalize_sound.return_value = mock_result + + response = await authenticated_admin_client.post( + "/api/v1/sounds/normalize/42" + ) + + assert response.status_code == 200 + data = response.json() + + assert "Sound normalization skipped" in data["message"] + assert data["status"] == "skipped" + assert data["reason"] == "already normalized" + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_unauthenticated(self, client: AsyncClient): + """Test normalizing a specific sound without authentication.""" + response = await client.post("/api/v1/sounds/normalize/42") + + assert response.status_code == 401 + data = response.json() + assert "Could not validate credentials" in data["detail"] + + @pytest.mark.asyncio + async def test_normalize_sound_by_id_non_admin( + self, + test_app, + test_user: User, + ): + """Test normalizing a specific sound with non-admin user.""" + from app.core.dependencies import get_current_active_user_flexible + + # Override the dependency to return regular user + async def override_get_current_user(): + test_user.role = "user" + return test_user + + test_app.dependency_overrides[get_current_active_user_flexible] = ( + override_get_current_user + ) + + headers = {"API-TOKEN": "test_api_token"} + + async with AsyncClient( + transport=ASGITransport(app=test_app), + base_url="http://test", + ) as client: + response = await client.post("/api/v1/sounds/normalize/42", headers=headers) + + assert response.status_code == 403 + data = response.json() + assert "Only administrators can normalize sounds" in data["detail"] + + # Clean up override + test_app.dependency_overrides.pop(get_current_active_user_flexible, None) diff --git a/tests/api/v1/test_sound_normalizer_endpoints.py b/tests/api/v1/test_sound_normalizer_endpoints.py deleted file mode 100644 index 7f240ca..0000000 --- a/tests/api/v1/test_sound_normalizer_endpoints.py +++ /dev/null @@ -1,613 +0,0 @@ -"""Tests for sound normalizer API endpoints.""" - -from unittest.mock import patch - -import pytest -from httpx import ASGITransport, AsyncClient - -from app.models.user import User -from app.services.sound_normalizer import NormalizationResults - - -class TestSoundNormalizerEndpoints: - """Test sound normalizer API endpoints.""" - - @pytest.mark.asyncio - async def test_normalize_all_sounds_success( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test successful normalization of all sounds.""" - mock_results: NormalizationResults = { - "processed": 3, - "normalized": 2, - "skipped": 1, - "errors": 0, - "files": [ - { - "filename": "test1.mp3", - "status": "normalized", - "reason": None, - "original_path": "/fake/test1.mp3", - "normalized_path": "/fake/test1_normalized.mp3", - "normalized_filename": "test1_normalized.mp3", - "normalized_duration": 5000, - "normalized_size": 1024, - "normalized_hash": "norm_hash1", - "id": 1, - "error": None, - }, - { - "filename": "test2.wav", - "status": "normalized", - "reason": None, - "original_path": "/fake/test2.wav", - "normalized_path": "/fake/test2_normalized.mp3", - "normalized_filename": "test2_normalized.mp3", - "normalized_duration": 7000, - "normalized_size": 2048, - "normalized_hash": "norm_hash2", - "id": 2, - "error": None, - }, - { - "filename": "test3.mp3", - "status": "skipped", - "reason": "already normalized", - "original_path": None, - "normalized_path": None, - "normalized_filename": None, - "normalized_duration": None, - "normalized_size": None, - "normalized_hash": None, - "id": 3, - "error": None, - }, - ], - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" - ) as mock_normalize: - mock_normalize.return_value = mock_results - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/all" - ) - - assert response.status_code == 200 - data = response.json() - - assert "message" in data - assert "Sound normalization completed" in data["message"] - assert "results" in data - - results = data["results"] - assert results["processed"] == 3 - assert results["normalized"] == 2 - assert results["skipped"] == 1 - assert results["errors"] == 0 - assert len(results["files"]) == 3 - - @pytest.mark.asyncio - async def test_normalize_all_sounds_with_force( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization with force parameter.""" - mock_results: NormalizationResults = { - "processed": 1, - "normalized": 1, - "skipped": 0, - "errors": 0, - "files": [], - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" - ) as mock_normalize: - mock_normalize.return_value = mock_results - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/all", params={"force": True} - ) - - assert response.status_code == 200 - - # Verify force parameter was passed - mock_normalize.assert_called_once_with(force=True, one_pass=None) - - @pytest.mark.asyncio - async def test_normalize_all_sounds_with_one_pass( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization with one_pass parameter.""" - mock_results: NormalizationResults = { - "processed": 1, - "normalized": 1, - "skipped": 0, - "errors": 0, - "files": [], - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" - ) as mock_normalize: - mock_normalize.return_value = mock_results - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/all", params={"one_pass": True} - ) - - assert response.status_code == 200 - - # Verify one_pass parameter was passed - mock_normalize.assert_called_once_with(force=False, one_pass=True) - - @pytest.mark.asyncio - async def test_normalize_all_sounds_unauthenticated(self, client: AsyncClient): - """Test normalizing sounds without authentication.""" - response = await client.post("/api/v1/sounds/normalize/all") - - assert response.status_code == 401 - data = response.json() - assert "Could not validate credentials" in data["detail"] - - @pytest.mark.asyncio - async def test_normalize_all_sounds_non_admin( - self, - test_app, - test_user: User, - ): - """Test normalizing sounds with non-admin user.""" - from app.core.dependencies import get_current_active_user_flexible - - # Override the dependency to return regular user - async def override_get_current_user(): - test_user.role = "user" - return test_user - - test_app.dependency_overrides[get_current_active_user_flexible] = ( - override_get_current_user - ) - - headers = {"API-TOKEN": "test_api_token"} - - async with AsyncClient( - transport=ASGITransport(app=test_app), - base_url="http://test", - ) as client: - response = await client.post( - "/api/v1/sounds/normalize/all", headers=headers - ) - - assert response.status_code == 403 - data = response.json() - assert "Only administrators can normalize sounds" in data["detail"] - - # Clean up override - test_app.dependency_overrides.pop(get_current_active_user_flexible, None) - - @pytest.mark.asyncio - async def test_normalize_all_sounds_service_error( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization when service raises an error.""" - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds" - ) as mock_normalize: - mock_normalize.side_effect = Exception("Normalization service failed") - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/all" - ) - - assert response.status_code == 500 - data = response.json() - assert "Failed to normalize sounds" in data["detail"] - assert "Normalization service failed" in data["detail"] - - @pytest.mark.asyncio - async def test_normalize_sounds_by_type_success( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test successful normalization by sound type.""" - mock_results: NormalizationResults = { - "processed": 2, - "normalized": 2, - "skipped": 0, - "errors": 0, - "files": [ - { - "filename": "sdb1.mp3", - "status": "normalized", - "reason": None, - "original_path": "/fake/sdb1.mp3", - "normalized_path": "/fake/sdb1_normalized.mp3", - "normalized_filename": "sdb1_normalized.mp3", - "normalized_duration": 4000, - "normalized_size": 800, - "normalized_hash": "sdb_hash1", - "id": 10, - "error": None, - }, - { - "filename": "sdb2.wav", - "status": "normalized", - "reason": None, - "original_path": "/fake/sdb2.wav", - "normalized_path": "/fake/sdb2_normalized.mp3", - "normalized_filename": "sdb2_normalized.mp3", - "normalized_duration": 6000, - "normalized_size": 1200, - "normalized_hash": "sdb_hash2", - "id": 11, - "error": None, - }, - ], - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_sounds_by_type" - ) as mock_normalize: - mock_normalize.return_value = mock_results - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/type/SDB" - ) - - assert response.status_code == 200 - data = response.json() - - assert "Normalization of SDB sounds completed" in data["message"] - assert "results" in data - - results = data["results"] - assert results["processed"] == 2 - assert results["normalized"] == 2 - assert len(results["files"]) == 2 - - # Verify the service was called with correct type - mock_normalize.assert_called_once_with( - sound_type="SDB", force=False, one_pass=None - ) - - @pytest.mark.asyncio - async def test_normalize_sounds_by_type_invalid_type( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization with invalid sound type.""" - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/type/INVALID" - ) - - assert response.status_code == 400 - data = response.json() - assert "Invalid sound type" in data["detail"] - assert "Must be one of: SDB, TTS, EXT" in data["detail"] - - @pytest.mark.asyncio - async def test_normalize_sounds_by_type_with_params( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization by type with force and one_pass parameters.""" - mock_results: NormalizationResults = { - "processed": 1, - "normalized": 1, - "skipped": 0, - "errors": 0, - "files": [], - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_sounds_by_type" - ) as mock_normalize: - mock_normalize.return_value = mock_results - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/type/TTS", - params={"force": True, "one_pass": False}, - ) - - assert response.status_code == 200 - - # Verify parameters were passed correctly - mock_normalize.assert_called_once_with( - sound_type="TTS", force=True, one_pass=False - ) - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_success( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test successful normalization of a specific sound.""" - # Mock the sound - mock_sound = type( - "Sound", - (), - { - "id": 42, - "filename": "specific_sound.mp3", - "type": "SDB", - "name": "Specific Sound", - }, - )() - - # Mock normalization result - mock_result = { - "filename": "specific_sound.mp3", - "status": "normalized", - "reason": None, - "original_path": "/fake/specific_sound.mp3", - "normalized_path": "/fake/specific_sound_normalized.mp3", - "normalized_filename": "specific_sound_normalized.mp3", - "normalized_duration": 8000, - "normalized_size": 1600, - "normalized_hash": "specific_hash", - "id": 42, - "error": None, - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" - ) as mock_normalize_sound, patch( - "app.repositories.sound.SoundRepository.get_by_id" - ) as mock_get_sound: - - mock_get_sound.return_value = mock_sound - mock_normalize_sound.return_value = mock_result - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/42" - ) - - assert response.status_code == 200 - data = response.json() - - assert "Sound normalization normalized" in data["message"] - assert "specific_sound.mp3" in data["message"] - assert data["status"] == "normalized" - assert data["normalized_filename"] == "specific_sound_normalized.mp3" - - # Verify sound was retrieved and normalized - mock_get_sound.assert_called_once_with(42) - mock_normalize_sound.assert_called_once() - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_not_found( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization of non-existent sound.""" - with patch( - "app.repositories.sound.SoundRepository.get_by_id" - ) as mock_get_sound: - mock_get_sound.return_value = None - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/999" - ) - - assert response.status_code == 404 - data = response.json() - assert "Sound with ID 999 not found" in data["detail"] - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_normalization_error( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization when the sound normalization fails.""" - # Mock the sound - mock_sound = type( - "Sound", - (), - { - "id": 42, - "filename": "error_sound.mp3", - "type": "SDB", - "name": "Error Sound", - }, - )() - - # Mock normalization error result - mock_result = { - "filename": "error_sound.mp3", - "status": "error", - "reason": None, - "original_path": "/fake/error_sound.mp3", - "normalized_path": None, - "normalized_filename": None, - "normalized_duration": None, - "normalized_size": None, - "normalized_hash": None, - "id": 42, - "error": "File format not supported", - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" - ) as mock_normalize_sound, patch( - "app.repositories.sound.SoundRepository.get_by_id" - ) as mock_get_sound: - - mock_get_sound.return_value = mock_sound - mock_normalize_sound.return_value = mock_result - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/42" - ) - - assert response.status_code == 500 - data = response.json() - assert "Failed to normalize sound" in data["detail"] - assert "File format not supported" in data["detail"] - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_with_params( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test sound normalization with force and one_pass parameters.""" - # Mock the sound - mock_sound = type( - "Sound", - (), - { - "id": 42, - "filename": "param_sound.mp3", - "type": "SDB", - "name": "Param Sound", - }, - )() - - # Mock normalization result - mock_result = { - "filename": "param_sound.mp3", - "status": "normalized", - "reason": None, - "original_path": "/fake/param_sound.mp3", - "normalized_path": "/fake/param_sound_normalized.mp3", - "normalized_filename": "param_sound_normalized.mp3", - "normalized_duration": 5000, - "normalized_size": 1000, - "normalized_hash": "param_hash", - "id": 42, - "error": None, - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" - ) as mock_normalize_sound, patch( - "app.repositories.sound.SoundRepository.get_by_id" - ) as mock_get_sound: - - mock_get_sound.return_value = mock_sound - mock_normalize_sound.return_value = mock_result - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/42", - params={"force": True, "one_pass": True}, - ) - - assert response.status_code == 200 - - # Verify parameters were passed to normalize_sound - call_args = mock_normalize_sound.call_args - assert call_args[1]["force"] == True - assert call_args[1]["one_pass"] == True - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_skipped( - self, - authenticated_admin_client: AsyncClient, - admin_user: User, - ): - """Test normalization when sound is already normalized and not forced.""" - # Mock the sound - mock_sound = type( - "Sound", - (), - { - "id": 42, - "filename": "already_normalized.mp3", - "type": "SDB", - "name": "Already Normalized", - }, - )() - - # Mock skipped result - mock_result = { - "filename": "already_normalized.mp3", - "status": "skipped", - "reason": "already normalized", - "original_path": None, - "normalized_path": None, - "normalized_filename": None, - "normalized_duration": None, - "normalized_size": None, - "normalized_hash": None, - "id": 42, - "error": None, - } - - with patch( - "app.services.sound_normalizer.SoundNormalizerService.normalize_sound" - ) as mock_normalize_sound, patch( - "app.repositories.sound.SoundRepository.get_by_id" - ) as mock_get_sound: - - mock_get_sound.return_value = mock_sound - mock_normalize_sound.return_value = mock_result - - response = await authenticated_admin_client.post( - "/api/v1/sounds/normalize/42" - ) - - assert response.status_code == 200 - data = response.json() - - assert "Sound normalization skipped" in data["message"] - assert data["status"] == "skipped" - assert data["reason"] == "already normalized" - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_unauthenticated(self, client: AsyncClient): - """Test normalizing a specific sound without authentication.""" - response = await client.post("/api/v1/sounds/normalize/42") - - assert response.status_code == 401 - data = response.json() - assert "Could not validate credentials" in data["detail"] - - @pytest.mark.asyncio - async def test_normalize_sound_by_id_non_admin( - self, - test_app, - test_user: User, - ): - """Test normalizing a specific sound with non-admin user.""" - from app.core.dependencies import get_current_active_user_flexible - - # Override the dependency to return regular user - async def override_get_current_user(): - test_user.role = "user" - return test_user - - test_app.dependency_overrides[get_current_active_user_flexible] = ( - override_get_current_user - ) - - headers = {"API-TOKEN": "test_api_token"} - - async with AsyncClient( - transport=ASGITransport(app=test_app), - base_url="http://test", - ) as client: - response = await client.post( - "/api/v1/sounds/normalize/42", headers=headers - ) - - assert response.status_code == 403 - data = response.json() - assert "Only administrators can normalize sounds" in data["detail"] - - # Clean up override - test_app.dependency_overrides.pop(get_current_active_user_flexible, None) \ No newline at end of file