7dee6e320e
- Implement tests for admin extraction API endpoints including status retrieval, deletion of extractions, and permission checks. - Add tests for user extraction deletion, ensuring proper handling of permissions and non-existent extractions. - Enhance sound endpoint tests to include duplicate handling in responses. - Refactor favorite service tests to utilize mock dependencies for better maintainability and clarity. - Update sound scanner tests to improve file handling and ensure proper deletion of associated files.
572 lines
19 KiB
Python
572 lines
19 KiB
Python
"""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)
|