Add tests for extraction API endpoints and enhance existing tests
Some checks failed
Backend CI / lint (push) Successful in 9m25s
Backend CI / test (push) Failing after 4m48s

- 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.
This commit is contained in:
JSC
2025-08-25 21:40:31 +02:00
parent d3ce17f10d
commit 7dee6e320e
15 changed files with 1560 additions and 721 deletions

View File

@@ -0,0 +1,154 @@
"""Tests for admin extraction API endpoints."""
import pytest
from httpx import AsyncClient
from app.models.extraction import Extraction
from app.models.user import User
class TestAdminExtractionEndpoints:
"""Test admin extraction endpoints."""
@pytest.mark.asyncio
async def test_get_extraction_processor_status(self, authenticated_admin_client):
"""Test getting extraction processor status."""
response = await authenticated_admin_client.get(
"/api/v1/admin/extractions/status",
)
assert response.status_code == 200
data = response.json()
# Check expected status fields (match actual processor status format)
assert "currently_processing" in data
assert "max_concurrent" in data
assert "available_slots" in data
assert "processing_ids" in data
assert isinstance(data["currently_processing"], int)
assert isinstance(data["max_concurrent"], int)
assert isinstance(data["available_slots"], int)
assert isinstance(data["processing_ids"], list)
@pytest.mark.asyncio
async def test_admin_delete_extraction_success(
self,
authenticated_admin_client,
test_session,
test_plan,
):
"""Test admin successfully deleting any extraction."""
# Create a test user
user = User(
name="Test User",
email="test@example.com",
is_active=True,
plan_id=test_plan.id,
)
test_session.add(user)
await test_session.commit()
await test_session.refresh(user)
# Create test extraction
extraction = Extraction(
url="https://example.com/video",
user_id=user.id,
status="completed",
)
test_session.add(extraction)
await test_session.commit()
await test_session.refresh(extraction)
# Admin delete the extraction
response = await authenticated_admin_client.delete(
f"/api/v1/admin/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_admin_delete_extraction_not_found(self, authenticated_admin_client):
"""Test admin deleting non-existent extraction."""
response = await authenticated_admin_client.delete(
"/api/v1/admin/extractions/999",
)
assert response.status_code == 404
data = response.json()
assert "not found" in data["detail"].lower()
@pytest.mark.asyncio
async def test_admin_delete_extraction_any_user(
self,
authenticated_admin_client,
test_session,
test_plan,
):
"""Test admin deleting extraction owned by any user."""
# Create another user and their extraction
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)
# Admin can delete any user's extraction
response = await authenticated_admin_client.delete(
f"/api/v1/admin/extractions/{extraction.id}",
)
assert response.status_code == 200
data = response.json()
assert data["message"] == f"Extraction {extraction.id} deleted successfully"
@pytest.mark.asyncio
async def test_delete_extraction_non_admin(self, authenticated_client, test_user, test_session):
"""Test non-admin user cannot access admin deletion endpoint."""
# 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)
# Non-admin user cannot access admin endpoint
response = await authenticated_client.delete(
f"/api/v1/admin/extractions/{extraction.id}",
)
assert response.status_code == 403
data = response.json()
assert "permissions" in data["detail"].lower()
@pytest.mark.asyncio
async def test_admin_endpoints_unauthenticated(self, client: AsyncClient):
"""Test admin endpoints require authentication."""
# Status endpoint
response = await client.get("/api/v1/admin/extractions/status")
assert response.status_code == 401
# Delete endpoint
response = await client.delete("/api/v1/admin/extractions/1")
assert response.status_code == 401

View File

@@ -31,6 +31,7 @@ class TestAdminSoundEndpoints:
"deleted": 1,
"skipped": 0,
"errors": 0,
"duplicates": 0,
"files": [
{
"filename": "test1.mp3",
@@ -176,6 +177,7 @@ class TestAdminSoundEndpoints:
"deleted": 0,
"skipped": 0,
"errors": 0,
"duplicates": 0,
"files": [
{
"filename": "custom1.wav",