92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
"""Tests for extraction API endpoints."""
|
|
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestExtractionEndpoints:
|
|
"""Test extraction API endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_extraction_success(
|
|
self, test_client: AsyncClient, auth_cookies: dict[str, str],
|
|
) -> None:
|
|
"""Test successful extraction creation."""
|
|
# Set cookies on client instance to avoid deprecation warning
|
|
test_client.cookies.update(auth_cookies)
|
|
|
|
response = await test_client.post(
|
|
"/api/v1/sounds/extract",
|
|
params={"url": "https://www.youtube.com/watch?v=test"},
|
|
)
|
|
|
|
# This will fail because we don't have actual extraction service mocked
|
|
# But at least we'll get past authentication
|
|
assert response.status_code in [200, 400, 500] # Allow any non-auth error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_extraction_unauthenticated(self, test_client: AsyncClient) -> None:
|
|
"""Test extraction creation without authentication."""
|
|
response = await test_client.post(
|
|
"/api/v1/sounds/extract",
|
|
params={"url": "https://www.youtube.com/watch?v=test"},
|
|
)
|
|
|
|
# Should return 401 for missing authentication
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_extraction_unauthenticated(self, test_client: AsyncClient) -> None:
|
|
"""Test extraction retrieval without authentication."""
|
|
response = await test_client.get("/api/v1/sounds/extract/1")
|
|
|
|
# Should return 401 for missing authentication
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_processor_status_admin(
|
|
self, test_client: AsyncClient, admin_cookies: dict[str, str],
|
|
) -> None:
|
|
"""Test getting processor status as admin."""
|
|
# Set cookies on client instance to avoid deprecation warning
|
|
test_client.cookies.update(admin_cookies)
|
|
|
|
response = await test_client.get("/api/v1/sounds/extract/status")
|
|
|
|
# Should succeed for admin users
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
assert "max_concurrent" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_processor_status_non_admin(
|
|
self, test_client: AsyncClient, auth_cookies: dict[str, str],
|
|
) -> None:
|
|
"""Test getting processor status as non-admin user."""
|
|
# Set cookies on client instance to avoid deprecation warning
|
|
test_client.cookies.update(auth_cookies)
|
|
|
|
response = await test_client.get("/api/v1/sounds/extract/status")
|
|
|
|
# Should return 403 for non-admin users
|
|
assert response.status_code == 403
|
|
assert "Only administrators" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_extractions(
|
|
self, test_client: AsyncClient, auth_cookies: dict[str, str],
|
|
) -> None:
|
|
"""Test getting user extractions."""
|
|
# Set cookies on client instance to avoid deprecation warning
|
|
test_client.cookies.update(auth_cookies)
|
|
|
|
response = await test_client.get("/api/v1/sounds/extract")
|
|
|
|
# Should succeed and return empty list (no extractions in test DB)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "extractions" in data
|
|
assert isinstance(data["extractions"], list)
|