Complete Task 8: Database Support for NFO Status
- Added 5 NFO tracking fields to AnimeSeries model - Fields: has_nfo, nfo_created_at, nfo_updated_at, tmdb_id, tvdb_id - Added 3 service methods to AnimeService for NFO operations - Methods: update_nfo_status, get_series_without_nfo, get_nfo_statistics - SQLAlchemy auto-migration (no manual migration needed) - Backward compatible with existing data - 15 new tests added (19/19 passing) - Tests: database models, service methods, integration queries
This commit is contained in:
224
tests/integration/test_nfo_database.py
Normal file
224
tests/integration/test_nfo_database.py
Normal file
@@ -0,0 +1,224 @@
|
||||
"""Integration tests for NFO database operations.
|
||||
|
||||
Tests NFO status tracking across database models and services.
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from src.server.database.base import Base
|
||||
from src.server.database.models import AnimeSeries
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_engine():
|
||||
"""Create in-memory SQLite database for testing."""
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(engine)
|
||||
return engine
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session(db_engine):
|
||||
"""Create database session for testing."""
|
||||
SessionLocal = sessionmaker(bind=db_engine)
|
||||
session = SessionLocal()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
class TestNFODatabaseIntegration:
|
||||
"""Integration tests for NFO database operations."""
|
||||
|
||||
def test_create_series_with_nfo_tracking(self, db_session):
|
||||
"""Test creating series with NFO tracking fields."""
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
series = AnimeSeries(
|
||||
key="test-series-nfo",
|
||||
name="Test Series",
|
||||
site="https://example.com",
|
||||
folder="/anime/test",
|
||||
has_nfo=True,
|
||||
nfo_created_at=now,
|
||||
tmdb_id=12345,
|
||||
tvdb_id=67890
|
||||
)
|
||||
|
||||
db_session.add(series)
|
||||
db_session.commit()
|
||||
|
||||
# Retrieve and verify
|
||||
retrieved = db_session.query(AnimeSeries).filter_by(
|
||||
key="test-series-nfo"
|
||||
).first()
|
||||
|
||||
assert retrieved is not None
|
||||
assert retrieved.has_nfo is True
|
||||
assert retrieved.tmdb_id == 12345
|
||||
assert retrieved.tvdb_id == 67890
|
||||
assert retrieved.nfo_created_at is not None
|
||||
|
||||
def test_query_series_without_nfo(self, db_session):
|
||||
"""Test querying series without NFO files."""
|
||||
# Create series with and without NFO
|
||||
series_with = AnimeSeries(
|
||||
key="with-nfo",
|
||||
name="Series With NFO",
|
||||
site="https://example.com",
|
||||
folder="/anime/with",
|
||||
has_nfo=True,
|
||||
)
|
||||
series_without_1 = AnimeSeries(
|
||||
key="without-nfo-1",
|
||||
name="Series Without NFO 1",
|
||||
site="https://example.com",
|
||||
folder="/anime/without1",
|
||||
has_nfo=False,
|
||||
)
|
||||
series_without_2 = AnimeSeries(
|
||||
key="without-nfo-2",
|
||||
name="Series Without NFO 2",
|
||||
site="https://example.com",
|
||||
folder="/anime/without2",
|
||||
has_nfo=False,
|
||||
)
|
||||
|
||||
db_session.add_all([series_with, series_without_1, series_without_2])
|
||||
db_session.commit()
|
||||
|
||||
# Query series without NFO
|
||||
without_nfo = db_session.query(AnimeSeries).filter(
|
||||
AnimeSeries.has_nfo == False # noqa: E712
|
||||
).all()
|
||||
|
||||
assert len(without_nfo) == 2
|
||||
assert all(s.has_nfo is False for s in without_nfo)
|
||||
keys = [s.key for s in without_nfo]
|
||||
assert "without-nfo-1" in keys
|
||||
assert "without-nfo-2" in keys
|
||||
|
||||
def test_query_series_by_tmdb_id(self, db_session):
|
||||
"""Test querying series by TMDB ID."""
|
||||
series1 = AnimeSeries(
|
||||
key="series-1",
|
||||
name="Series 1",
|
||||
site="https://example.com",
|
||||
folder="/anime/1",
|
||||
tmdb_id=111,
|
||||
)
|
||||
series2 = AnimeSeries(
|
||||
key="series-2",
|
||||
name="Series 2",
|
||||
site="https://example.com",
|
||||
folder="/anime/2",
|
||||
tmdb_id=222,
|
||||
)
|
||||
series3 = AnimeSeries(
|
||||
key="series-3",
|
||||
name="Series 3",
|
||||
site="https://example.com",
|
||||
folder="/anime/3",
|
||||
)
|
||||
|
||||
db_session.add_all([series1, series2, series3])
|
||||
db_session.commit()
|
||||
|
||||
# Query specific TMDB ID
|
||||
result = db_session.query(AnimeSeries).filter_by(
|
||||
tmdb_id=111
|
||||
).first()
|
||||
assert result.key == "series-1"
|
||||
|
||||
# Query series with any TMDB ID
|
||||
with_tmdb = db_session.query(AnimeSeries).filter(
|
||||
AnimeSeries.tmdb_id.isnot(None)
|
||||
).all()
|
||||
assert len(with_tmdb) == 2
|
||||
|
||||
def test_update_nfo_status(self, db_session):
|
||||
"""Test updating NFO status after creation."""
|
||||
series = AnimeSeries(
|
||||
key="update-test",
|
||||
name="Update Test",
|
||||
site="https://example.com",
|
||||
folder="/anime/update",
|
||||
has_nfo=False,
|
||||
)
|
||||
db_session.add(series)
|
||||
db_session.commit()
|
||||
|
||||
# Update NFO status
|
||||
now = datetime.now(timezone.utc)
|
||||
series.has_nfo = True
|
||||
series.nfo_created_at = now
|
||||
series.nfo_updated_at = now
|
||||
series.tmdb_id = 99999
|
||||
db_session.commit()
|
||||
|
||||
# Refresh and verify
|
||||
db_session.refresh(series)
|
||||
assert series.has_nfo is True
|
||||
assert series.nfo_created_at is not None
|
||||
assert series.nfo_updated_at is not None
|
||||
assert series.tmdb_id == 99999
|
||||
|
||||
def test_backward_compatibility(self, db_session):
|
||||
"""Test backward compatibility with existing series."""
|
||||
# Create series without NFO fields (like existing data)
|
||||
series = AnimeSeries(
|
||||
key="old-series",
|
||||
name="Old Series",
|
||||
site="https://example.com",
|
||||
folder="/anime/old",
|
||||
)
|
||||
db_session.add(series)
|
||||
db_session.commit()
|
||||
|
||||
# Verify default values
|
||||
db_session.refresh(series)
|
||||
assert series.has_nfo is False
|
||||
assert series.nfo_created_at is None
|
||||
assert series.nfo_updated_at is None
|
||||
assert series.tmdb_id is None
|
||||
assert series.tvdb_id is None
|
||||
|
||||
def test_nfo_statistics_queries(self, db_session):
|
||||
"""Test queries for NFO statistics."""
|
||||
# Create mixed data
|
||||
for i in range(10):
|
||||
has_nfo = i < 7 # 7 with NFO, 3 without
|
||||
tmdb = i * 100 if i < 8 else None # 8 with TMDB
|
||||
tvdb = i * 1000 if i < 5 else None # 5 with TVDB
|
||||
|
||||
series = AnimeSeries(
|
||||
key=f"series-{i}",
|
||||
name=f"Series {i}",
|
||||
site="https://example.com",
|
||||
folder=f"/anime/{i}",
|
||||
has_nfo=has_nfo,
|
||||
tmdb_id=tmdb,
|
||||
tvdb_id=tvdb,
|
||||
)
|
||||
db_session.add(series)
|
||||
|
||||
db_session.commit()
|
||||
|
||||
# Count statistics
|
||||
total = db_session.query(AnimeSeries).count()
|
||||
with_nfo = db_session.query(AnimeSeries).filter(
|
||||
AnimeSeries.has_nfo == True # noqa: E712
|
||||
).count()
|
||||
with_tmdb = db_session.query(AnimeSeries).filter(
|
||||
AnimeSeries.tmdb_id.isnot(None)
|
||||
).count()
|
||||
with_tvdb = db_session.query(AnimeSeries).filter(
|
||||
AnimeSeries.tvdb_id.isnot(None)
|
||||
).count()
|
||||
|
||||
assert total == 10
|
||||
assert with_nfo == 7
|
||||
assert with_tmdb == 8
|
||||
assert with_tvdb == 5
|
||||
@@ -341,6 +341,139 @@ class TestConcurrency:
|
||||
assert all(len(r) == 1 for r in results)
|
||||
|
||||
|
||||
class TestNFOTracking:
|
||||
"""Test NFO status tracking methods."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_nfo_status_success(self, anime_service):
|
||||
"""Test successful NFO status update."""
|
||||
mock_series = MagicMock()
|
||||
mock_series.key = "test-series"
|
||||
mock_series.has_nfo = False
|
||||
mock_series.nfo_created_at = None
|
||||
mock_series.nfo_updated_at = None
|
||||
mock_series.tmdb_id = None
|
||||
|
||||
mock_query = MagicMock()
|
||||
mock_query.filter.return_value.first.return_value = mock_series
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value = mock_query
|
||||
|
||||
# Update NFO status
|
||||
await anime_service.update_nfo_status(
|
||||
key="test-series",
|
||||
has_nfo=True,
|
||||
tmdb_id=12345,
|
||||
db=mock_db
|
||||
)
|
||||
|
||||
# Verify series was updated
|
||||
assert mock_series.has_nfo is True
|
||||
assert mock_series.tmdb_id == 12345
|
||||
assert mock_series.nfo_created_at is not None
|
||||
assert mock_series.nfo_updated_at is not None
|
||||
mock_db.commit.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_nfo_status_not_found(self, anime_service):
|
||||
"""Test NFO status update when series not found."""
|
||||
mock_query = MagicMock()
|
||||
mock_query.filter.return_value.first.return_value = None
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value = mock_query
|
||||
|
||||
# Should not raise, just log warning
|
||||
await anime_service.update_nfo_status(
|
||||
key="nonexistent",
|
||||
has_nfo=True,
|
||||
db=mock_db
|
||||
)
|
||||
|
||||
# Should not commit if series not found
|
||||
mock_db.commit.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_series_without_nfo(self, anime_service):
|
||||
"""Test getting series without NFO files."""
|
||||
mock_series1 = MagicMock()
|
||||
mock_series1.key = "series-1"
|
||||
mock_series1.name = "Series 1"
|
||||
mock_series1.folder = "Series 1 (2020)"
|
||||
mock_series1.tmdb_id = 123
|
||||
mock_series1.tvdb_id = None
|
||||
|
||||
mock_series2 = MagicMock()
|
||||
mock_series2.key = "series-2"
|
||||
mock_series2.name = "Series 2"
|
||||
mock_series2.folder = "Series 2 (2021)"
|
||||
mock_series2.tmdb_id = None
|
||||
mock_series2.tvdb_id = 456
|
||||
|
||||
mock_query = MagicMock()
|
||||
mock_query.filter.return_value.all.return_value = [
|
||||
mock_series1,
|
||||
mock_series2
|
||||
]
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value = mock_query
|
||||
|
||||
result = await anime_service.get_series_without_nfo(db=mock_db)
|
||||
|
||||
assert len(result) == 2
|
||||
assert result[0]["key"] == "series-1"
|
||||
assert result[0]["has_nfo"] is False
|
||||
assert result[0]["tmdb_id"] == 123
|
||||
assert result[1]["key"] == "series-2"
|
||||
assert result[1]["tvdb_id"] == 456
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_nfo_statistics(self, anime_service):
|
||||
"""Test getting NFO statistics."""
|
||||
mock_db = MagicMock()
|
||||
|
||||
# Mock total count
|
||||
mock_total_query = MagicMock()
|
||||
mock_total_query.count.return_value = 100
|
||||
|
||||
# Mock with_nfo count
|
||||
mock_with_nfo_query = MagicMock()
|
||||
mock_with_nfo_filter = MagicMock()
|
||||
mock_with_nfo_filter.count.return_value = 75
|
||||
mock_with_nfo_query.filter.return_value = mock_with_nfo_filter
|
||||
|
||||
# Mock with_tmdb count
|
||||
mock_with_tmdb_query = MagicMock()
|
||||
mock_with_tmdb_filter = MagicMock()
|
||||
mock_with_tmdb_filter.count.return_value = 80
|
||||
mock_with_tmdb_query.filter.return_value = mock_with_tmdb_filter
|
||||
|
||||
# Mock with_tvdb count
|
||||
mock_with_tvdb_query = MagicMock()
|
||||
mock_with_tvdb_filter = MagicMock()
|
||||
mock_with_tvdb_filter.count.return_value = 60
|
||||
mock_with_tvdb_query.filter.return_value = mock_with_tvdb_filter
|
||||
|
||||
# Configure mock to return different queries for each call
|
||||
query_returns = [
|
||||
mock_total_query,
|
||||
mock_with_nfo_query,
|
||||
mock_with_tmdb_query,
|
||||
mock_with_tvdb_query
|
||||
]
|
||||
mock_db.query.side_effect = query_returns
|
||||
|
||||
result = await anime_service.get_nfo_statistics(db=mock_db)
|
||||
|
||||
assert result["total"] == 100
|
||||
assert result["with_nfo"] == 75
|
||||
assert result["without_nfo"] == 25
|
||||
assert result["with_tmdb_id"] == 80
|
||||
assert result["with_tvdb_id"] == 60
|
||||
|
||||
|
||||
class TestFactoryFunction:
|
||||
"""Test factory function."""
|
||||
|
||||
|
||||
@@ -144,6 +144,169 @@ class TestAnimeSeries:
|
||||
)
|
||||
assert result.scalar_one_or_none() is None
|
||||
|
||||
def test_anime_series_nfo_fields_default_values(self, db_session: Session):
|
||||
"""Test NFO fields have correct default values."""
|
||||
series = AnimeSeries(
|
||||
key="nfo-test",
|
||||
name="NFO Test Series",
|
||||
site="https://example.com",
|
||||
folder="/anime/nfo-test",
|
||||
)
|
||||
db_session.add(series)
|
||||
db_session.commit()
|
||||
|
||||
# Verify NFO fields default values
|
||||
assert series.has_nfo is False
|
||||
assert series.nfo_created_at is None
|
||||
assert series.nfo_updated_at is None
|
||||
assert series.tmdb_id is None
|
||||
assert series.tvdb_id is None
|
||||
|
||||
def test_anime_series_nfo_fields_set_values(self, db_session: Session):
|
||||
"""Test setting NFO field values."""
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
series = AnimeSeries(
|
||||
key="nfo-values-test",
|
||||
name="NFO Values Test",
|
||||
site="https://example.com",
|
||||
folder="/anime/nfo-values",
|
||||
has_nfo=True,
|
||||
nfo_created_at=now,
|
||||
nfo_updated_at=now,
|
||||
tmdb_id=12345,
|
||||
tvdb_id=67890,
|
||||
)
|
||||
db_session.add(series)
|
||||
db_session.commit()
|
||||
|
||||
# Verify NFO fields are saved
|
||||
assert series.has_nfo is True
|
||||
assert series.nfo_created_at is not None
|
||||
assert series.nfo_updated_at is not None
|
||||
# Check time is close (within 1 second)
|
||||
# Timezone may be lost in SQLite
|
||||
created_delta = abs(
|
||||
(series.nfo_created_at.replace(tzinfo=timezone.utc) - now)
|
||||
.total_seconds()
|
||||
)
|
||||
updated_delta = abs(
|
||||
(series.nfo_updated_at.replace(tzinfo=timezone.utc) - now)
|
||||
.total_seconds()
|
||||
)
|
||||
assert created_delta < 1
|
||||
assert updated_delta < 1
|
||||
assert series.tmdb_id == 12345
|
||||
assert series.tvdb_id == 67890
|
||||
|
||||
def test_anime_series_update_nfo_status(self, db_session: Session):
|
||||
"""Test updating NFO status fields."""
|
||||
series = AnimeSeries(
|
||||
key="nfo-update-test",
|
||||
name="NFO Update Test",
|
||||
site="https://example.com",
|
||||
folder="/anime/nfo-update",
|
||||
)
|
||||
db_session.add(series)
|
||||
db_session.commit()
|
||||
|
||||
# Initially no NFO
|
||||
assert series.has_nfo is False
|
||||
|
||||
# Update NFO status
|
||||
now = datetime.now(timezone.utc)
|
||||
series.has_nfo = True
|
||||
series.nfo_created_at = now
|
||||
series.nfo_updated_at = now
|
||||
series.tmdb_id = 99999
|
||||
db_session.commit()
|
||||
|
||||
# Verify update
|
||||
db_session.refresh(series)
|
||||
assert series.has_nfo is True
|
||||
assert series.nfo_created_at is not None
|
||||
assert series.nfo_updated_at is not None
|
||||
assert series.tmdb_id == 99999
|
||||
|
||||
def test_anime_series_query_by_nfo_status(self, db_session: Session):
|
||||
"""Test querying series by NFO status."""
|
||||
# Create series with and without NFO
|
||||
series_with_nfo = AnimeSeries(
|
||||
key="with-nfo",
|
||||
name="Series With NFO",
|
||||
site="https://example.com",
|
||||
folder="/anime/with-nfo",
|
||||
has_nfo=True,
|
||||
tmdb_id=111,
|
||||
)
|
||||
series_without_nfo = AnimeSeries(
|
||||
key="without-nfo",
|
||||
name="Series Without NFO",
|
||||
site="https://example.com",
|
||||
folder="/anime/without-nfo",
|
||||
has_nfo=False,
|
||||
)
|
||||
|
||||
db_session.add_all([series_with_nfo, series_without_nfo])
|
||||
db_session.commit()
|
||||
|
||||
# Query series with NFO
|
||||
with_nfo = db_session.execute(
|
||||
select(AnimeSeries).where(
|
||||
AnimeSeries.has_nfo == True # noqa: E712
|
||||
)
|
||||
).scalars().all()
|
||||
assert len(with_nfo) == 1
|
||||
assert with_nfo[0].key == "with-nfo"
|
||||
|
||||
# Query series without NFO
|
||||
without_nfo = db_session.execute(
|
||||
select(AnimeSeries).where(
|
||||
AnimeSeries.has_nfo == False # noqa: E712
|
||||
)
|
||||
).scalars().all()
|
||||
assert len(without_nfo) == 1
|
||||
assert without_nfo[0].key == "without-nfo"
|
||||
|
||||
def test_anime_series_query_by_tmdb_id(self, db_session: Session):
|
||||
"""Test querying series by TMDB ID."""
|
||||
series1 = AnimeSeries(
|
||||
key="tmdb-test-1",
|
||||
name="TMDB Test 1",
|
||||
site="https://example.com",
|
||||
folder="/anime/tmdb-1",
|
||||
tmdb_id=12345,
|
||||
)
|
||||
series2 = AnimeSeries(
|
||||
key="tmdb-test-2",
|
||||
name="TMDB Test 2",
|
||||
site="https://example.com",
|
||||
folder="/anime/tmdb-2",
|
||||
tmdb_id=67890,
|
||||
)
|
||||
series3 = AnimeSeries(
|
||||
key="tmdb-test-3",
|
||||
name="TMDB Test 3",
|
||||
site="https://example.com",
|
||||
folder="/anime/tmdb-3",
|
||||
)
|
||||
|
||||
db_session.add_all([series1, series2, series3])
|
||||
db_session.commit()
|
||||
|
||||
# Query by specific TMDB ID
|
||||
result = db_session.execute(
|
||||
select(AnimeSeries).where(AnimeSeries.tmdb_id == 12345)
|
||||
).scalar_one_or_none()
|
||||
assert result is not None
|
||||
assert result.key == "tmdb-test-1"
|
||||
|
||||
# Query series with any TMDB ID
|
||||
with_tmdb = db_session.execute(
|
||||
select(AnimeSeries).where(AnimeSeries.tmdb_id.isnot(None))
|
||||
).scalars().all()
|
||||
assert len(with_tmdb) == 2
|
||||
|
||||
|
||||
class TestEpisode:
|
||||
"""Test cases for Episode model."""
|
||||
|
||||
Reference in New Issue
Block a user