"""Integration tests for ``rescan_show``. Uses the real filesystem (``tmp_path``), the real release knowledge base, and the real ``.alfred`` repository. Only the media prober is stubbed — ffprobe needs real bytes and a binary. """ from __future__ import annotations import pytest # Phase 3 (refactor/dot-alfred-v2): the v1 rescan_show + v1 # DotAlfredTVShowRepository stack is intentionally left broken # while the TVShow/Movie aggregates are slimmed to TMDB-only. # Phase 4 rewrites rescan on top of the v2 release repositories + # library index, then deletes this quarantine block alongside the # v1 code. pytest.skip( "v1 rescan + v1 dot_alfred — replaced in Phase 4", allow_module_level=True, ) from pathlib import Path from alfred.application.library import rescan_show from alfred.domain.shared.media import ( AudioTrack, MediaInfo, SubtitleTrack, VideoTrack, ) from alfred.domain.shared.value_objects import ImdbId 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 import ( DotAlfredTVShowRepository, ) from alfred.infrastructure.persistence.dot_alfred.repository import ( SIDECAR_FILENAME, ) _KB = YamlReleaseKnowledge() _SCANNER = PathlibFilesystemScanner() # --------------------------------------------------------------------------- # # 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, *, episodic_episodes: tuple[int, ...] = (1, 2, 3), include_pack_s2: bool = True, ) -> Path: """Build a fake Foundation show folder under ``root``. Layout (matches the Foundation fixture naming): root/Foundation/ Foundation.S01.1080p.x265-ELiTE/ ← EPISODIC Foundation.S01E01.1080p.x265-ELiTE.mkv Foundation.S01E02.1080p.x265-ELiTE.mkv Foundation.S01E03.1080p.x265-ELiTE.mkv Foundation.S02.1080p.x265-ELiTE/ ← PACK Foundation.S02.1080p.x265-ELiTE.mkv """ show_root = root / "Foundation" show_root.mkdir() s1 = show_root / "Foundation.S01.1080p.x265-ELiTE" s1.mkdir() for n in episodic_episodes: (s1 / f"Foundation.S01E{n:02d}.1080p.x265-ELiTE.mkv").write_bytes(b"") if include_pack_s2: s2 = show_root / "Foundation.S02.1080p.x265-ELiTE" s2.mkdir() (s2 / "Foundation.S02.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_show_with_episodic_and_pack_seasons(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredTVShowRepository(library) prober = _StubProber() show = rescan_show( show_root, imdb_id="tt0804484", tmdb_id=84958, repository=repo, scanner=_SCANNER, prober=prober, kb=_KB, ) assert show.imdb_id == ImdbId("tt0804484") assert show.tmdb_id == 84958 assert show.title == "Foundation" assert show.seasons_count == 2 s1 = show.get_season(SeasonNumber(1)) s2 = show.get_season(SeasonNumber(2)) assert s1 is not None and s2 is not None # S01 EPISODIC: three episodes, season-level tracks empty. assert s1.episode_count == 3 assert s1.audio_tracks == () assert s1.subtitle_tracks == () # S02 PACK: no episodes, season-level tracks populated. assert s2.episode_count == 0 assert len(s2.audio_tracks) == 1 assert s2.audio_tracks[0].language == "eng" assert len(s2.subtitle_tracks) == 1 def test_episode_paths_are_relative_to_show_root(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredTVShowRepository(library) show = rescan_show( show_root, imdb_id="tt0804484", repository=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) s1 = show.get_season(SeasonNumber(1)) assert s1 is not None for ep in s1.episodes: assert ep.file_path is not None path_str = str(ep.file_path) # Must NOT be absolute and must start with the season folder. assert not Path(path_str).is_absolute() assert path_str.startswith("Foundation.S01.1080p.x265-ELiTE/") def test_persists_sidecar_on_disk(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredTVShowRepository(library) rescan_show( show_root, imdb_id="tt0804484", repository=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert (show_root / SIDECAR_FILENAME).is_file() # Round-trip via the repo. recovered = repo.find_by_imdb_id(ImdbId("tt0804484")) assert recovered is not None assert recovered.seasons_count == 2 def test_probe_called_once_per_video(self, tmp_path): library, show_root = _library_with_show(tmp_path) repo = DotAlfredTVShowRepository(library) prober = _StubProber() rescan_show( show_root, imdb_id="tt0804484", repository=repo, scanner=_SCANNER, prober=prober, kb=_KB, ) # 3 episodes + 1 pack video = 4 probes. assert len(prober.calls) == 4 # --------------------------------------------------------------------------- # # Edge cases # # --------------------------------------------------------------------------- # class TestEdgeCases: def test_empty_show_root_yields_empty_show(self, tmp_path): library = tmp_path / "library" library.mkdir() show_root = library / "Empty" show_root.mkdir() repo = DotAlfredTVShowRepository(library) show = rescan_show( show_root, imdb_id="tt0000001", repository=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert show.seasons_count == 0 assert (show_root / SIDECAR_FILENAME).is_file() def test_season_folder_without_videos_is_skipped(self, tmp_path, caplog): library = tmp_path / "library" library.mkdir() show_root = library / "Foundation" show_root.mkdir() (show_root / "Foundation.S01.WEB").mkdir() # empty season folder repo = DotAlfredTVShowRepository(library) with caplog.at_level("WARNING"): show = rescan_show( show_root, imdb_id="tt0804484", repository=repo, scanner=_SCANNER, prober=_StubProber(), kb=_KB, ) assert show.seasons_count == 0 assert any("no video file" 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, episodic_episodes=(1,), include_pack_s2=False ) repo = DotAlfredTVShowRepository(library) # Prober returns None — inspect_release skips enrichment, tracks empty. show = rescan_show( show_root, imdb_id="tt0804484", repository=repo, scanner=_SCANNER, prober=_StubProber(info=None), kb=_KB, ) s1 = show.get_season(SeasonNumber(1)) assert s1 is not None assert s1.episode_count == 1 ep = s1.episodes[0] assert ep.audio_tracks == () assert ep.subtitle_tracks == ()