556 lines
16 KiB
Python
556 lines
16 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
|
|
|
|
|
|
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"
|