"""Integration tests for the v2 ``rescan_show``. Uses the real filesystem (``tmp_path``), the real release knowledge base, the real v2 ``.alfred`` series repository, and the real scanner. Only the media prober is stubbed — ffprobe needs real bytes and a binary. """ from __future__ import annotations from pathlib import Path from alfred.application.tv_shows import rescan_show from alfred.domain.releases.value_objects import ReleaseMode from alfred.domain.shared.media import ( AudioTrack, MediaInfo, SubtitleTrack, VideoTrack, ) from alfred.domain.shared.value_objects import ImdbId, TmdbId from alfred.domain.tv_shows.value_objects import SeasonNumber from alfred.infrastructure.filesystem.scanner import PathlibFilesystemScanner from alfred.infrastructure.knowledge.release_kb import YamlReleaseKnowledge from alfred.infrastructure.persistence.dot_alfred.v2.repository import ( SIDECAR_FILENAME, DotAlfredSeriesReleaseRepository, ) _KB = YamlReleaseKnowledge() _SCANNER = PathlibFilesystemScanner() _TMDB_ID = TmdbId(84958) _IMDB_ID = ImdbId("tt0804484") # --------------------------------------------------------------------------- # # Helpers # # --------------------------------------------------------------------------- # _MISSING = object() class _StubProber: """Return a canned MediaInfo for every probe call, regardless of path.""" def __init__(self, info=_MISSING) -> None: self._info: MediaInfo | None = ( _default_info() if info is _MISSING else info # type: ignore[assignment] ) self.calls: list[Path] = [] def probe(self, video: Path) -> MediaInfo | None: self.calls.append(video) return self._info def list_subtitle_streams(self, video: Path): # pragma: no cover - unused return [] def _default_info() -> MediaInfo: return MediaInfo( video_tracks=(VideoTrack(index=0, codec="hevc", width=1920, height=1080),), audio_tracks=( AudioTrack( index=0, codec="eac3", channels=6, channel_layout="5.1", language="eng", ), ), subtitle_tracks=( SubtitleTrack( index=0, codec="subrip", language="eng", is_default=False, is_forced=False, ), ), ) def _make_foundation_library( root: Path, *, pack_episodes: tuple[int, ...] = (1, 2, 3), episodic_episodes: tuple[int, ...] = (1, 2), ) -> Path: """Build a fake Foundation show folder under ``root``. Layout:: root/Foundation/ Foundation.S01.1080p.x265-ELiTE/ ← PACK Foundation.S01E01.1080p.x265-ELiTE.mkv (flat) Foundation.S01E02.1080p.x265-ELiTE.mkv Foundation.S01E03.1080p.x265-ELiTE.mkv Foundation.S02/ ← EPISODIC Foundation.S02E01.1080p.x265-ELiTE/ (subfolder) Foundation.S02E01.1080p.x265-ELiTE.mkv Foundation.S02E02.1080p.x265-ELiTE/ Foundation.S02E02.1080p.x265-ELiTE.mkv ``pack_episodes`` and ``episodic_episodes`` can each be set to empty tuples to omit the corresponding season. """ show_root = root / "Foundation" show_root.mkdir() if pack_episodes: s1 = show_root / "Foundation.S01.1080p.x265-ELiTE" s1.mkdir() for n in pack_episodes: (s1 / f"Foundation.S01E{n:02d}.1080p.x265-ELiTE.mkv").write_bytes(b"") if episodic_episodes: s2 = show_root / "Foundation.S02" s2.mkdir() for n in episodic_episodes: sub = s2 / f"Foundation.S02E{n:02d}.1080p.x265-ELiTE" sub.mkdir() (sub / f"Foundation.S02E{n:02d}.1080p.x265-ELiTE.mkv").write_bytes(b"") return show_root def _library_with_show(tmp_path: Path) -> tuple[Path, Path]: """Return (library_root, show_root) prepared for the repository.""" library = tmp_path / "library" library.mkdir() show_root = _make_foundation_library(library) return library, show_root # --------------------------------------------------------------------------- # # Happy path # # --------------------------------------------------------------------------- # class TestHappyPath: def test_builds_release_with_pack_and_episodic_seasons(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) prober = _StubProber() release = rescan_show( show_root, tmdb_id=_TMDB_ID, imdb_id=_IMDB_ID, series_repo=repo, scanner=_SCANNER, prober=prober, kb=_KB, ) assert release.tmdb_id == _TMDB_ID assert release.imdb_id == _IMDB_ID assert len(release.seasons) == 2 s1 = release.get_season(SeasonNumber(1)) s2 = release.get_season(SeasonNumber(2)) assert s1 is not None and s2 is not None # S01 PACK: three single-episode releases, each carrying its # own TrackProfile. assert s1.mode is ReleaseMode.PACK assert s1.episode_count() == 3 for ep in s1.episodes: assert ep.episodes.is_single() assert len(ep.tracks.audio_tracks) == 1 assert ep.tracks.audio_tracks[0].language == "eng" assert len(ep.tracks.subtitle_tracks) == 1 # S02 EPISODIC: two single-episode releases, each in its own # subfolder. assert s2.mode is ReleaseMode.EPISODIC assert s2.episode_count() == 2 def test_pack_episode_paths_relative_and_flat(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, imdb_id=_IMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) s1 = release.get_season(SeasonNumber(1)) assert s1 is not None for ep in s1.episodes: path_str = str(ep.file_path) assert not Path(path_str).is_absolute() # PACK files sit directly inside the season folder. assert path_str.startswith("Foundation.S01.1080p.x265-ELiTE/") assert path_str.count("/") == 1 def test_episodic_episode_paths_descend_into_subfolders(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, imdb_id=_IMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) s2 = release.get_season(SeasonNumber(2)) assert s2 is not None for ep in s2.episodes: path_str = str(ep.file_path) assert not Path(path_str).is_absolute() # EPISODIC: season/episode_subdir/video.mkv → 2 separators. assert path_str.startswith("Foundation.S02/") assert path_str.count("/") == 2 def test_season_folder_names_recorded(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, imdb_id=_IMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) s1 = release.get_season(SeasonNumber(1)) s2 = release.get_season(SeasonNumber(2)) assert s1 is not None and s2 is not None assert s1.folder == "Foundation.S01.1080p.x265-ELiTE" assert s2.folder == "Foundation.S02" def test_persists_sidecar_on_disk(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) rescan_show( show_root, tmdb_id=_TMDB_ID, imdb_id=_IMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert (show_root / SIDECAR_FILENAME).is_file() recovered = repo.find_by_tmdb_id(_TMDB_ID) assert recovered is not None assert len(recovered.seasons) == 2 def test_probe_called_once_per_video(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) prober = _StubProber() rescan_show( show_root, tmdb_id=_TMDB_ID, imdb_id=_IMDB_ID, series_repo=repo, scanner=_SCANNER, prober=prober, kb=_KB, ) # 3 PACK files + 2 EPISODIC files = 5 probes. assert len(prober.calls) == 5 def test_imdb_id_optional(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert release.imdb_id is None # --------------------------------------------------------------------------- # # Edge cases # # --------------------------------------------------------------------------- # class TestEdgeCases: def test_empty_show_root_yields_empty_release(self, tmp_path): library = tmp_path / "library" library.mkdir() show_root = library / "Empty" show_root.mkdir() repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert release.seasons == () assert (show_root / SIDECAR_FILENAME).is_file() def test_empty_season_folder_is_skipped(self, tmp_path): library = tmp_path / "library" library.mkdir() show_root = library / "Foundation" show_root.mkdir() (show_root / "Foundation.S01.WEB").mkdir() # empty repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert release.seasons == () def test_malformed_mixed_season_is_skipped(self, tmp_path, caplog): library = tmp_path / "library" library.mkdir() show_root = library / "Foundation" show_root.mkdir() season = show_root / "Foundation.S01.Mixed" season.mkdir() # Flat video + subfolder → malformed. (season / "Foundation.S01E01.mkv").write_bytes(b"") sub = season / "Foundation.S01E02-RG" sub.mkdir() (sub / "Foundation.S01E02-RG.mkv").write_bytes(b"") repo = DotAlfredSeriesReleaseRepository(library) with caplog.at_level("WARNING"): release = rescan_show( show_root, tmdb_id=_TMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert release.seasons == () assert any( "mixes flat videos and subfolders" in r.message for r in caplog.records ) def test_single_unnumbered_pack_file_is_skipped(self, tmp_path, caplog): # A season folder with a single video whose name has no Exx # marker: the season number is parsed but the episode is not, # so the file is skipped and (no other parseable files →) # the whole season is skipped. library = tmp_path / "library" library.mkdir() show_root = library / "Foundation" show_root.mkdir() season = show_root / "Foundation.S01.Complete" season.mkdir() (season / "Foundation.S01.Complete.1080p.mkv").write_bytes(b"") repo = DotAlfredSeriesReleaseRepository(library) with caplog.at_level("WARNING"): release = rescan_show( show_root, tmdb_id=_TMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert release.seasons == () assert any( "no parseable episodes" in r.message or "no episode number parsed" in r.message for r in caplog.records ) def test_prober_returning_none_still_produces_episodes(self, tmp_path): library = tmp_path / "library" library.mkdir() show_root = _make_foundation_library( library, pack_episodes=(1,), episodic_episodes=() ) repo = DotAlfredSeriesReleaseRepository(library) release = rescan_show( show_root, tmdb_id=_TMDB_ID, series_repo=repo, scanner=_SCANNER, prober=_StubProber(info=None), kb=_KB, ) s1 = release.get_season(SeasonNumber(1)) assert s1 is not None assert s1.episode_count() == 1 ep = s1.episodes[0] assert ep.tracks.audio_tracks == () assert ep.tracks.subtitle_tracks == ()