Lukas
cbd53ef2a0
- Add migration_legacy_files_completed flag to SystemSettings model - Create legacy_file_migration service to migrate series from key/data files - Integrate legacy migration into initialization_service startup flow - Add integration tests for legacy file migration - Update DATABASE.md documentation with migration details - Fix various test and service issues (nfo_repair, tmdb_client, download_service) - Add test_database_schema unit tests
389 lines
13 KiB
Python
389 lines
13 KiB
Python
"""Unit tests for database schema verification.
|
|
|
|
Tests that the database schema supports all fields that were previously
|
|
stored in file-based storage (key/data files).
|
|
|
|
Ref: Task 1 - Verify Database Schema Supports All File-Based Data
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from src.server.database.base import Base
|
|
from src.server.database.models import AnimeSeries, Episode
|
|
|
|
|
|
@pytest.fixture
|
|
def db_engine():
|
|
"""Create in-memory SQLite database engine 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 TestAnimeSeriesHasAllRequiredFields:
|
|
"""Verify AnimeSeries model has all Serie properties."""
|
|
|
|
def test_anime_series_has_id_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has an id primary key column."""
|
|
series = AnimeSeries(
|
|
key="test-key",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.id).where(AnimeSeries.key == "test-key"))
|
|
assert result.scalar_one_or_none() is not None
|
|
|
|
def test_anime_series_has_key_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has a key column for provider identifier."""
|
|
series = AnimeSeries(
|
|
key="unique-provider-key",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.key).where(AnimeSeries.key == "unique-provider-key"))
|
|
assert result.scalar_one_or_none() == "unique-provider-key"
|
|
|
|
def test_anime_series_has_name_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has a name column."""
|
|
series = AnimeSeries(
|
|
key="name-test",
|
|
name="My Custom Name",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.name).where(AnimeSeries.key == "name-test"))
|
|
assert result.scalar_one_or_none() == "My Custom Name"
|
|
|
|
def test_anime_series_has_site_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has a site column."""
|
|
series = AnimeSeries(
|
|
key="site-test",
|
|
name="Test Series",
|
|
site="https://aniworld.to/watch/series",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.site).where(AnimeSeries.key == "site-test"))
|
|
assert result.scalar_one_or_none() == "https://aniworld.to/watch/series"
|
|
|
|
def test_anime_series_has_folder_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has a folder column."""
|
|
series = AnimeSeries(
|
|
key="folder-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/My Series Folder (2024)",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.folder).where(AnimeSeries.key == "folder-test"))
|
|
assert result.scalar_one_or_none() == "/anime/My Series Folder (2024)"
|
|
|
|
def test_anime_series_has_year_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has an optional year column."""
|
|
series = AnimeSeries(
|
|
key="year-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
year=2024,
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.year).where(AnimeSeries.key == "year-test"))
|
|
assert result.scalar_one_or_none() == 2024
|
|
|
|
def test_anime_series_year_is_nullable(self, db_session: Session):
|
|
"""Test that year column is optional (nullable)."""
|
|
series = AnimeSeries(
|
|
key="no-year-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.year).where(AnimeSeries.key == "no-year-test"))
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
def test_anime_series_has_nfo_path_column(self, db_session: Session):
|
|
"""Test that AnimeSeries has an optional nfo_path column."""
|
|
series = AnimeSeries(
|
|
key="nfo-path-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
nfo_path="/anime/test/tvshow.nfo",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.nfo_path).where(AnimeSeries.key == "nfo-path-test"))
|
|
assert result.scalar_one_or_none() == "/anime/test/tvshow.nfo"
|
|
|
|
def test_anime_series_nfo_path_is_nullable(self, db_session: Session):
|
|
"""Test that nfo_path column is optional (nullable)."""
|
|
series = AnimeSeries(
|
|
key="no-nfo-path-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(AnimeSeries.nfo_path).where(AnimeSeries.key == "no-nfo-path-test"))
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
def test_anime_series_has_timestamps(self, db_session: Session):
|
|
"""Test that AnimeSeries has created_at and updated_at timestamps."""
|
|
series = AnimeSeries(
|
|
key="timestamps-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
assert series.created_at is not None
|
|
assert series.updated_at is not None
|
|
|
|
|
|
class TestEpisodeModelTracksMissingEpisodes:
|
|
"""Verify Episode model can store missing episodes."""
|
|
|
|
def test_episode_has_season_column(self, db_session: Session):
|
|
"""Test that Episode has a season column."""
|
|
series = AnimeSeries(
|
|
key="episode-season-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=2,
|
|
episode_number=5,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(Episode.season).where(Episode.id == episode.id))
|
|
assert result.scalar_one_or_none() == 2
|
|
|
|
def test_episode_has_episode_number_column(self, db_session: Session):
|
|
"""Test that Episode has an episode_number column."""
|
|
series = AnimeSeries(
|
|
key="episode-num-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=12,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(Episode.episode_number).where(Episode.id == episode.id))
|
|
assert result.scalar_one_or_none() == 12
|
|
|
|
def test_episode_has_is_downloaded_column(self, db_session: Session):
|
|
"""Test that Episode has an is_downloaded column."""
|
|
series = AnimeSeries(
|
|
key="downloaded-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
is_downloaded=True,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(Episode.is_downloaded).where(Episode.id == episode.id))
|
|
assert result.scalar_one_or_none() is True
|
|
|
|
def test_episode_is_downloaded_defaults_false(self, db_session: Session):
|
|
"""Test that is_downloaded defaults to False."""
|
|
series = AnimeSeries(
|
|
key="default-downloaded-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(Episode.is_downloaded).where(Episode.id == episode.id))
|
|
assert result.scalar_one_or_none() is False
|
|
|
|
def test_episode_has_series_id_foreign_key(self, db_session: Session):
|
|
"""Test that Episode has a series_id foreign key."""
|
|
series = AnimeSeries(
|
|
key="fk-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(Episode.series_id).where(Episode.id == episode.id))
|
|
assert result.scalar_one_or_none() == series.id
|
|
|
|
|
|
class TestEpisodeRelationshipFromSeries:
|
|
"""Verify Series.episodes relationship works."""
|
|
|
|
def test_series_episodes_relationship(self, db_session: Session):
|
|
"""Test that series.episodes returns all episodes."""
|
|
series = AnimeSeries(
|
|
key="episodes-rel-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode1 = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
title="First Episode",
|
|
)
|
|
episode2 = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=2,
|
|
title="Second Episode",
|
|
)
|
|
episode3 = Episode(
|
|
series_id=series.id,
|
|
season=2,
|
|
episode_number=1,
|
|
title="Season 2 Premiere",
|
|
)
|
|
db_session.add_all([episode1, episode2, episode3])
|
|
db_session.commit()
|
|
|
|
assert len(series.episodes) == 3
|
|
episode_titles = [ep.title for ep in series.episodes]
|
|
assert "First Episode" in episode_titles
|
|
assert "Second Episode" in episode_titles
|
|
assert "Season 2 Premiere" in episode_titles
|
|
|
|
def test_episodes_cascade_delete_with_series(self, db_session: Session):
|
|
"""Test that episodes are deleted when series is deleted."""
|
|
series = AnimeSeries(
|
|
key="cascade-delete-test",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
series_id = series.id
|
|
episode_id = episode.id
|
|
|
|
db_session.delete(series)
|
|
db_session.commit()
|
|
|
|
result = db_session.execute(select(Episode).where(Episode.id == episode_id))
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
def test_series_episodes_filtered_by_season(self, db_session: Session):
|
|
"""Test that episodes relationship returns all seasons."""
|
|
series = AnimeSeries(
|
|
key="multi-season-test",
|
|
name="Multi Season Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
for season in range(1, 4):
|
|
for ep_num in range(1, 4):
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=season,
|
|
episode_number=ep_num,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
assert len(series.episodes) == 9
|
|
seasons = {ep.season for ep in series.episodes}
|
|
assert seasons == {1, 2, 3}
|