1117 lines
38 KiB
Python
1117 lines
38 KiB
Python
"""Tests for 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 TestSoundEndpoints:
|
|
"""Test 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,
|
|
"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) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> 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/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,
|
|
"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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> 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/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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> 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/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/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,
|
|
) -> None:
|
|
"""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,
|
|
) -> 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/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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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"]
|
|
assert call_args[1]["one_pass"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normalize_sound_by_id_skipped(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
) -> None:
|
|
"""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) -> None:
|
|
"""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,
|
|
) -> None:
|
|
"""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)
|