b4f0f54516
- Updated ExtractionInfo to include user_id, created_at, and updated_at fields. - Modified ExtractionService to return user and timestamp information in extraction responses. - Enhanced sound serialization in PlayerState to include extraction URL if available. - Adjusted PlaylistRepository to load sound extractions when retrieving playlist sounds. - Added tests for new fields in extraction and sound endpoints, ensuring proper response structure. - Created new test file endpoints for sound downloads and thumbnail retrievals, including success and error cases. - Refactored various test cases for consistency and clarity, ensuring proper mocking and assertions.
187 lines
6.3 KiB
Python
187 lines
6.3 KiB
Python
"""Tests for extraction API endpoints."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from app.models.extraction import Extraction
|
|
from app.models.user import User
|
|
|
|
|
|
class TestExtractionEndpoints:
|
|
"""Test extraction API endpoints."""
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_extraction(
|
|
self,
|
|
test_session: AsyncSession,
|
|
authenticated_user: User,
|
|
) -> Extraction:
|
|
"""Create a test extraction."""
|
|
extraction = Extraction(
|
|
url="https://www.youtube.com/watch?v=test",
|
|
user_id=authenticated_user.id,
|
|
service="youtube",
|
|
service_id="test",
|
|
title="Test Video",
|
|
status="completed",
|
|
)
|
|
test_session.add(extraction)
|
|
await test_session.commit()
|
|
await test_session.refresh(extraction)
|
|
return extraction
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_extraction_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test successful extraction creation with proper response format."""
|
|
# Store user ID to avoid session issues
|
|
user_id = authenticated_user.id
|
|
|
|
with patch(
|
|
"app.services.extraction_processor.extraction_processor.queue_extraction",
|
|
):
|
|
response = await authenticated_client.post(
|
|
"/api/v1/extractions/",
|
|
params={"url": "https://www.youtube.com/watch?v=test"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Verify response structure
|
|
assert "message" in data
|
|
assert "extraction" in data
|
|
|
|
extraction_data = data["extraction"]
|
|
# Verify all required fields including timestamps
|
|
assert "id" in extraction_data
|
|
assert "url" in extraction_data
|
|
assert "user_id" in extraction_data
|
|
assert "status" in extraction_data
|
|
assert "created_at" in extraction_data
|
|
assert "updated_at" in extraction_data
|
|
assert extraction_data["url"] == "https://www.youtube.com/watch?v=test"
|
|
assert extraction_data["user_id"] == user_id
|
|
assert extraction_data["status"] == "pending"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_extraction_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
test_extraction: Extraction,
|
|
) -> None:
|
|
"""Test successful extraction retrieval with timestamp fields."""
|
|
response = await authenticated_client.get(
|
|
f"/api/v1/extractions/{test_extraction.id}",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Verify all fields including timestamps
|
|
assert data["id"] == test_extraction.id
|
|
assert data["url"] == test_extraction.url
|
|
assert data["user_id"] == test_extraction.user_id
|
|
assert data["service"] == test_extraction.service
|
|
assert data["service_id"] == test_extraction.service_id
|
|
assert data["title"] == test_extraction.title
|
|
assert data["status"] == test_extraction.status
|
|
assert "created_at" in data
|
|
assert "updated_at" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_extractions_success(
|
|
self,
|
|
authenticated_client: AsyncClient,
|
|
authenticated_user: User,
|
|
test_extraction: Extraction,
|
|
) -> None:
|
|
"""Test successful user extractions retrieval with timestamp fields."""
|
|
response = await authenticated_client.get(
|
|
"/api/v1/extractions/",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Verify response structure
|
|
assert "extractions" in data
|
|
assert len(data["extractions"]) >= 1
|
|
|
|
extraction_data = data["extractions"][0]
|
|
# Verify all fields including timestamps
|
|
assert "id" in extraction_data
|
|
assert "url" in extraction_data
|
|
assert "user_id" in extraction_data
|
|
assert "status" in extraction_data
|
|
assert "created_at" in extraction_data
|
|
assert "updated_at" in extraction_data
|
|
|
|
@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/extractions/",
|
|
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/extractions/1")
|
|
|
|
# Should return 401 for missing authentication
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_processor_status_moved_to_admin(
|
|
self,
|
|
test_client: AsyncClient,
|
|
admin_cookies: dict[str, str],
|
|
) -> None:
|
|
"""Test that processor status endpoint was moved to admin."""
|
|
# Set cookies on client instance to avoid deprecation warning
|
|
test_client.cookies.update(admin_cookies)
|
|
|
|
# The new admin endpoint should work
|
|
response = await test_client.get("/api/v1/admin/extractions/status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data or "is_running" in data
|
|
assert "max_concurrent" in data
|
|
|
|
@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/extractions/")
|
|
|
|
# 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)
|