lukas.pupkalipinski/Aniworld
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Aniworld/tests/unit/test_database_models.py
Lukas 17e5a551e1 feat: migrate to Pydantic V2 and implement rate limiting middleware 
- Migrate settings.py to Pydantic V2 (SettingsConfigDict, validation_alias)
- Update config models to use @field_validator with @classmethod
- Replace deprecated datetime.utcnow() with datetime.now(timezone.utc)
- Migrate FastAPI app from @app.on_event to lifespan context manager
- Implement comprehensive rate limiting middleware with:
  * Endpoint-specific rate limits (login: 5/min, register: 3/min)
  * IP-based and user-based tracking
  * Authenticated user multiplier (2x limits)
  * Bypass paths for health, docs, static, websocket endpoints
  * Rate limit headers in responses
- Add 13 comprehensive tests for rate limiting (all passing)
- Update instructions.md to mark completed tasks
- Fix asyncio.create_task usage in anime_service.py

All 714 tests passing. No deprecation warnings.
2025-10-23 22:03:15 +02:00

562 lines
17 KiB
Python
Raw Blame History

"""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,
DownloadPriority,
DownloadQueueItem,
DownloadStatus,
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",
description="Epic anime about titans",
status="completed",
total_episodes=75,
cover_url="https://example.com/cover.jpg",
episode_dict={1: [1, 2, 3], 2: [1, 2, 3, 4]},
)
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",
file_size=524288000, # 500 MB
is_downloaded=True,
download_date=datetime.now(timezone.utc),
)
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()
item = DownloadQueueItem(
series_id=series.id,
season=1,
episode_number=3,
status=DownloadStatus.DOWNLOADING,
priority=DownloadPriority.HIGH,
progress_percent=45.5,
downloaded_bytes=250000000,
total_bytes=550000000,
download_speed=2500000.0,
retry_count=0,
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.status == DownloadStatus.DOWNLOADING
assert item.priority == DownloadPriority.HIGH
assert item.progress_percent == 45.5
assert item.retry_count == 0
def test_download_item_status_enum(self, db_session: Session):
"""Test download status enum values."""
series = AnimeSeries(
key="status-test",
name="Status Test",
site="https://example.com",
folder="/anime/status",
)
db_session.add(series)
db_session.commit()
item = DownloadQueueItem(
series_id=series.id,
season=1,
episode_number=1,
status=DownloadStatus.PENDING,
)
db_session.add(item)
db_session.commit()
# Update status
item.status = DownloadStatus.COMPLETED
db_session.commit()
# Verify status change
assert item.status == DownloadStatus.COMPLETED
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()
item = DownloadQueueItem(
series_id=series.id,
season=1,
episode_number=1,
status=DownloadStatus.FAILED,
error_message="Network timeout after 30 seconds",
retry_count=2,
)
db_session.add(item)
db_session.commit()
# Verify error info
assert item.status == DownloadStatus.FAILED
assert item.error_message == "Network timeout after 30 seconds"
assert item.retry_count == 2
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 items with different statuses
for i, status in enumerate([
DownloadStatus.PENDING,
DownloadStatus.DOWNLOADING,
DownloadStatus.COMPLETED,
]):
item = DownloadQueueItem(
series_id=series.id,
season=1,
episode_number=i + 1,
status=status,
)
db_session.add(item)
db_session.commit()
# Query pending items
result = db_session.execute(
select(DownloadQueueItem).where(
DownloadQueueItem.status == DownloadStatus.PENDING
)
)
pending = result.scalars().all()
# Verify query
assert len(pending) == 1
assert pending[0].episode_number == 1
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"
Reference in New Issue View Git Blame Copy Permalink