7dee6e320e
- Implement tests for admin extraction API endpoints including status retrieval, deletion of extractions, and permission checks. - Add tests for user extraction deletion, ensuring proper handling of permissions and non-existent extractions. - Enhance sound endpoint tests to include duplicate handling in responses. - Refactor favorite service tests to utilize mock dependencies for better maintainability and clarity. - Update sound scanner tests to improve file handling and ensure proper deletion of associated files.
302 lines
10 KiB
Python
302 lines
10 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)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_processing_extractions(
|
|
self,
|
|
test_client: AsyncClient,
|
|
auth_cookies: dict[str, str],
|
|
test_session: AsyncSession,
|
|
authenticated_user: User,
|
|
) -> None:
|
|
"""Test getting currently processing extractions."""
|
|
# Create a processing extraction
|
|
processing_extraction = Extraction(
|
|
url="https://www.youtube.com/watch?v=processing",
|
|
user_id=authenticated_user.id,
|
|
service="youtube",
|
|
service_id="processing123",
|
|
title="Processing Video",
|
|
status="processing",
|
|
)
|
|
test_session.add(processing_extraction)
|
|
await test_session.commit()
|
|
await test_session.refresh(processing_extraction)
|
|
|
|
# Set cookies on client instance
|
|
test_client.cookies.update(auth_cookies)
|
|
|
|
response = await test_client.get("/api/v1/extractions/processing/current")
|
|
|
|
# Should succeed and return the processing extraction
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 1
|
|
|
|
# Find our processing extraction in the results
|
|
processing_found = False
|
|
for extraction in data:
|
|
if extraction["id"] == processing_extraction.id:
|
|
processing_found = True
|
|
assert extraction["status"] == "processing"
|
|
assert extraction["title"] == "Processing Video"
|
|
assert extraction["url"] == "https://www.youtube.com/watch?v=processing"
|
|
break
|
|
|
|
assert processing_found, "Processing extraction not found in results"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_extraction_success(self, authenticated_client, test_user, test_session):
|
|
"""Test successful deletion of user's own extraction."""
|
|
# Create test extraction
|
|
extraction = Extraction(
|
|
url="https://example.com/video",
|
|
user_id=test_user.id,
|
|
status="completed",
|
|
)
|
|
test_session.add(extraction)
|
|
await test_session.commit()
|
|
await test_session.refresh(extraction)
|
|
|
|
# Delete the extraction
|
|
response = await authenticated_client.delete(f"/api/v1/extractions/{extraction.id}")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == f"Extraction {extraction.id} deleted successfully"
|
|
|
|
# Verify extraction was deleted from database
|
|
deleted_extraction = await test_session.get(Extraction, extraction.id)
|
|
assert deleted_extraction is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_extraction_not_found(self, authenticated_client):
|
|
"""Test deleting non-existent extraction."""
|
|
response = await authenticated_client.delete("/api/v1/extractions/999")
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "not found" in data["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_extraction_permission_denied(self, authenticated_client, test_session, test_plan):
|
|
"""Test deleting another user's extraction."""
|
|
# Create extraction owned by different user
|
|
other_user = User(
|
|
name="Other User",
|
|
email="other@example.com",
|
|
is_active=True,
|
|
plan_id=test_plan.id,
|
|
)
|
|
test_session.add(other_user)
|
|
await test_session.commit()
|
|
await test_session.refresh(other_user)
|
|
|
|
extraction = Extraction(
|
|
url="https://example.com/video",
|
|
user_id=other_user.id,
|
|
status="completed",
|
|
)
|
|
test_session.add(extraction)
|
|
await test_session.commit()
|
|
await test_session.refresh(extraction)
|
|
|
|
# Try to delete other user's extraction
|
|
response = await authenticated_client.delete(f"/api/v1/extractions/{extraction.id}")
|
|
|
|
assert response.status_code == 403
|
|
data = response.json()
|
|
assert "permission" in data["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_extraction_unauthenticated(self, client):
|
|
"""Test deleting extraction without authentication."""
|
|
response = await client.delete("/api/v1/extractions/1")
|
|
|
|
assert response.status_code == 401
|