Lukas
c84f968916
The 'episodes added to download queue never shown' symptom has two
layered causes that masked each other:
1. The 'episodes' table has no UNIQUE constraint on
(series_id, season, episode_number), so historical scans can leave
duplicate rows behind. AnimeSeries.episodeDict iterated the
SQLAlchemy 'episodes' relationship without deduping, so the dict
exposed duplicate entries to list_missing() and the queue UI.
The frontend forwarded the duplicated episode list verbatim to
POST /api/queue/add; the backend's pending-episode dedup then
rejected every duplicate as 'already pending' and the user saw
'Skipped 44 duplicate episodes, Added 0'.
2. selection-manager.downloadSelected counted 'episodes.length'
(the input array) instead of data.added_items.length (the
server-confirmed count). Combined with the backend returning
success on an empty add, the user saw a misleading 'Added 44
episode(s)' toast for an empty queue.
Fix: dedupe at the read boundary. The episodeDict property now
filters duplicate (season, episode_number) pairs from both the
DB-loaded relationship and the legacy _episode_dict_cache path.
Existing duplicate rows in the user's DB are inert — the read
filter makes them invisible to the rest of the stack. The
frontend now trusts the server response, logs a console warning
when an input list shrinks to zero added items, and shows an
accurate toast.
Tests:
- TestEpisodeDictDedup class with 4 regression tests covering:
* duplicate relationship rows deduped
* is_downloaded rows still filtered out
* _episode_dict_cache path also deduped (set by scanners/loaders
that may store duplicates)
* dedup is per-(season, ep_num), preserving legitimate
same-ep-num-across-different-seasons entries
Verified manually against the user's backup DB: 'erased' has 44
duplicate rows in the episodes table; episodeDict now returns
{1: [1..12]} instead of {1: [1,1,2,2,3,3,3,3,...]}, matching the
12-episode canonical list.
838 lines
26 KiB
Python
838 lines
26 KiB
Python
"""Unit tests for database models and connection management.
|
|
|
|
Tests SQLAlchemy models, relationships, session management, and database
|
|
operations. Uses an in-memory SQLite database for isolated testing.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from src.server.database.base import Base, SoftDeleteMixin, TimestampMixin
|
|
from src.server.database.models import (
|
|
AnimeSeries,
|
|
DownloadQueueItem,
|
|
Episode,
|
|
UserSession,
|
|
)
|
|
|
|
|
|
@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 TestAnimeSeries:
|
|
"""Test cases for AnimeSeries model."""
|
|
|
|
def test_create_anime_series(self, db_session: Session):
|
|
"""Test creating an anime series."""
|
|
series = AnimeSeries(
|
|
key="attack-on-titan",
|
|
name="Attack on Titan",
|
|
site="https://aniworld.to",
|
|
folder="/anime/attack-on-titan",
|
|
)
|
|
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
# Verify saved
|
|
assert series.id is not None
|
|
assert series.key == "attack-on-titan"
|
|
assert series.name == "Attack on Titan"
|
|
assert series.created_at is not None
|
|
assert series.updated_at is not None
|
|
|
|
def test_anime_series_unique_key(self, db_session: Session):
|
|
"""Test that series key must be unique."""
|
|
series1 = AnimeSeries(
|
|
key="unique-key",
|
|
name="Series 1",
|
|
site="https://example.com",
|
|
folder="/anime/series1",
|
|
)
|
|
series2 = AnimeSeries(
|
|
key="unique-key",
|
|
name="Series 2",
|
|
site="https://example.com",
|
|
folder="/anime/series2",
|
|
)
|
|
|
|
db_session.add(series1)
|
|
db_session.commit()
|
|
|
|
db_session.add(series2)
|
|
with pytest.raises(Exception): # IntegrityError
|
|
db_session.commit()
|
|
|
|
def test_anime_series_relationships(self, db_session: Session):
|
|
"""Test relationships with episodes and download items."""
|
|
series = AnimeSeries(
|
|
key="test-series",
|
|
name="Test Series",
|
|
site="https://example.com",
|
|
folder="/anime/test",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
# Add episodes
|
|
episode1 = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
title="Episode 1",
|
|
)
|
|
episode2 = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=2,
|
|
title="Episode 2",
|
|
)
|
|
db_session.add_all([episode1, episode2])
|
|
db_session.commit()
|
|
|
|
# Verify relationship
|
|
assert len(series.episodes) == 2
|
|
assert series.episodes[0].title == "Episode 1"
|
|
|
|
def test_anime_series_cascade_delete(self, db_session: Session):
|
|
"""Test that deleting series cascades to episodes."""
|
|
series = AnimeSeries(
|
|
key="cascade-test",
|
|
name="Cascade Test",
|
|
site="https://example.com",
|
|
folder="/anime/cascade",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
# Add episodes
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=1,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
series_id = series.id
|
|
|
|
# Delete series
|
|
db_session.delete(series)
|
|
db_session.commit()
|
|
|
|
# Verify episodes are deleted
|
|
result = db_session.execute(
|
|
select(Episode).where(Episode.series_id == series_id)
|
|
)
|
|
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 TestEpisodeDictDedup:
|
|
"""Regression tests for the duplicate-episode-row guard.
|
|
|
|
The ``episodes`` table has no UNIQUE constraint on
|
|
``(series_id, season, episode_number)``, so historical scans can
|
|
leave duplicate rows behind. ``AnimeSeries.episodeDict`` is the
|
|
read boundary consumed by ``list_missing()`` and the queue UI;
|
|
it must never expose duplicates, or the frontend forwards them
|
|
verbatim to the queue API and every duplicate is rejected as
|
|
"already pending" — the user sees an empty queue and a misleading
|
|
"Added N" toast.
|
|
"""
|
|
|
|
def _make_series(self, db_session: Session, key: str) -> AnimeSeries:
|
|
series = AnimeSeries(
|
|
key=key,
|
|
name=key.replace("-", " ").title(),
|
|
site="https://aniworld.to",
|
|
folder=f"/anime/{key}",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
return series
|
|
|
|
def test_episodeDict_dedupes_duplicate_relationship_rows(
|
|
self, db_session: Session
|
|
):
|
|
"""Duplicate Episode rows in the same series must not
|
|
duplicate the entries in ``episodeDict``."""
|
|
series = self._make_series(db_session, "dedup-rel")
|
|
|
|
# Three duplicate rows for (S1, E3) — exactly the pattern the
|
|
# scanner accumulates across repeated rescans.
|
|
for _ in range(3):
|
|
db_session.add(
|
|
Episode(series_id=series.id, season=1, episode_number=3)
|
|
)
|
|
# One row each for the surrounding unique episodes.
|
|
for ep in (1, 2, 4):
|
|
db_session.add(
|
|
Episode(series_id=series.id, season=1, episode_number=ep)
|
|
)
|
|
db_session.commit()
|
|
|
|
result = series.episodeDict
|
|
|
|
# Order depends on SQLAlchemy row order, which is not strictly
|
|
# insertion order — only the *set* of episodes matters here.
|
|
assert set(result.keys()) == {1}
|
|
assert set(result[1]) == {1, 2, 3, 4}
|
|
# Defensive: no season has duplicate episode numbers.
|
|
for season, eps in result.items():
|
|
assert len(eps) == len(set(eps)), (
|
|
f"season {season} has duplicate episode numbers: {eps}"
|
|
)
|
|
|
|
def test_episodeDict_excludes_downloaded_episodes(
|
|
self, db_session: Session
|
|
):
|
|
"""is_downloaded rows must still be filtered out, and a
|
|
duplicate non-downloaded row must not appear twice."""
|
|
series = self._make_series(db_session, "dedup-downloaded")
|
|
|
|
db_session.add(
|
|
Episode(series_id=series.id, season=1, episode_number=1)
|
|
)
|
|
# Downloaded row at (S1, E2): must be filtered out.
|
|
db_session.add(
|
|
Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=2,
|
|
is_downloaded=True,
|
|
)
|
|
)
|
|
db_session.commit()
|
|
|
|
assert series.episodeDict == {1: [1]}
|
|
|
|
def test_episodeDict_dedupes_cached_value(self, db_session: Session):
|
|
"""The legacy ``_episode_dict_cache`` path (used by
|
|
``SerieScanner.scan_single_series`` for new series and by the
|
|
legacy JSON loader) must also dedupe, since loaders that set
|
|
the cache directly can introduce duplicates too."""
|
|
series = self._make_series(db_session, "dedup-cache")
|
|
|
|
# No episodes in the DB at all — the property will fall back
|
|
# to the cache.
|
|
series._episode_dict_cache = {1: [3, 3, 3, 4, 4]}
|
|
|
|
assert series.episodeDict == {1: [3, 4]}
|
|
|
|
def test_episodeDict_preserves_unique_entries_across_seasons(
|
|
self, db_session: Session
|
|
):
|
|
"""Dedup must be per-(season, episode_number), not
|
|
per-episode_number alone — same ep number in different
|
|
seasons is legitimate and must be preserved."""
|
|
series = self._make_series(db_session, "multi-season")
|
|
|
|
for season, ep in [(1, 1), (1, 2), (2, 1), (2, 2), (2, 2)]:
|
|
db_session.add(
|
|
Episode(
|
|
series_id=series.id,
|
|
season=season,
|
|
episode_number=ep,
|
|
)
|
|
)
|
|
db_session.commit()
|
|
|
|
result = series.episodeDict
|
|
|
|
# Order is not guaranteed across SQLAlchemy relationships;
|
|
# compare as sets.
|
|
assert set(result.keys()) == {1, 2}
|
|
assert set(result[1]) == {1, 2}
|
|
assert set(result[2]) == {1, 2}
|
|
|
|
|
|
class TestEpisode:
|
|
"""Test cases for Episode model."""
|
|
|
|
def test_create_episode(self, db_session: Session):
|
|
"""Test creating an episode."""
|
|
series = AnimeSeries(
|
|
key="test-series",
|
|
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=5,
|
|
title="The Fifth Episode",
|
|
file_path="/anime/test/S01E05.mp4",
|
|
is_downloaded=True,
|
|
)
|
|
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
# Verify saved
|
|
assert episode.id is not None
|
|
assert episode.season == 1
|
|
assert episode.episode_number == 5
|
|
assert episode.is_downloaded is True
|
|
assert episode.created_at is not None
|
|
|
|
def test_episode_relationship_to_series(self, db_session: Session):
|
|
"""Test episode relationship to series."""
|
|
series = AnimeSeries(
|
|
key="relationship-test",
|
|
name="Relationship Test",
|
|
site="https://example.com",
|
|
folder="/anime/relationship",
|
|
)
|
|
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()
|
|
|
|
# Verify relationship
|
|
assert episode.series.name == "Relationship Test"
|
|
assert episode.series.key == "relationship-test"
|
|
|
|
|
|
class TestDownloadQueueItem:
|
|
"""Test cases for DownloadQueueItem model."""
|
|
|
|
def test_create_download_item(self, db_session: Session):
|
|
"""Test creating a download queue item."""
|
|
series = AnimeSeries(
|
|
key="download-test",
|
|
name="Download Test",
|
|
site="https://example.com",
|
|
folder="/anime/download",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=3,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
item = DownloadQueueItem(
|
|
series_id=series.id,
|
|
episode_id=episode.id,
|
|
download_url="https://example.com/download/ep3",
|
|
file_destination="/anime/download/S01E03.mp4",
|
|
)
|
|
|
|
db_session.add(item)
|
|
db_session.commit()
|
|
|
|
# Verify saved
|
|
assert item.id is not None
|
|
assert item.episode_id == episode.id
|
|
assert item.series_id == series.id
|
|
|
|
def test_download_item_episode_relationship(self, db_session: Session):
|
|
"""Test download item episode relationship."""
|
|
series = AnimeSeries(
|
|
key="relationship-test",
|
|
name="Relationship Test",
|
|
site="https://example.com",
|
|
folder="/anime/relationship",
|
|
)
|
|
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()
|
|
|
|
item = DownloadQueueItem(
|
|
series_id=series.id,
|
|
episode_id=episode.id,
|
|
)
|
|
db_session.add(item)
|
|
db_session.commit()
|
|
|
|
# Verify relationship
|
|
assert item.episode.id == episode.id
|
|
assert item.series.id == series.id
|
|
|
|
def test_download_item_error_handling(self, db_session: Session):
|
|
"""Test download item with error information."""
|
|
series = AnimeSeries(
|
|
key="error-test",
|
|
name="Error Test",
|
|
site="https://example.com",
|
|
folder="/anime/error",
|
|
)
|
|
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()
|
|
|
|
item = DownloadQueueItem(
|
|
series_id=series.id,
|
|
episode_id=episode.id,
|
|
error_message="Network timeout after 30 seconds",
|
|
)
|
|
db_session.add(item)
|
|
db_session.commit()
|
|
|
|
# Verify error info
|
|
assert item.error_message == "Network timeout after 30 seconds"
|
|
|
|
|
|
class TestUserSession:
|
|
"""Test cases for UserSession model."""
|
|
|
|
def test_create_user_session(self, db_session: Session):
|
|
"""Test creating a user session."""
|
|
expires = datetime.now(timezone.utc) + timedelta(hours=24)
|
|
|
|
session = UserSession(
|
|
session_id="test-session-123",
|
|
token_hash="hashed-token-value",
|
|
user_id="user-1",
|
|
ip_address="192.168.1.100",
|
|
user_agent="Mozilla/5.0",
|
|
expires_at=expires,
|
|
is_active=True,
|
|
)
|
|
|
|
db_session.add(session)
|
|
db_session.commit()
|
|
|
|
# Verify saved
|
|
assert session.id is not None
|
|
assert session.session_id == "test-session-123"
|
|
assert session.is_active is True
|
|
assert session.created_at is not None
|
|
|
|
def test_session_unique_session_id(self, db_session: Session):
|
|
"""Test that session_id must be unique."""
|
|
expires = datetime.now(timezone.utc) + timedelta(hours=24)
|
|
|
|
session1 = UserSession(
|
|
session_id="duplicate-id",
|
|
token_hash="hash1",
|
|
expires_at=expires,
|
|
)
|
|
session2 = UserSession(
|
|
session_id="duplicate-id",
|
|
token_hash="hash2",
|
|
expires_at=expires,
|
|
)
|
|
|
|
db_session.add(session1)
|
|
db_session.commit()
|
|
|
|
db_session.add(session2)
|
|
with pytest.raises(Exception): # IntegrityError
|
|
db_session.commit()
|
|
|
|
def test_session_is_expired(self, db_session: Session):
|
|
"""Test session expiration check."""
|
|
# Create expired session
|
|
expired = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
session = UserSession(
|
|
session_id="expired-session",
|
|
token_hash="hash",
|
|
expires_at=expired,
|
|
)
|
|
|
|
db_session.add(session)
|
|
db_session.commit()
|
|
|
|
# Verify is_expired
|
|
assert session.is_expired is True
|
|
|
|
def test_session_revoke(self, db_session: Session):
|
|
"""Test session revocation."""
|
|
expires = datetime.now(timezone.utc) + timedelta(hours=24)
|
|
session = UserSession(
|
|
session_id="revoke-test",
|
|
token_hash="hash",
|
|
expires_at=expires,
|
|
is_active=True,
|
|
)
|
|
|
|
db_session.add(session)
|
|
db_session.commit()
|
|
|
|
# Revoke session
|
|
session.revoke()
|
|
db_session.commit()
|
|
|
|
# Verify revoked
|
|
assert session.is_active is False
|
|
|
|
|
|
class TestTimestampMixin:
|
|
"""Test cases for TimestampMixin."""
|
|
|
|
def test_timestamp_auto_creation(self, db_session: Session):
|
|
"""Test that timestamps are automatically created."""
|
|
series = AnimeSeries(
|
|
key="timestamp-test",
|
|
name="Timestamp Test",
|
|
site="https://example.com",
|
|
folder="/anime/timestamp",
|
|
)
|
|
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
# Verify timestamps exist
|
|
assert series.created_at is not None
|
|
assert series.updated_at is not None
|
|
assert series.created_at == series.updated_at
|
|
|
|
def test_timestamp_auto_update(self, db_session: Session):
|
|
"""Test that updated_at is automatically updated."""
|
|
series = AnimeSeries(
|
|
key="update-test",
|
|
name="Update Test",
|
|
site="https://example.com",
|
|
folder="/anime/update",
|
|
)
|
|
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
original_updated = series.updated_at
|
|
|
|
# Update and save
|
|
series.name = "Updated Name"
|
|
db_session.commit()
|
|
|
|
# Verify updated_at changed
|
|
# Note: This test may be flaky due to timing
|
|
assert series.created_at is not None
|
|
|
|
|
|
class TestSoftDeleteMixin:
|
|
"""Test cases for SoftDeleteMixin."""
|
|
|
|
def test_soft_delete_not_applied_to_models(self):
|
|
"""Test that SoftDeleteMixin is not applied to current models.
|
|
|
|
This is a documentation test - models don't currently use
|
|
SoftDeleteMixin, but it's available for future use.
|
|
"""
|
|
# Verify models don't have deleted_at attribute
|
|
series = AnimeSeries(
|
|
key="soft-delete-test",
|
|
name="Soft Delete Test",
|
|
site="https://example.com",
|
|
folder="/anime/soft-delete",
|
|
)
|
|
|
|
# Models shouldn't have soft delete attributes
|
|
assert not hasattr(series, "deleted_at")
|
|
assert not hasattr(series, "is_deleted")
|
|
assert not hasattr(series, "soft_delete")
|
|
|
|
|
|
class TestDatabaseQueries:
|
|
"""Test complex database queries and operations."""
|
|
|
|
def test_query_series_with_episodes(self, db_session: Session):
|
|
"""Test querying series with their episodes."""
|
|
# Create series with episodes
|
|
series = AnimeSeries(
|
|
key="query-test",
|
|
name="Query Test",
|
|
site="https://example.com",
|
|
folder="/anime/query",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
# Add multiple episodes
|
|
for i in range(1, 6):
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=i,
|
|
title=f"Episode {i}",
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
# Query series with episodes
|
|
result = db_session.execute(
|
|
select(AnimeSeries).where(AnimeSeries.key == "query-test")
|
|
)
|
|
queried_series = result.scalar_one()
|
|
|
|
# Verify episodes loaded
|
|
assert len(queried_series.episodes) == 5
|
|
|
|
def test_query_download_queue_by_status(self, db_session: Session):
|
|
"""Test querying download queue by status."""
|
|
series = AnimeSeries(
|
|
key="queue-query-test",
|
|
name="Queue Query Test",
|
|
site="https://example.com",
|
|
folder="/anime/queue-query",
|
|
)
|
|
db_session.add(series)
|
|
db_session.commit()
|
|
|
|
# Create episodes and items
|
|
for i in range(3):
|
|
episode = Episode(
|
|
series_id=series.id,
|
|
season=1,
|
|
episode_number=i + 1,
|
|
)
|
|
db_session.add(episode)
|
|
db_session.commit()
|
|
|
|
item = DownloadQueueItem(
|
|
series_id=series.id,
|
|
episode_id=episode.id,
|
|
)
|
|
db_session.add(item)
|
|
db_session.commit()
|
|
|
|
# Query all items
|
|
result = db_session.execute(
|
|
select(DownloadQueueItem)
|
|
)
|
|
items = result.scalars().all()
|
|
|
|
# Verify query
|
|
assert len(items) == 3
|
|
|
|
def test_query_active_sessions(self, db_session: Session):
|
|
"""Test querying active user sessions."""
|
|
expires = datetime.now(timezone.utc) + timedelta(hours=24)
|
|
|
|
# Create active and inactive sessions
|
|
active = UserSession(
|
|
session_id="active-1",
|
|
token_hash="hash1",
|
|
expires_at=expires,
|
|
is_active=True,
|
|
)
|
|
inactive = UserSession(
|
|
session_id="inactive-1",
|
|
token_hash="hash2",
|
|
expires_at=expires,
|
|
is_active=False,
|
|
)
|
|
|
|
db_session.add_all([active, inactive])
|
|
db_session.commit()
|
|
|
|
# Query active sessions
|
|
result = db_session.execute(
|
|
select(UserSession).where(UserSession.is_active == True)
|
|
)
|
|
active_sessions = result.scalars().all()
|
|
|
|
# Verify query
|
|
assert len(active_sessions) == 1
|
|
assert active_sessions[0].session_id == "active-1"
|