6068599a47
- Adjusted function signatures in various test files to enhance clarity by aligning parameters. - Updated patching syntax for better readability across test cases. - Improved formatting and spacing in test assertions and mock setups. - Ensured consistent use of async/await patterns in async test functions. - Enhanced comments for better understanding of test intentions.
309 lines
10 KiB
Python
309 lines
10 KiB
Python
"""Tests for sound API endpoints (non-admin endpoints only)."""
|
|
|
|
from typing import TYPE_CHECKING
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.models.user import User
|
|
|
|
if TYPE_CHECKING:
|
|
from app.services.extraction import ExtractionInfo
|
|
|
|
|
|
class TestSoundEndpoints:
|
|
"""Test sound API endpoints (non-admin only)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_extraction_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test successful extraction creation."""
|
|
mock_extraction_info: ExtractionInfo = {
|
|
"id": 1,
|
|
"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
|
|
"title": None,
|
|
"service": None,
|
|
"service_id": None,
|
|
"status": "pending",
|
|
"error": None,
|
|
"sound_id": None,
|
|
}
|
|
|
|
with (
|
|
patch(
|
|
"app.services.extraction.ExtractionService.create_extraction",
|
|
) as mock_create,
|
|
patch(
|
|
"app.services.extraction_processor.extraction_processor.queue_extraction",
|
|
) as mock_queue,
|
|
):
|
|
mock_create.return_value = mock_extraction_info
|
|
mock_queue.return_value = None
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/v1/sounds/extract",
|
|
params={"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "Extraction queued successfully"
|
|
assert data["extraction"]["id"] == 1
|
|
assert (
|
|
data["extraction"]["url"]
|
|
== "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_extraction_unauthenticated(self, client: AsyncClient) -> None:
|
|
"""Test extraction creation without authentication."""
|
|
response = await client.post(
|
|
"/api/v1/sounds/extract",
|
|
params={"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert "Could not validate credentials" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_extraction_invalid_url(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test extraction creation with invalid URL."""
|
|
with patch(
|
|
"app.services.extraction.ExtractionService.create_extraction",
|
|
) as mock_create:
|
|
mock_create.side_effect = ValueError("Invalid URL")
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/v1/sounds/extract",
|
|
params={"url": "invalid-url"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert "Invalid URL" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_extraction_by_id_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test getting extraction by ID."""
|
|
mock_extraction_info: ExtractionInfo = {
|
|
"id": 1,
|
|
"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
|
|
"title": "Never Gonna Give You Up",
|
|
"service": "youtube",
|
|
"service_id": "dQw4w9WgXcQ",
|
|
"status": "completed",
|
|
"error": None,
|
|
"sound_id": 42,
|
|
}
|
|
|
|
with patch(
|
|
"app.services.extraction.ExtractionService.get_extraction_by_id",
|
|
) as mock_get:
|
|
mock_get.return_value = mock_extraction_info
|
|
|
|
response = await authenticated_client.get("/api/v1/sounds/extract/1")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == 1
|
|
assert data["title"] == "Never Gonna Give You Up"
|
|
assert data["status"] == "completed"
|
|
assert data["sound_id"] == 42
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_extraction_by_id_not_found(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test getting non-existent extraction."""
|
|
with patch(
|
|
"app.services.extraction.ExtractionService.get_extraction_by_id",
|
|
) as mock_get:
|
|
mock_get.return_value = None
|
|
|
|
response = await authenticated_client.get("/api/v1/sounds/extract/999")
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "Extraction 999 not found" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_extractions_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test getting user extractions."""
|
|
mock_extractions: list[ExtractionInfo] = [
|
|
{
|
|
"id": 1,
|
|
"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
|
|
"title": "Never Gonna Give You Up",
|
|
"service": "youtube",
|
|
"service_id": "dQw4w9WgXcQ",
|
|
"status": "completed",
|
|
"error": None,
|
|
"sound_id": 42,
|
|
},
|
|
{
|
|
"id": 2,
|
|
"url": "https://soundcloud.com/example/track",
|
|
"title": "Example Track",
|
|
"service": "soundcloud",
|
|
"service_id": "example-track",
|
|
"status": "pending",
|
|
"error": None,
|
|
"sound_id": None,
|
|
},
|
|
]
|
|
|
|
with patch(
|
|
"app.services.extraction.ExtractionService.get_user_extractions",
|
|
) as mock_get:
|
|
mock_get.return_value = mock_extractions
|
|
|
|
response = await authenticated_client.get("/api/v1/sounds/extract")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["extractions"]) == 2
|
|
assert data["extractions"][0]["title"] == "Never Gonna Give You Up"
|
|
assert data["extractions"][1]["status"] == "pending"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_play_sound_with_vlc_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test playing sound with VLC successfully."""
|
|
# Mock the sound
|
|
mock_sound = type(
|
|
"Sound",
|
|
(),
|
|
{
|
|
"id": 1,
|
|
"name": "Test Sound",
|
|
"filename": "test.mp3",
|
|
"type": "SDB",
|
|
},
|
|
)()
|
|
|
|
with (
|
|
patch("app.repositories.sound.SoundRepository.get_by_id") as mock_get_sound,
|
|
patch(
|
|
"app.services.credit.CreditService.validate_and_reserve_credits",
|
|
) as mock_validate,
|
|
patch("app.services.vlc_player.VLCPlayerService.play_sound") as mock_play,
|
|
patch("app.services.credit.CreditService.deduct_credits") as mock_deduct,
|
|
):
|
|
mock_get_sound.return_value = mock_sound
|
|
mock_validate.return_value = None # No exception means validation passed
|
|
mock_play.return_value = True # Success
|
|
mock_deduct.return_value = None
|
|
|
|
response = await authenticated_client.post("/api/v1/sounds/play/1")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "Test Sound" in data["message"]
|
|
assert data["sound_id"] == 1
|
|
assert data["success"] is True
|
|
assert data["credits_deducted"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_play_sound_with_vlc_sound_not_found(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test playing non-existent sound with VLC."""
|
|
with patch(
|
|
"app.repositories.sound.SoundRepository.get_by_id",
|
|
) as mock_get_sound:
|
|
mock_get_sound.return_value = None
|
|
|
|
response = await authenticated_client.post("/api/v1/sounds/play/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_play_sound_with_vlc_insufficient_credits(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test playing sound with VLC when user has insufficient credits."""
|
|
from app.services.credit import InsufficientCreditsError
|
|
|
|
# Mock the sound
|
|
mock_sound = type(
|
|
"Sound",
|
|
(),
|
|
{
|
|
"id": 1,
|
|
"name": "Test Sound",
|
|
"filename": "test.mp3",
|
|
"type": "SDB",
|
|
},
|
|
)()
|
|
|
|
with (
|
|
patch("app.repositories.sound.SoundRepository.get_by_id") as mock_get_sound,
|
|
patch(
|
|
"app.services.credit.CreditService.validate_and_reserve_credits",
|
|
) as mock_validate,
|
|
):
|
|
mock_get_sound.return_value = mock_sound
|
|
mock_validate.side_effect = InsufficientCreditsError(
|
|
required=1,
|
|
available=0,
|
|
)
|
|
|
|
response = await authenticated_client.post("/api/v1/sounds/play/1")
|
|
|
|
assert response.status_code == 402
|
|
data = response.json()
|
|
assert "Insufficient credits" in data["detail"]
|
|
assert "1 required, 0 available" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_all_vlc_instances_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test stopping all VLC instances."""
|
|
mock_result = {
|
|
"message": "All VLC instances stopped",
|
|
"stopped_count": 3,
|
|
}
|
|
|
|
with patch(
|
|
"app.services.vlc_player.VLCPlayerService.stop_all_vlc_instances",
|
|
) as mock_stop:
|
|
mock_stop.return_value = mock_result
|
|
|
|
response = await authenticated_client.post("/api/v1/sounds/stop")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "All VLC instances stopped"
|
|
assert data["stopped_count"] == 3
|