491 lines
15 KiB
Python
491 lines
15 KiB
Python
"""Tests for NFO API endpoints.
|
|
|
|
This module tests all NFO management REST API endpoints.
|
|
"""
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from src.server.fastapi_app import app
|
|
from src.server.models.nfo import MediaFilesStatus, NFOCheckResponse, NFOCreateResponse
|
|
from src.server.services.auth_service import auth_service
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_auth():
|
|
"""Reset authentication state before each test."""
|
|
original_hash = auth_service._hash
|
|
auth_service._hash = None
|
|
auth_service._failed.clear()
|
|
yield
|
|
auth_service._hash = original_hash
|
|
auth_service._failed.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
"""Create an async test client."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
async def authenticated_client(client):
|
|
"""Create an authenticated test client with token."""
|
|
# Setup master password
|
|
await client.post(
|
|
"/api/auth/setup",
|
|
json={"master_password": "TestPassword123!"}
|
|
)
|
|
|
|
# Login to get token
|
|
response = await client.post(
|
|
"/api/auth/login",
|
|
json={"password": "TestPassword123!"}
|
|
)
|
|
token = response.json()["access_token"]
|
|
|
|
# Add token to default headers
|
|
client.headers.update({"Authorization": f"Bearer {token}"})
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_series_app():
|
|
"""Create mock series app."""
|
|
app_mock = Mock()
|
|
serie = Mock()
|
|
serie.key = "test-anime"
|
|
serie.folder = "Test Anime (2024)"
|
|
serie.name = "Test Anime"
|
|
serie.ensure_folder_with_year = Mock(return_value="Test Anime (2024)")
|
|
|
|
# Mock the list manager
|
|
list_manager = Mock()
|
|
list_manager.GetList = Mock(return_value=[serie])
|
|
app_mock.list = list_manager
|
|
|
|
return app_mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_nfo_service():
|
|
"""Create mock NFO service."""
|
|
service = Mock()
|
|
service.check_nfo_exists = AsyncMock(return_value=False)
|
|
service.create_tvshow_nfo = AsyncMock(return_value="/path/to/tvshow.nfo")
|
|
service.update_tvshow_nfo = AsyncMock(return_value="/path/to/tvshow.nfo")
|
|
return service
|
|
|
|
|
|
@pytest.fixture
|
|
def override_nfo_service_for_auth_tests():
|
|
"""Placeholder fixture for auth tests.
|
|
|
|
Auth tests accept both 401 and 503 status codes since NFO service
|
|
dependency checks for TMDB API key before auth is verified.
|
|
"""
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def override_dependencies(mock_series_app, mock_nfo_service):
|
|
"""Override dependencies for authenticated NFO tests."""
|
|
from src.server.api.nfo import get_nfo_service
|
|
from src.server.utils.dependencies import get_series_app
|
|
|
|
app.dependency_overrides[get_series_app] = lambda: mock_series_app
|
|
app.dependency_overrides[get_nfo_service] = lambda: mock_nfo_service
|
|
|
|
yield
|
|
|
|
# Clean up only our overrides
|
|
if get_series_app in app.dependency_overrides:
|
|
del app.dependency_overrides[get_series_app]
|
|
if get_nfo_service in app.dependency_overrides:
|
|
del app.dependency_overrides[get_nfo_service]
|
|
|
|
|
|
class TestNFOCheckEndpoint:
|
|
"""Tests for GET /api/nfo/{serie_id}/check endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_nfo_requires_auth(
|
|
self,
|
|
override_nfo_service_for_auth_tests,
|
|
client
|
|
):
|
|
"""Test that check endpoint requires authentication.
|
|
|
|
Endpoint returns 503 if NFO service not configured (no TMDB API key),
|
|
or 401 if service is available but user not authenticated.
|
|
Both indicate endpoint is protected.
|
|
"""
|
|
response = await client.get("/api/nfo/test-anime/check")
|
|
assert response.status_code in (401, 503)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_nfo_series_not_found(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
override_dependencies
|
|
):
|
|
"""Test check endpoint with non-existent series."""
|
|
mock_series_app.list.GetList = Mock(return_value=[])
|
|
|
|
response = await authenticated_client.get(
|
|
"/api/nfo/nonexistent/check"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_nfo_success(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test successful NFO check."""
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.get(
|
|
"/api/nfo/test-anime/check"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["serie_id"] == "test-anime"
|
|
assert data["serie_folder"] == "Test Anime (2024)"
|
|
assert data["has_nfo"] is False
|
|
|
|
|
|
class TestNFOCreateEndpoint:
|
|
"""Tests for POST /api/nfo/{serie_id}/create endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_nfo_requires_auth(
|
|
self,
|
|
client,
|
|
override_nfo_service_for_auth_tests
|
|
):
|
|
"""Test that create endpoint requires authentication."""
|
|
response = await client.post(
|
|
"/api/nfo/test-anime/create",
|
|
json={}
|
|
)
|
|
assert response.status_code in (401, 503)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_nfo_success(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test successful NFO creation."""
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/nfo/test-anime/create",
|
|
json={
|
|
"download_poster": True,
|
|
"download_logo": True,
|
|
"download_fanart": True
|
|
}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["serie_id"] == "test-anime"
|
|
assert "NFO and media files created" in data["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_nfo_already_exists(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test NFO creation when NFO already exists."""
|
|
mock_nfo_service.check_nfo_exists = AsyncMock(return_value=True)
|
|
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/nfo/test-anime/create",
|
|
json={"overwrite_existing": False}
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_nfo_with_year(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test NFO creation with year parameter."""
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/nfo/test-anime/create",
|
|
json={
|
|
"year": 2024,
|
|
"download_poster": True
|
|
}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Verify year was passed to service
|
|
mock_nfo_service.create_tvshow_nfo.assert_called_once()
|
|
call_kwargs = mock_nfo_service.create_tvshow_nfo.call_args[1]
|
|
assert call_kwargs["year"] == 2024
|
|
|
|
|
|
class TestNFOUpdateEndpoint:
|
|
"""Tests for PUT /api/nfo/{serie_id}/update endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nfo_requires_auth(
|
|
self,
|
|
client,
|
|
override_nfo_service_for_auth_tests
|
|
):
|
|
"""Test that update endpoint requires authentication."""
|
|
response = await client.put("/api/nfo/test-anime/update")
|
|
assert response.status_code in (401, 503)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nfo_not_found(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test update when NFO doesn't exist."""
|
|
mock_nfo_service.check_nfo_exists = AsyncMock(return_value=False)
|
|
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.put(
|
|
"/api/nfo/test-anime/update"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nfo_success(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test successful NFO update."""
|
|
mock_nfo_service.check_nfo_exists = AsyncMock(return_value=True)
|
|
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.put(
|
|
"/api/nfo/test-anime/update?download_media=true"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "NFO updated successfully" in data["message"]
|
|
|
|
|
|
class TestNFOContentEndpoint:
|
|
"""Tests for GET /api/nfo/{serie_id}/content endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_content_requires_auth(
|
|
self,
|
|
client,
|
|
override_nfo_service_for_auth_tests
|
|
):
|
|
"""Test that content endpoint requires authentication."""
|
|
response = await client.get("/api/nfo/test-anime/content")
|
|
assert response.status_code in (401, 503)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_content_nfo_not_found(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test get content when NFO doesn't exist."""
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.get(
|
|
"/api/nfo/test-anime/content"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_content_success(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test successful content retrieval."""
|
|
# Create NFO file
|
|
anime_dir = tmp_path / "Test Anime (2024)"
|
|
anime_dir.mkdir()
|
|
nfo_file = anime_dir / "tvshow.nfo"
|
|
nfo_file.write_text("<tvshow><title>Test</title></tvshow>")
|
|
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.get(
|
|
"/api/nfo/test-anime/content"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "<tvshow>" in data["content"]
|
|
assert data["file_size"] > 0
|
|
|
|
|
|
class TestNFOMissingEndpoint:
|
|
"""Tests for GET /api/nfo/missing endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_missing_requires_auth(
|
|
self,
|
|
client,
|
|
override_nfo_service_for_auth_tests
|
|
):
|
|
"""Test that missing endpoint requires authentication."""
|
|
response = await client.get("/api/nfo/missing")
|
|
assert response.status_code in (401, 503)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_missing_success(
|
|
self,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path,
|
|
override_dependencies
|
|
):
|
|
"""Test getting list of series without NFO."""
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.get("/api/nfo/missing")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_series" in data
|
|
assert "missing_nfo_count" in data
|
|
assert "series" in data
|
|
|
|
|
|
class TestNFOBatchCreateEndpoint:
|
|
"""Tests for POST /api/nfo/batch/create endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_requires_auth(
|
|
self,
|
|
client,
|
|
override_nfo_service_for_auth_tests
|
|
):
|
|
"""Test that batch create endpoint requires authentication."""
|
|
response = await client.post(
|
|
"/api/nfo/batch/create",
|
|
json={"serie_ids": ["test1", "test2"]}
|
|
)
|
|
assert response.status_code in (401, 503)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_success(
|
|
self,
|
|
override_dependencies,
|
|
authenticated_client,
|
|
mock_series_app,
|
|
mock_nfo_service,
|
|
tmp_path
|
|
):
|
|
"""Test successful batch NFO creation."""
|
|
with patch('src.server.api.nfo.settings') as mock_settings:
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/nfo/batch/create",
|
|
json={
|
|
"serie_ids": ["test-anime"],
|
|
"download_media": True,
|
|
"skip_existing": False,
|
|
"max_concurrent": 3
|
|
}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 1
|
|
assert "successful" in data
|
|
assert "results" in data
|
|
|
|
|
|
class TestNFOServiceDependency:
|
|
"""Tests for NFO service dependency."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_service_unavailable_without_api_key(
|
|
self,
|
|
authenticated_client
|
|
):
|
|
"""Test NFO endpoints fail gracefully without TMDB API key.
|
|
|
|
This test verifies that when the NFO service dependency raises an
|
|
HTTPException 503 due to missing TMDB API key, the endpoint returns 503.
|
|
"""
|
|
from fastapi import HTTPException, status
|
|
|
|
from src.server.api.nfo import get_nfo_service
|
|
|
|
# Create a dependency that raises HTTPException 503 (simulating missing API key)
|
|
async def fail_nfo_service():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="NFO service not configured: TMDB API key not available"
|
|
)
|
|
|
|
# Override NFO service to simulate missing API key
|
|
app.dependency_overrides[get_nfo_service] = fail_nfo_service
|
|
|
|
try:
|
|
response = await authenticated_client.get(
|
|
"/api/nfo/test-anime/check"
|
|
)
|
|
assert response.status_code == 503
|
|
data = response.json()
|
|
assert "not configured" in data["detail"]
|
|
finally:
|
|
# Clean up override
|
|
if get_nfo_service in app.dependency_overrides:
|
|
del app.dependency_overrides[get_nfo_service]
|