423 lines
13 KiB
Python
423 lines
13 KiB
Python
"""Tests for JSON repositories."""
|
|
|
|
from alfred.domain.movies.entities import Movie
|
|
from alfred.domain.movies.value_objects import MovieTitle, Quality, ReleaseYear
|
|
from alfred.domain.shared.value_objects import FilePath, FileSize, ImdbId
|
|
from alfred.domain.subtitles.entities import Subtitle
|
|
from alfred.domain.subtitles.value_objects import Language, SubtitleFormat, TimingOffset
|
|
from alfred.domain.tv_shows.entities import TVShow
|
|
from alfred.domain.tv_shows.value_objects import ShowStatus
|
|
from alfred.infrastructure.persistence.json import (
|
|
JsonMovieRepository,
|
|
JsonSubtitleRepository,
|
|
JsonTVShowRepository,
|
|
)
|
|
|
|
|
|
class TestJsonMovieRepository:
|
|
"""Tests for JsonMovieRepository."""
|
|
|
|
def test_save_movie(self, memory):
|
|
"""Should save a movie."""
|
|
repo = JsonMovieRepository()
|
|
movie = Movie(
|
|
imdb_id=ImdbId("tt1375666"),
|
|
title=MovieTitle("Inception"),
|
|
release_year=ReleaseYear(2010),
|
|
quality=Quality.FULL_HD,
|
|
)
|
|
|
|
repo.save(movie)
|
|
|
|
assert len(memory.ltm.library["movies"]) == 1
|
|
assert memory.ltm.library["movies"][0]["imdb_id"] == "tt1375666"
|
|
|
|
def test_save_updates_existing(self, memory):
|
|
"""Should update existing movie."""
|
|
repo = JsonMovieRepository()
|
|
movie1 = Movie(
|
|
imdb_id=ImdbId("tt1375666"),
|
|
title=MovieTitle("Inception"),
|
|
quality=Quality.HD,
|
|
)
|
|
movie2 = Movie(
|
|
imdb_id=ImdbId("tt1375666"),
|
|
title=MovieTitle("Inception"),
|
|
quality=Quality.FULL_HD,
|
|
)
|
|
|
|
repo.save(movie1)
|
|
repo.save(movie2)
|
|
|
|
assert len(memory.ltm.library["movies"]) == 1
|
|
assert memory.ltm.library["movies"][0]["quality"] == "1080p"
|
|
|
|
def test_find_by_imdb_id(self, memory_with_library):
|
|
"""Should find movie by IMDb ID."""
|
|
repo = JsonMovieRepository()
|
|
|
|
movie = repo.find_by_imdb_id(ImdbId("tt1375666"))
|
|
|
|
assert movie is not None
|
|
assert movie.title.value == "Inception"
|
|
|
|
def test_find_by_imdb_id_not_found(self, memory):
|
|
"""Should return None if not found."""
|
|
repo = JsonMovieRepository()
|
|
|
|
movie = repo.find_by_imdb_id(ImdbId("tt9999999"))
|
|
|
|
assert movie is None
|
|
|
|
def test_find_all(self, memory_with_library):
|
|
"""Should return all movies."""
|
|
repo = JsonMovieRepository()
|
|
|
|
movies = repo.find_all()
|
|
|
|
assert len(movies) >= 2
|
|
titles = [m.title.value for m in movies]
|
|
assert "Inception" in titles
|
|
assert "Interstellar" in titles
|
|
|
|
def test_find_all_empty(self, memory):
|
|
"""Should return empty list if no movies."""
|
|
repo = JsonMovieRepository()
|
|
|
|
movies = repo.find_all()
|
|
|
|
assert movies == []
|
|
|
|
def test_delete(self, memory_with_library):
|
|
"""Should delete movie."""
|
|
repo = JsonMovieRepository()
|
|
|
|
result = repo.delete(ImdbId("tt1375666"))
|
|
|
|
assert result is True
|
|
assert len(memory_with_library.ltm.library["movies"]) == 1
|
|
|
|
def test_delete_not_found(self, memory):
|
|
"""Should return False if not found."""
|
|
repo = JsonMovieRepository()
|
|
|
|
result = repo.delete(ImdbId("tt9999999"))
|
|
|
|
assert result is False
|
|
|
|
def test_exists(self, memory_with_library):
|
|
"""Should check if movie exists."""
|
|
repo = JsonMovieRepository()
|
|
|
|
assert repo.exists(ImdbId("tt1375666")) is True
|
|
assert repo.exists(ImdbId("tt9999999")) is False
|
|
|
|
def test_preserves_all_fields(self, memory):
|
|
"""Should preserve all movie fields."""
|
|
repo = JsonMovieRepository()
|
|
movie = Movie(
|
|
imdb_id=ImdbId("tt1375666"),
|
|
title=MovieTitle("Inception"),
|
|
release_year=ReleaseYear(2010),
|
|
quality=Quality.FULL_HD,
|
|
file_path=FilePath("/movies/inception.mkv"),
|
|
file_size=FileSize(2500000000),
|
|
tmdb_id=27205,
|
|
)
|
|
|
|
repo.save(movie)
|
|
loaded = repo.find_by_imdb_id(ImdbId("tt1375666"))
|
|
|
|
assert loaded.title.value == "Inception"
|
|
assert loaded.release_year.value == 2010
|
|
assert loaded.quality.value == "1080p"
|
|
assert str(loaded.file_path) == "/movies/inception.mkv"
|
|
assert loaded.file_size.bytes == 2500000000
|
|
assert loaded.tmdb_id == 27205
|
|
|
|
|
|
class TestJsonTVShowRepository:
|
|
"""Tests for JsonTVShowRepository."""
|
|
|
|
def test_save_show(self, memory):
|
|
"""Should save a TV show."""
|
|
repo = JsonTVShowRepository()
|
|
show = TVShow(
|
|
imdb_id=ImdbId("tt0944947"),
|
|
title="Game of Thrones",
|
|
seasons_count=8,
|
|
status=ShowStatus.ENDED,
|
|
)
|
|
|
|
repo.save(show)
|
|
|
|
assert len(memory.ltm.library["tv_shows"]) == 1
|
|
assert memory.ltm.library["tv_shows"][0]["title"] == "Game of Thrones"
|
|
|
|
def test_save_updates_existing(self, memory):
|
|
"""Should update existing show."""
|
|
repo = JsonTVShowRepository()
|
|
show1 = TVShow(
|
|
imdb_id=ImdbId("tt0944947"),
|
|
title="Game of Thrones",
|
|
seasons_count=7,
|
|
status=ShowStatus.ONGOING,
|
|
)
|
|
show2 = TVShow(
|
|
imdb_id=ImdbId("tt0944947"),
|
|
title="Game of Thrones",
|
|
seasons_count=8,
|
|
status=ShowStatus.ENDED,
|
|
)
|
|
|
|
repo.save(show1)
|
|
repo.save(show2)
|
|
|
|
assert len(memory.ltm.library["tv_shows"]) == 1
|
|
assert memory.ltm.library["tv_shows"][0]["seasons_count"] == 8
|
|
|
|
def test_find_by_imdb_id(self, memory_with_library):
|
|
"""Should find show by IMDb ID."""
|
|
repo = JsonTVShowRepository()
|
|
|
|
show = repo.find_by_imdb_id(ImdbId("tt0944947"))
|
|
|
|
assert show is not None
|
|
assert show.title == "Game of Thrones"
|
|
|
|
def test_find_by_imdb_id_not_found(self, memory):
|
|
"""Should return None if not found."""
|
|
repo = JsonTVShowRepository()
|
|
|
|
show = repo.find_by_imdb_id(ImdbId("tt9999999"))
|
|
|
|
assert show is None
|
|
|
|
def test_find_all(self, memory_with_library):
|
|
"""Should return all shows."""
|
|
repo = JsonTVShowRepository()
|
|
|
|
shows = repo.find_all()
|
|
|
|
assert len(shows) == 1
|
|
assert shows[0].title == "Game of Thrones"
|
|
|
|
def test_delete(self, memory_with_library):
|
|
"""Should delete show."""
|
|
repo = JsonTVShowRepository()
|
|
|
|
result = repo.delete(ImdbId("tt0944947"))
|
|
|
|
assert result is True
|
|
assert len(memory_with_library.ltm.library["tv_shows"]) == 0
|
|
|
|
def test_exists(self, memory_with_library):
|
|
"""Should check if show exists."""
|
|
repo = JsonTVShowRepository()
|
|
|
|
assert repo.exists(ImdbId("tt0944947")) is True
|
|
assert repo.exists(ImdbId("tt9999999")) is False
|
|
|
|
def test_preserves_status(self, memory):
|
|
"""Should preserve show status."""
|
|
repo = JsonTVShowRepository()
|
|
|
|
for i, status in enumerate(
|
|
[ShowStatus.ONGOING, ShowStatus.ENDED, ShowStatus.UNKNOWN]
|
|
):
|
|
show = TVShow(
|
|
imdb_id=ImdbId(f"tt{i+1000000:07d}"),
|
|
title=f"Show {status.value}",
|
|
seasons_count=1,
|
|
status=status,
|
|
)
|
|
repo.save(show)
|
|
loaded = repo.find_by_imdb_id(ImdbId(f"tt{i+1000000:07d}"))
|
|
assert loaded.status == status
|
|
|
|
|
|
class TestJsonSubtitleRepository:
|
|
"""Tests for JsonSubtitleRepository."""
|
|
|
|
def test_save_subtitle(self, memory):
|
|
"""Should save a subtitle."""
|
|
repo = JsonSubtitleRepository()
|
|
subtitle = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/inception.en.srt"),
|
|
)
|
|
|
|
repo.save(subtitle)
|
|
|
|
assert "subtitles" in memory.ltm.library
|
|
assert len(memory.ltm.library["subtitles"]) == 1
|
|
|
|
def test_save_multiple_for_same_media(self, memory):
|
|
"""Should allow multiple subtitles for same media."""
|
|
repo = JsonSubtitleRepository()
|
|
sub_en = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/inception.en.srt"),
|
|
)
|
|
sub_fr = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.FRENCH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/inception.fr.srt"),
|
|
)
|
|
|
|
repo.save(sub_en)
|
|
repo.save(sub_fr)
|
|
|
|
assert len(memory.ltm.library["subtitles"]) == 2
|
|
|
|
def test_find_by_media(self, memory):
|
|
"""Should find subtitles by media ID."""
|
|
repo = JsonSubtitleRepository()
|
|
subtitle = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/inception.en.srt"),
|
|
)
|
|
repo.save(subtitle)
|
|
|
|
results = repo.find_by_media(ImdbId("tt1375666"))
|
|
|
|
assert len(results) == 1
|
|
assert results[0].language == Language.ENGLISH
|
|
|
|
def test_find_by_media_with_language_filter(self, memory):
|
|
"""Should filter by language."""
|
|
repo = JsonSubtitleRepository()
|
|
repo.save(
|
|
Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/en.srt"),
|
|
)
|
|
)
|
|
repo.save(
|
|
Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.FRENCH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/fr.srt"),
|
|
)
|
|
)
|
|
|
|
results = repo.find_by_media(ImdbId("tt1375666"), language=Language.FRENCH)
|
|
|
|
assert len(results) == 1
|
|
assert results[0].language == Language.FRENCH
|
|
|
|
def test_find_by_media_with_episode_filter(self, memory):
|
|
"""Should filter by season/episode."""
|
|
repo = JsonSubtitleRepository()
|
|
repo.save(
|
|
Subtitle(
|
|
media_imdb_id=ImdbId("tt0944947"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/s01e01.srt"),
|
|
season_number=1,
|
|
episode_number=1,
|
|
)
|
|
)
|
|
repo.save(
|
|
Subtitle(
|
|
media_imdb_id=ImdbId("tt0944947"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/s01e02.srt"),
|
|
season_number=1,
|
|
episode_number=2,
|
|
)
|
|
)
|
|
|
|
results = repo.find_by_media(
|
|
ImdbId("tt0944947"),
|
|
season=1,
|
|
episode=1,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0].episode_number == 1
|
|
|
|
def test_find_by_media_not_found(self, memory):
|
|
"""Should return empty list if not found."""
|
|
repo = JsonSubtitleRepository()
|
|
|
|
results = repo.find_by_media(ImdbId("tt9999999"))
|
|
|
|
assert results == []
|
|
|
|
def test_delete(self, memory):
|
|
"""Should delete subtitle by file path."""
|
|
repo = JsonSubtitleRepository()
|
|
subtitle = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/inception.en.srt"),
|
|
)
|
|
repo.save(subtitle)
|
|
|
|
result = repo.delete(subtitle)
|
|
|
|
assert result is True
|
|
assert len(memory.ltm.library["subtitles"]) == 0
|
|
|
|
def test_delete_not_found(self, memory):
|
|
"""Should return False if not found."""
|
|
repo = JsonSubtitleRepository()
|
|
subtitle = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/nonexistent.srt"),
|
|
)
|
|
|
|
result = repo.delete(subtitle)
|
|
|
|
assert result is False
|
|
|
|
def test_preserves_all_fields(self, memory):
|
|
"""Should preserve all subtitle fields."""
|
|
repo = JsonSubtitleRepository()
|
|
subtitle = Subtitle(
|
|
media_imdb_id=ImdbId("tt1375666"),
|
|
language=Language.ENGLISH,
|
|
format=SubtitleFormat.SRT,
|
|
file_path=FilePath("/subs/inception.en.srt"),
|
|
season_number=1,
|
|
episode_number=5,
|
|
timing_offset=TimingOffset(500),
|
|
hearing_impaired=True,
|
|
forced=False,
|
|
source="OpenSubtitles",
|
|
uploader="user123",
|
|
download_count=1000,
|
|
rating=8.5,
|
|
)
|
|
|
|
repo.save(subtitle)
|
|
results = repo.find_by_media(ImdbId("tt1375666"))
|
|
|
|
assert len(results) == 1
|
|
loaded = results[0]
|
|
assert loaded.season_number == 1
|
|
assert loaded.episode_number == 5
|
|
assert loaded.timing_offset.milliseconds == 500
|
|
assert loaded.hearing_impaired is True
|
|
assert loaded.forced is False
|
|
assert loaded.source == "OpenSubtitles"
|
|
assert loaded.uploader == "user123"
|
|
assert loaded.download_count == 1000
|
|
assert loaded.rating == 8.5
|