313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""Integration tests for database edge cases.
|
|
|
|
Tests boundary values, foreign key constraints, model validation, session
|
|
lifecycle, and large batch operations on database models.
|
|
"""
|
|
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from src.server.database.models import (
|
|
AnimeSeries,
|
|
DownloadPriority,
|
|
DownloadQueueItem,
|
|
DownloadStatus,
|
|
Episode,
|
|
SystemSettings,
|
|
UserSession,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Boundary value tests for AnimeSeries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAnimeSeriesBoundaries:
|
|
"""Boundary conditions for AnimeSeries model."""
|
|
|
|
def test_empty_key_rejected(self):
|
|
"""Empty string key triggers validation error."""
|
|
with pytest.raises((ValueError, Exception)):
|
|
AnimeSeries(key="", name="Test", site="https://x.com", folder="Test")
|
|
|
|
def test_max_length_key(self):
|
|
"""Key at max length (255) is accepted."""
|
|
key = "a" * 255
|
|
a = AnimeSeries(key=key, name="Test", site="https://x.com", folder="Test")
|
|
assert len(a.key) == 255
|
|
|
|
def test_empty_name_rejected(self):
|
|
"""Empty name triggers validation error."""
|
|
with pytest.raises((ValueError, Exception)):
|
|
AnimeSeries(key="k", name="", site="https://x.com", folder="Test")
|
|
|
|
def test_long_name(self):
|
|
"""Name up to 500 chars is accepted."""
|
|
name = "X" * 500
|
|
a = AnimeSeries(key="k", name=name, site="https://x.com", folder="F")
|
|
assert len(a.name) == 500
|
|
|
|
def test_unicode_name(self):
|
|
"""Unicode characters in name are stored correctly."""
|
|
a = AnimeSeries(
|
|
key="unicode-test",
|
|
name="進撃の巨人 Attack on Titan",
|
|
site="https://x.com",
|
|
folder="AOT",
|
|
)
|
|
assert "進撃の巨人" in a.name
|
|
|
|
def test_default_values(self):
|
|
"""Default booleans and nullables are set correctly."""
|
|
a = AnimeSeries(key="def", name="Def", site="https://x.com", folder="D")
|
|
# Before DB insert, mapped_column defaults may not be applied
|
|
assert a.has_nfo in (None, False)
|
|
assert a.year is None
|
|
assert a.tmdb_id is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Episode boundary values
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEpisodeBoundaries:
|
|
"""Boundary conditions for Episode model."""
|
|
|
|
def test_min_season_and_episode(self):
|
|
"""Season 0, episode 0 are valid (specials/movies)."""
|
|
ep = Episode(series_id=1, season=0, episode_number=0, title="Special")
|
|
assert ep.season == 0
|
|
assert ep.episode_number == 0
|
|
|
|
def test_max_season_and_episode(self):
|
|
"""Maximum allowed values for season and episode."""
|
|
ep = Episode(series_id=1, season=1000, episode_number=10000, title="Max")
|
|
assert ep.season == 1000
|
|
assert ep.episode_number == 10000
|
|
|
|
def test_negative_season_rejected(self):
|
|
"""Negative season triggers validation error."""
|
|
with pytest.raises((ValueError, Exception)):
|
|
Episode(series_id=1, season=-1, episode_number=1, title="Bad")
|
|
|
|
def test_negative_episode_rejected(self):
|
|
"""Negative episode number triggers validation error."""
|
|
with pytest.raises((ValueError, Exception)):
|
|
Episode(series_id=1, season=1, episode_number=-1, title="Bad")
|
|
|
|
def test_empty_title(self):
|
|
"""Empty title may be allowed (depends on implementation)."""
|
|
# Some episodes don't have titles
|
|
ep = Episode(series_id=1, season=1, episode_number=1, title="")
|
|
assert ep.title == ""
|
|
|
|
def test_long_file_path(self):
|
|
"""file_path up to 1000 chars is accepted."""
|
|
path = "/a/b/" + "c" * 990
|
|
ep = Episode(
|
|
series_id=1, season=1, episode_number=1,
|
|
title="T", file_path=path
|
|
)
|
|
assert len(ep.file_path) == 995
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DownloadQueueItem edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDownloadQueueItemEdgeCases:
|
|
"""Edge cases for DownloadQueueItem model."""
|
|
|
|
def test_all_status_values(self):
|
|
"""Every DownloadStatus enum member has a string value."""
|
|
for st in DownloadStatus:
|
|
assert isinstance(st.value, str)
|
|
# Verify expected members exist
|
|
assert DownloadStatus.PENDING.value == "pending"
|
|
assert DownloadStatus.DOWNLOADING.value == "downloading"
|
|
assert DownloadStatus.COMPLETED.value == "completed"
|
|
assert DownloadStatus.FAILED.value == "failed"
|
|
|
|
def test_all_priority_values(self):
|
|
"""Every DownloadPriority enum member has a string value."""
|
|
for p in DownloadPriority:
|
|
assert isinstance(p.value, str)
|
|
# Verify expected members exist
|
|
assert DownloadPriority.LOW.value == "low"
|
|
assert DownloadPriority.NORMAL.value == "normal"
|
|
assert DownloadPriority.HIGH.value == "high"
|
|
|
|
def test_error_message_can_be_none(self):
|
|
"""error_message defaults to None."""
|
|
item = DownloadQueueItem(
|
|
series_id=1,
|
|
episode_id=1,
|
|
download_url="https://example.com",
|
|
file_destination="/tmp/ep.mp4",
|
|
)
|
|
assert item.error_message is None
|
|
|
|
def test_long_error_message(self):
|
|
"""A very long error message is stored."""
|
|
msg = "Error: " + "x" * 2000
|
|
item = DownloadQueueItem(
|
|
series_id=1,
|
|
episode_id=1,
|
|
download_url="https://example.com",
|
|
file_destination="/tmp/ep.mp4",
|
|
error_message=msg,
|
|
)
|
|
assert len(item.error_message) > 2000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UserSession edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUserSessionEdgeCases:
|
|
"""Edge cases for UserSession model."""
|
|
|
|
def test_is_expired_property(self):
|
|
"""Session expired when expires_at is in the past."""
|
|
session = UserSession(
|
|
session_id="sess1",
|
|
token_hash="hash",
|
|
user_id="user1",
|
|
ip_address="127.0.0.1",
|
|
user_agent="test",
|
|
expires_at=datetime.now(timezone.utc) - timedelta(hours=1),
|
|
is_active=True,
|
|
)
|
|
assert session.is_expired is True
|
|
|
|
def test_not_expired(self):
|
|
"""Session not expired when expires_at is in the future."""
|
|
session = UserSession(
|
|
session_id="sess2",
|
|
token_hash="hash",
|
|
user_id="user1",
|
|
ip_address="127.0.0.1",
|
|
user_agent="test",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
is_active=True,
|
|
)
|
|
assert session.is_expired is False
|
|
|
|
def test_revoke_sets_inactive(self):
|
|
"""revoke() sets is_active to False."""
|
|
session = UserSession(
|
|
session_id="sess3",
|
|
token_hash="hash",
|
|
user_id="user1",
|
|
ip_address="127.0.0.1",
|
|
user_agent="test",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
is_active=True,
|
|
)
|
|
session.revoke()
|
|
assert session.is_active is False
|
|
|
|
def test_ipv6_address(self):
|
|
"""IPv6 address fits in ip_address field (max 45 chars)."""
|
|
session = UserSession(
|
|
session_id="sess4",
|
|
token_hash="hash",
|
|
user_id="user1",
|
|
ip_address="::ffff:192.168.1.1",
|
|
user_agent="test",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
is_active=True,
|
|
)
|
|
assert session.ip_address == "::ffff:192.168.1.1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SystemSettings edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSystemSettingsEdgeCases:
|
|
"""Edge cases for SystemSettings model."""
|
|
|
|
def test_default_flags(self):
|
|
"""Default boolean flags are False or None before DB insert."""
|
|
ss = SystemSettings()
|
|
# Before DB insert, mapped_column defaults may not be applied
|
|
assert ss.initial_scan_completed in (None, False)
|
|
assert ss.initial_nfo_scan_completed in (None, False)
|
|
assert ss.initial_media_scan_completed in (None, False)
|
|
|
|
def test_last_scan_timestamp_nullable(self):
|
|
"""last_scan_timestamp can be None."""
|
|
ss = SystemSettings()
|
|
assert ss.last_scan_timestamp is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Large batch simulation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestLargeBatch:
|
|
"""Creating many model instances in a batch."""
|
|
|
|
def test_create_100_episodes(self):
|
|
"""100 Episode objects can be created without error."""
|
|
episodes = [
|
|
Episode(
|
|
series_id=1,
|
|
season=1,
|
|
episode_number=i,
|
|
title=f"Episode {i}",
|
|
)
|
|
for i in range(1, 101)
|
|
]
|
|
assert len(episodes) == 100
|
|
assert episodes[-1].episode_number == 100
|
|
|
|
def test_create_100_download_items(self):
|
|
"""100 DownloadQueueItem objects can be created."""
|
|
items = [
|
|
DownloadQueueItem(
|
|
series_id=1,
|
|
episode_id=i,
|
|
download_url=f"https://example.com/{i}",
|
|
file_destination=f"/tmp/ep{i}.mp4",
|
|
)
|
|
for i in range(100)
|
|
]
|
|
assert len(items) == 100
|
|
# All URLs unique
|
|
urls = {item.download_url for item in items}
|
|
assert len(urls) == 100
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Foreign key reference integrity (model-level)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestForeignKeyReferences:
|
|
"""Verify FK fields accept valid values."""
|
|
|
|
def test_episode_series_id(self):
|
|
"""Episode.series_id can reference any positive integer."""
|
|
ep = Episode(series_id=999, season=1, episode_number=1, title="T")
|
|
assert ep.series_id == 999
|
|
|
|
def test_download_item_references(self):
|
|
"""DownloadQueueItem links to series_id and episode_id."""
|
|
item = DownloadQueueItem(
|
|
series_id=42,
|
|
episode_id=7,
|
|
download_url="https://example.com",
|
|
file_destination="/tmp/ep.mp4",
|
|
)
|
|
assert item.series_id == 42
|
|
assert item.episode_id == 7
|