34e6289f92
- Implemented endpoints for normalizing all sounds, normalizing sounds by type, and normalizing a specific sound by ID in the sounds API. - Added dependency injection for the SoundNormalizerService. - Included role-based access control to restrict normalization actions to admin users. - Created comprehensive test cases for the new normalization endpoints, covering success scenarios, parameter handling, and error cases. - Removed redundant test file for sound normalizer endpoints as tests are now integrated into the main sound endpoints test suite.
1118 lines
37 KiB
Python
1118 lines
37 KiB
Python
"""Tests for sound 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
|
|
from app.services.sound_scanner import ScanResults
|
|
|
|
|
|
class TestSoundEndpoints:
|
|
"""Test sound API endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_sounds_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""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,
|
|
"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/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):
|
|
"""Test scanning sounds without authentication."""
|
|
response = await client.post("/api/v1/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,
|
|
):
|
|
"""Test scanning 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
|
|
)
|
|
|
|
# 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/sounds/scan", headers=headers)
|
|
|
|
assert response.status_code == 403
|
|
data = response.json()
|
|
assert "Only administrators can sync sounds" in data["detail"]
|
|
|
|
# Clean up override
|
|
test_app.dependency_overrides.pop(get_current_active_user_flexible, None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_sounds_admin_user(
|
|
self,
|
|
test_app,
|
|
admin_user: User,
|
|
):
|
|
"""Test scanning sounds with admin user."""
|
|
from app.core.dependencies import get_current_active_user_flexible
|
|
|
|
mock_results: ScanResults = {
|
|
"scanned": 1,
|
|
"added": 1,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"skipped": 0,
|
|
"errors": 0,
|
|
"files": [],
|
|
}
|
|
|
|
# Override the dependency to return admin user
|
|
async def override_get_current_user():
|
|
return admin_user
|
|
|
|
test_app.dependency_overrides[get_current_active_user_flexible] = (
|
|
override_get_current_user
|
|
)
|
|
|
|
headers = {"API-TOKEN": "admin_api_token"}
|
|
|
|
with patch(
|
|
"app.services.sound_scanner.SoundScannerService.scan_soundboard_directory"
|
|
) as mock_scan:
|
|
mock_scan.return_value = mock_results
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=test_app),
|
|
base_url="http://test",
|
|
) as client:
|
|
response = await client.post("/api/v1/sounds/scan", headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "Sound sync completed" in data["message"]
|
|
|
|
# Clean up override
|
|
test_app.dependency_overrides.pop(get_current_active_user_flexible, None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_sounds_service_error(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""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/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,
|
|
):
|
|
"""Test successful custom directory scanning."""
|
|
mock_results: ScanResults = {
|
|
"scanned": 2,
|
|
"added": 2,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"skipped": 0,
|
|
"errors": 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/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_scan_custom_directory_default_type(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""Test custom directory scanning with default sound type."""
|
|
mock_results: ScanResults = {
|
|
"scanned": 1,
|
|
"added": 1,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"skipped": 0,
|
|
"errors": 0,
|
|
"files": [],
|
|
}
|
|
|
|
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/sounds/scan/custom",
|
|
params={"directory": "/another/path"}, # No sound_type param
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Verify the service was called with default SDB type
|
|
mock_scan.assert_called_once_with("/another/path", "SDB")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_custom_directory_invalid_path(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""Test custom directory scanning with invalid path."""
|
|
with patch(
|
|
"app.services.sound_scanner.SoundScannerService.scan_directory"
|
|
) as mock_scan:
|
|
mock_scan.side_effect = ValueError(
|
|
"Directory does not exist: /invalid/path"
|
|
)
|
|
|
|
response = await authenticated_admin_client.post(
|
|
"/api/v1/sounds/scan/custom", params={"directory": "/invalid/path"}
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert "Directory does not exist" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_custom_directory_unauthenticated(self, client: AsyncClient):
|
|
"""Test custom directory scanning without authentication."""
|
|
response = await client.post(
|
|
"/api/v1/sounds/scan/custom", params={"directory": "/some/path"}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert "Could not validate credentials" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_custom_directory_non_admin(
|
|
self,
|
|
test_app,
|
|
test_user: User,
|
|
):
|
|
"""Test custom directory scanning 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/scan/custom",
|
|
headers=headers,
|
|
params={"directory": "/some/path"},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
data = response.json()
|
|
assert "Only administrators can sync sounds" in data["detail"]
|
|
|
|
# Clean up override
|
|
test_app.dependency_overrides.pop(get_current_active_user_flexible, None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_custom_directory_service_error(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""Test custom directory scanning when service raises an error."""
|
|
with patch(
|
|
"app.services.sound_scanner.SoundScannerService.scan_directory"
|
|
) as mock_scan:
|
|
mock_scan.side_effect = Exception("Permission denied")
|
|
|
|
response = await authenticated_admin_client.post(
|
|
"/api/v1/sounds/scan/custom", params={"directory": "/restricted/path"}
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
data = response.json()
|
|
assert "Failed to sync directory" in data["detail"]
|
|
assert "Permission denied" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_results_with_errors(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""Test scanning with some errors in results."""
|
|
mock_results: ScanResults = {
|
|
"scanned": 3,
|
|
"added": 1,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"skipped": 1,
|
|
"errors": 1,
|
|
"files": [
|
|
{
|
|
"filename": "good.mp3",
|
|
"status": "added",
|
|
"reason": None,
|
|
"name": "Good",
|
|
"duration": 5000,
|
|
"size": 1024,
|
|
"id": 1,
|
|
"error": None,
|
|
"changes": None,
|
|
},
|
|
{
|
|
"filename": "unchanged.mp3",
|
|
"status": "skipped",
|
|
"reason": "file unchanged",
|
|
"name": "Unchanged",
|
|
"duration": 3000,
|
|
"size": 512,
|
|
"id": 2,
|
|
"error": None,
|
|
"changes": None,
|
|
},
|
|
{
|
|
"filename": "bad.mp3",
|
|
"status": "error",
|
|
"reason": None,
|
|
"name": None,
|
|
"duration": None,
|
|
"size": None,
|
|
"id": None,
|
|
"error": "Invalid audio format",
|
|
"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/sounds/scan")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
results = data["results"]
|
|
assert results["errors"] == 1
|
|
assert results["added"] == 1
|
|
assert results["skipped"] == 1
|
|
|
|
# Check error file details
|
|
error_file = next(f for f in results["files"] if f["status"] == "error")
|
|
assert error_file["filename"] == "bad.mp3"
|
|
assert error_file["error"] == "Invalid audio format"
|
|
assert error_file["id"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_endpoint_response_structure(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
):
|
|
"""Test that endpoint response has correct structure."""
|
|
mock_results: ScanResults = {
|
|
"scanned": 0,
|
|
"added": 0,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"skipped": 0,
|
|
"errors": 0,
|
|
"files": [],
|
|
}
|
|
|
|
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/sounds/scan")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Check top-level structure
|
|
assert isinstance(data, dict)
|
|
assert "message" in data
|
|
assert "results" in data
|
|
assert isinstance(data["message"], str)
|
|
assert isinstance(data["results"], dict)
|
|
|
|
# Check results structure
|
|
results = data["results"]
|
|
required_fields = [
|
|
"scanned",
|
|
"added",
|
|
"updated",
|
|
"deleted",
|
|
"skipped",
|
|
"errors",
|
|
"files",
|
|
]
|
|
for field in required_fields:
|
|
assert field in results
|
|
if field == "files":
|
|
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)
|