"""Tests for admin sound API endpoints.""" from typing import TYPE_CHECKING from unittest.mock import patch import pytest from httpx import ASGITransport, AsyncClient from app.models.user import User if TYPE_CHECKING: from app.services.sound_normalizer import NormalizationResults from app.services.sound_scanner import ScanResults class TestAdminSoundEndpoints: """Test admin sound API endpoints.""" @pytest.mark.asyncio async def test_scan_sounds_success( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """Test successful sound scanning.""" # Mock the scanner service to return successful results mock_results: ScanResults = { "scanned": 5, "added": 3, "updated": 1, "deleted": 1, "skipped": 0, "errors": 0, "duplicates": 0, "files": [ { "filename": "test1.mp3", "status": "added", "reason": None, "name": "Test1", "duration": 5000, "size": 1024, "id": 1, "error": None, "changes": None, }, { "filename": "test2.mp3", "status": "updated", "reason": "file was modified", "name": "Test2", "duration": 7500, "size": 2048, "id": 2, "error": None, "changes": ["hash", "duration", "size"], }, { "filename": "test3.mp3", "status": "deleted", "reason": "file no longer exists", "name": "Test3", "duration": 3000, "size": 512, "id": 3, "error": None, "changes": None, }, ], } with patch( "app.services.sound_scanner.SoundScannerService.scan_soundboard_directory", ) as mock_scan: mock_scan.return_value = mock_results response = await authenticated_admin_client.post( "/api/v1/admin/sounds/scan", ) assert response.status_code == 200 data = response.json() assert "message" in data assert "Sound sync completed" in data["message"] assert "results" in data results = data["results"] assert results["scanned"] == 5 assert results["added"] == 3 assert results["updated"] == 1 assert results["deleted"] == 1 assert results["skipped"] == 0 assert results["errors"] == 0 assert len(results["files"]) == 3 # Check file details assert results["files"][0]["filename"] == "test1.mp3" assert results["files"][0]["status"] == "added" assert results["files"][1]["status"] == "updated" assert results["files"][2]["status"] == "deleted" @pytest.mark.asyncio async def test_scan_sounds_unauthenticated(self, client: AsyncClient) -> None: """Test scanning sounds without authentication.""" response = await client.post("/api/v1/admin/sounds/scan") assert response.status_code == 401 data = response.json() assert "Could not validate credentials" in data["detail"] @pytest.mark.asyncio async def test_scan_sounds_non_admin( self, test_app, test_user: User, ) -> None: """Test scanning sounds with non-admin user.""" from fastapi import HTTPException from app.core.dependencies import get_admin_user # Override the admin dependency to raise 403 for non-admin users async def override_get_admin_user(): raise HTTPException(status_code=403, detail="Not enough permissions") test_app.dependency_overrides[get_admin_user] = override_get_admin_user # Create API token for regular 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/admin/sounds/scan", headers=headers) assert response.status_code == 403 data = response.json() assert "Not enough permissions" in data["detail"] # Clean up override test_app.dependency_overrides.pop(get_admin_user, None) @pytest.mark.asyncio async def test_scan_sounds_service_error( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """Test scanning sounds when service raises an error.""" with patch( "app.services.sound_scanner.SoundScannerService.scan_soundboard_directory", ) as mock_scan: mock_scan.side_effect = Exception("Directory not found") response = await authenticated_admin_client.post( "/api/v1/admin/sounds/scan", ) assert response.status_code == 500 data = response.json() assert "Failed to sync sounds" in data["detail"] assert "Directory not found" in data["detail"] @pytest.mark.asyncio async def test_scan_custom_directory_success( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """Test successful custom directory scanning.""" mock_results: ScanResults = { "scanned": 2, "added": 2, "updated": 0, "deleted": 0, "skipped": 0, "errors": 0, "duplicates": 0, "files": [ { "filename": "custom1.wav", "status": "added", "reason": None, "name": "Custom1", "duration": 4000, "size": 800, "id": 10, "error": None, "changes": None, }, { "filename": "custom2.wav", "status": "added", "reason": None, "name": "Custom2", "duration": 6000, "size": 1200, "id": 11, "error": None, "changes": None, }, ], } with patch( "app.services.sound_scanner.SoundScannerService.scan_directory", ) as mock_scan: mock_scan.return_value = mock_results response = await authenticated_admin_client.post( "/api/v1/admin/sounds/scan/custom", params={"directory": "/custom/path", "sound_type": "CUSTOM"}, ) assert response.status_code == 200 data = response.json() assert "Sync of directory '/custom/path' completed" in data["message"] assert "results" in data results = data["results"] assert results["scanned"] == 2 assert results["added"] == 2 assert len(results["files"]) == 2 # Verify the service was called with correct parameters mock_scan.assert_called_once_with("/custom/path", "CUSTOM") @pytest.mark.asyncio async def test_normalize_all_sounds_success( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """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/admin/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_unauthenticated( self, client: AsyncClient, ) -> None: """Test normalizing sounds without authentication.""" response = await client.post("/api/v1/admin/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, ) -> None: """Test normalizing sounds with non-admin user.""" from fastapi import HTTPException from app.core.dependencies import get_admin_user # Override the admin dependency to raise 403 for non-admin users async def override_get_admin_user(): raise HTTPException(status_code=403, detail="Not enough permissions") test_app.dependency_overrides[get_admin_user] = override_get_admin_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/admin/sounds/normalize/all", headers=headers, ) assert response.status_code == 403 data = response.json() assert "Not enough permissions" in data["detail"] # Clean up override test_app.dependency_overrides.pop(get_admin_user, None) @pytest.mark.asyncio async def test_normalize_sounds_by_type_success( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """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/admin/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, ) -> None: """Test normalization with invalid sound type.""" response = await authenticated_admin_client.post( "/api/v1/admin/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_sound_by_id_success( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """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/admin/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_get_extraction_processor_status( self, authenticated_admin_client: AsyncClient, admin_user: User, ) -> None: """Test getting extraction processor status.""" with patch( "app.services.extraction_processor.extraction_processor.get_status", ) as mock_get_status: mock_status = { "is_running": True, "queue_size": 2, "active_extractions": 1, "max_concurrent": 2, } mock_get_status.return_value = mock_status response = await authenticated_admin_client.get( "/api/v1/admin/extractions/status", ) assert response.status_code == 200 data = response.json() assert data == mock_status @pytest.mark.asyncio async def test_get_extraction_processor_status_unauthenticated( self, client: AsyncClient, ) -> None: """Test getting extraction processor status without authentication.""" response = await client.get("/api/v1/admin/extractions/status") assert response.status_code == 401 data = response.json() assert "Could not validate credentials" in data["detail"] @pytest.mark.asyncio async def test_get_extraction_processor_status_non_admin( self, test_app, test_user: User, ) -> None: """Test getting extraction processor status with non-admin user.""" from fastapi import HTTPException from app.core.dependencies import get_admin_user # Override the admin dependency to raise 403 for non-admin users async def override_get_admin_user(): raise HTTPException(status_code=403, detail="Not enough permissions") test_app.dependency_overrides[get_admin_user] = override_get_admin_user headers = {"API-TOKEN": "test_api_token"} async with AsyncClient( transport=ASGITransport(app=test_app), base_url="http://test", ) as client: response = await client.get( "/api/v1/admin/extractions/status", headers=headers, ) assert response.status_code == 403 data = response.json() assert "Not enough permissions" in data["detail"] # Clean up override test_app.dependency_overrides.pop(get_admin_user, None)