426 lines
16 KiB
Python
426 lines
16 KiB
Python
"""Unit tests for database integrity checker module.
|
|
|
|
Tests database integrity validation, orphaned record detection,
|
|
duplicate key checks, and data consistency verification.
|
|
|
|
NOTE: The database_integrity.py module has bugs in raw SQL queries:
|
|
- _check_invalid_references uses table 'episode' but actual name is 'episodes'
|
|
- _check_invalid_references uses table 'download_queue_item' but actual is 'download_queue'
|
|
- _check_duplicate_keys uses column 'anime_key' but actual column is 'key'
|
|
These bugs cause OperationalError at runtime. Tests document this behavior.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, PropertyMock, patch
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from src.server.database.base import Base
|
|
from src.server.database.models import AnimeSeries, DownloadQueueItem, Episode
|
|
|
|
|
|
@pytest.fixture
|
|
def engine():
|
|
"""Create an in-memory SQLite database engine with schema."""
|
|
eng = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(eng)
|
|
return eng
|
|
|
|
|
|
@pytest.fixture
|
|
def session(engine):
|
|
"""Create a database session for testing."""
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
sess = SessionLocal()
|
|
yield sess
|
|
sess.close()
|
|
|
|
|
|
def _make_series(session: Session, key: str = "test-anime", name: str = "Test Anime") -> AnimeSeries:
|
|
"""Helper to create and persist an AnimeSeries record."""
|
|
series = AnimeSeries(
|
|
key=key,
|
|
name=name,
|
|
site="https://aniworld.to/anime/stream/test-anime",
|
|
folder="Test Anime (2024)",
|
|
)
|
|
session.add(series)
|
|
session.commit()
|
|
session.refresh(series)
|
|
return series
|
|
|
|
|
|
def _make_episode(session: Session, series_id: int, season: int = 1, ep: int = 1) -> Episode:
|
|
"""Helper to create and persist an Episode record."""
|
|
episode = Episode(
|
|
series_id=series_id,
|
|
season=season,
|
|
episode_number=ep,
|
|
title=f"Episode {ep}",
|
|
)
|
|
session.add(episode)
|
|
session.commit()
|
|
session.refresh(episode)
|
|
return episode
|
|
|
|
|
|
class TestDatabaseIntegrityCheckerInit:
|
|
"""Tests for DatabaseIntegrityChecker initialization."""
|
|
|
|
def test_init_with_session(self, session: Session):
|
|
"""Checker initializes with provided session."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
assert checker.session is session
|
|
assert checker.issues == []
|
|
|
|
def test_init_without_session(self):
|
|
"""Checker can be created without a session."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker()
|
|
assert checker.session is None
|
|
|
|
def test_check_all_requires_session(self):
|
|
"""check_all raises ValueError when no session is set."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker()
|
|
with pytest.raises(ValueError, match="Session required"):
|
|
checker.check_all()
|
|
|
|
|
|
class TestOrphanedEpisodes:
|
|
"""Tests for orphaned episode detection."""
|
|
|
|
def test_no_orphaned_episodes_clean_db(self, session: Session):
|
|
"""Returns 0 when all episodes have parent series."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
_make_episode(session, series.id)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_orphaned_episodes()
|
|
assert count == 0
|
|
assert len(checker.issues) == 0
|
|
|
|
def test_no_episodes_returns_zero(self, session: Session):
|
|
"""Returns 0 when no episodes exist at all."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_orphaned_episodes()
|
|
assert count == 0
|
|
|
|
def test_detects_orphaned_episodes(self, session: Session):
|
|
"""Detects episodes whose series_id references nonexistent series."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
_make_episode(session, series.id, season=1, ep=1)
|
|
_make_episode(session, series.id, season=1, ep=2)
|
|
|
|
# Delete the series directly to create orphans
|
|
session.execute(
|
|
text("DELETE FROM anime_series WHERE id = :id"),
|
|
{"id": series.id},
|
|
)
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_orphaned_episodes()
|
|
assert count == 2
|
|
assert any("orphaned episodes" in issue for issue in checker.issues)
|
|
|
|
|
|
class TestOrphanedQueueItems:
|
|
"""Tests for orphaned download queue item detection."""
|
|
|
|
def test_no_orphaned_queue_items(self, session: Session):
|
|
"""Returns 0 when all queue items have parent series."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
episode = _make_episode(session, series.id)
|
|
item = DownloadQueueItem(
|
|
series_id=series.id,
|
|
episode_id=episode.id,
|
|
)
|
|
session.add(item)
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_orphaned_queue_items()
|
|
assert count == 0
|
|
|
|
def test_detects_orphaned_queue_items(self, session: Session):
|
|
"""Detects queue items whose series references no longer exist."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
episode = _make_episode(session, series.id)
|
|
item = DownloadQueueItem(
|
|
series_id=series.id,
|
|
episode_id=episode.id,
|
|
)
|
|
session.add(item)
|
|
session.commit()
|
|
|
|
# Remove series but keep orphaned items via raw SQL
|
|
session.execute(text("DELETE FROM anime_series WHERE id = :id"), {"id": series.id})
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_orphaned_queue_items()
|
|
assert count == 1
|
|
assert any("orphaned queue" in issue for issue in checker.issues)
|
|
|
|
|
|
class TestInvalidReferences:
|
|
"""Tests for invalid foreign key reference detection.
|
|
|
|
NOTE: The raw SQL in _check_invalid_references uses wrong table names:
|
|
- 'episode' instead of 'episodes'
|
|
- 'download_queue_item' instead of 'download_queue'
|
|
This causes OperationalError in SQLite.
|
|
"""
|
|
|
|
def test_invalid_references_raw_sql_uses_wrong_table_names(
|
|
self, session: Session
|
|
):
|
|
"""BUG: Raw SQL references 'episode' and 'download_queue_item'
|
|
but actual table names are 'episodes' and 'download_queue'.
|
|
This causes the check to error and return -1."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
result = checker._check_invalid_references()
|
|
# Returns -1 because the SQL fails with OperationalError
|
|
assert result == -1
|
|
assert any("Error checking invalid references" in i for i in checker.issues)
|
|
|
|
|
|
class TestDuplicateKeys:
|
|
"""Tests for duplicate primary key detection.
|
|
|
|
NOTE: The raw SQL in _check_duplicate_keys references column 'anime_key'
|
|
but the actual column name is 'key'. This causes OperationalError.
|
|
"""
|
|
|
|
def test_duplicate_keys_raw_sql_uses_wrong_column_name(
|
|
self, session: Session
|
|
):
|
|
"""BUG: Raw SQL references 'anime_key' column but actual column
|
|
is 'key'. This causes the check to error and return -1."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
result = checker._check_duplicate_keys()
|
|
# Returns -1 because the SQL fails on nonexistent column
|
|
assert result == -1
|
|
assert any("Error checking duplicate keys" in i for i in checker.issues)
|
|
|
|
|
|
class TestDataConsistency:
|
|
"""Tests for data consistency validation."""
|
|
|
|
def test_no_consistency_issues_clean_data(self, session: Session):
|
|
"""Returns 0 for valid episode data."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
_make_episode(session, series.id, season=1, ep=1)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_data_consistency()
|
|
assert count == 0
|
|
|
|
def test_detects_negative_season_number(self, session: Session):
|
|
"""Detects episodes with negative season numbers."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
# Insert invalid episode bypassing ORM validation
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO episodes (series_id, season, episode_number, is_downloaded) "
|
|
"VALUES (:sid, :season, :ep, 0)"
|
|
),
|
|
{"sid": series.id, "season": -1, "ep": 1},
|
|
)
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_data_consistency()
|
|
assert count == 1
|
|
assert any("invalid" in i.lower() for i in checker.issues)
|
|
|
|
def test_detects_negative_episode_number(self, session: Session):
|
|
"""Detects episodes with negative episode numbers."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO episodes (series_id, season, episode_number, is_downloaded) "
|
|
"VALUES (:sid, :season, :ep, 0)"
|
|
),
|
|
{"sid": series.id, "season": 1, "ep": -5},
|
|
)
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_data_consistency()
|
|
assert count == 1
|
|
|
|
def test_empty_database_no_issues(self, session: Session):
|
|
"""Empty database reports no consistency issues."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
count = checker._check_data_consistency()
|
|
assert count == 0
|
|
|
|
|
|
class TestCheckAll:
|
|
"""Tests for the check_all aggregation method."""
|
|
|
|
def test_check_all_returns_dict_with_expected_keys(self, session: Session):
|
|
"""check_all returns result dict with all expected keys."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
results = checker.check_all()
|
|
expected_keys = {
|
|
"orphaned_episodes",
|
|
"orphaned_queue_items",
|
|
"invalid_references",
|
|
"duplicate_keys",
|
|
"data_consistency",
|
|
"total_issues",
|
|
"issues",
|
|
}
|
|
assert set(results.keys()) == expected_keys
|
|
|
|
def test_check_all_aggregates_issues(self, session: Session):
|
|
"""check_all total_issues reflects all discovered issues."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
results = checker.check_all()
|
|
# At minimum, invalid_references and duplicate_keys fail due to SQL bugs
|
|
assert results["total_issues"] >= 2
|
|
assert len(results["issues"]) >= 2
|
|
|
|
def test_check_all_resets_issues_list(self, session: Session):
|
|
"""check_all clears previous issues before running."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
checker.issues = ["leftover issue"]
|
|
checker.check_all()
|
|
# Issues list should not contain the old "leftover issue"
|
|
assert "leftover issue" not in checker.issues
|
|
|
|
|
|
class TestRepairOrphanedRecords:
|
|
"""Tests for repair_orphaned_records method."""
|
|
|
|
def test_repair_requires_session(self):
|
|
"""repair_orphaned_records raises ValueError without session."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
checker = DatabaseIntegrityChecker()
|
|
with pytest.raises(ValueError, match="Session required"):
|
|
checker.repair_orphaned_records()
|
|
|
|
def test_repair_removes_orphaned_episodes(self, session: Session):
|
|
"""repair_orphaned_records removes episodes without parent series."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
_make_episode(session, series.id, season=1, ep=1)
|
|
_make_episode(session, series.id, season=1, ep=2)
|
|
|
|
# Create orphans
|
|
session.execute(
|
|
text("DELETE FROM anime_series WHERE id = :id"),
|
|
{"id": series.id},
|
|
)
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
removed = checker.repair_orphaned_records()
|
|
assert removed == 2
|
|
|
|
# Verify episodes are gone
|
|
count = session.execute(text("SELECT COUNT(*) FROM episodes")).scalar()
|
|
assert count == 0
|
|
|
|
def test_repair_removes_orphaned_queue_items(self, session: Session):
|
|
"""repair_orphaned_records removes queue items without parent series."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
episode = _make_episode(session, series.id)
|
|
item = DownloadQueueItem(series_id=series.id, episode_id=episode.id)
|
|
session.add(item)
|
|
session.commit()
|
|
|
|
session.execute(
|
|
text("DELETE FROM anime_series WHERE id = :id"),
|
|
{"id": series.id},
|
|
)
|
|
session.commit()
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
removed = checker.repair_orphaned_records()
|
|
assert removed >= 1
|
|
|
|
def test_repair_no_orphans_returns_zero(self, session: Session):
|
|
"""repair_orphaned_records returns 0 when no orphans exist."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
DatabaseIntegrityChecker,
|
|
)
|
|
series = _make_series(session)
|
|
_make_episode(session, series.id)
|
|
|
|
checker = DatabaseIntegrityChecker(session=session)
|
|
removed = checker.repair_orphaned_records()
|
|
assert removed == 0
|
|
|
|
|
|
class TestConvenienceFunction:
|
|
"""Tests for check_database_integrity convenience function."""
|
|
|
|
def test_check_database_integrity_returns_results(self, session: Session):
|
|
"""Convenience function returns check results dict."""
|
|
from src.infrastructure.security.database_integrity import (
|
|
check_database_integrity,
|
|
)
|
|
results = check_database_integrity(session)
|
|
assert isinstance(results, dict)
|
|
assert "total_issues" in results
|
|
assert "issues" in results
|