"""Tests for ``alfred.application.filesystem.resolve_destination``. Four use cases compute library paths from a release name + TMDB metadata: - ``resolve_season_destination`` — folder move (series + season). - ``resolve_episode_destination`` — file move (full library_file path). - ``resolve_movie_destination`` — file move (folder + library_file). - ``resolve_series_destination`` — folder move (whole multi-season pack). Coverage: - ``TestFindExistingTvshowFolders`` — empty root, prefix match (case + space → dot). - ``TestResolveSeriesFolderInternal`` — confirmed_folder, no existing, single match, ambiguous → _Clarification. - ``TestSeason`` — library_not_set, ok path, clarification path. - ``TestEpisode`` — library_not_set, ok path, filename includes episode_title, ext from source. - ``TestMovie`` — library_not_set, ok path, is_new_folder, sanitization. - ``TestSeries`` — library_not_set, ok path. - ``TestDTOToDict`` — each DTO's three states (ok / clarification / error). """ from __future__ import annotations import pytest from alfred.application.filesystem.resolve_destination import ( ResolvedEpisodeDestination, ResolvedMovieDestination, ResolvedSeasonDestination, ResolvedSeriesDestination, _Clarification, _find_existing_tvshow_folders, _resolve_series_folder, resolve_episode_destination as _resolve_episode_destination, resolve_movie_destination as _resolve_movie_destination, resolve_season_destination as _resolve_season_destination, resolve_series_destination as _resolve_series_destination, ) from alfred.infrastructure.knowledge.release_kb import YamlReleaseKnowledge from alfred.infrastructure.persistence import Memory, set_memory _KB = YamlReleaseKnowledge() class _NullProber: """Default prober stub — never returns probe data.""" def list_subtitle_streams(self, video): # pragma: no cover return [] def probe(self, video): return None _DEFAULT_PROBER = _NullProber() def resolve_season_destination(*args, prober=None, **kwargs): return _resolve_season_destination( *args, kb=_KB, prober=prober or _DEFAULT_PROBER, **kwargs ) def resolve_episode_destination(*args, prober=None, **kwargs): return _resolve_episode_destination( *args, kb=_KB, prober=prober or _DEFAULT_PROBER, **kwargs ) def resolve_movie_destination(*args, prober=None, **kwargs): return _resolve_movie_destination( *args, kb=_KB, prober=prober or _DEFAULT_PROBER, **kwargs ) def resolve_series_destination(*args, prober=None, **kwargs): return _resolve_series_destination( *args, kb=_KB, prober=prober or _DEFAULT_PROBER, **kwargs ) REL_EPISODE = "Oz.S01E01.1080p.WEBRip.x265-KONTRAST" REL_SEASON = "Oz.S03.1080p.WEBRip.x265-KONTRAST" REL_MOVIE = "Inception.2010.1080p.BluRay.x265-GROUP" REL_SERIES = "Oz.Complete.Series.1080p.WEBRip.x265-KONTRAST" # --------------------------------------------------------------------------- # # Helpers # # --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- # # _find_existing_tvshow_folders # # --------------------------------------------------------------------------- # class TestFindExistingTvshowFolders: def test_missing_root_returns_empty(self, tmp_path): assert _find_existing_tvshow_folders(tmp_path / "ghost", "Oz", 1997) == [] def test_no_match(self, tmp_path): (tmp_path / "OtherShow.1999").mkdir() assert _find_existing_tvshow_folders(tmp_path, "Oz", 1997) == [] def test_matches_prefix_case_insensitive_with_space_dot(self, tmp_path): (tmp_path / "Oz.1997.WEBRip-KONTRAST").mkdir() (tmp_path / "oz.1997.bluray-OTHER").mkdir() (tmp_path / "OtherShow.1999").mkdir() out = _find_existing_tvshow_folders(tmp_path, "Oz", 1997) assert out == ["Oz.1997.WEBRip-KONTRAST", "oz.1997.bluray-OTHER"] or set( out ) == { "Oz.1997.WEBRip-KONTRAST", "oz.1997.bluray-OTHER", } def test_files_ignored(self, tmp_path): (tmp_path / "Oz.1997.txt").write_text("not a folder") assert _find_existing_tvshow_folders(tmp_path, "Oz", 1997) == [] def test_space_in_title_becomes_dot(self, tmp_path): (tmp_path / "The.X.Files.1993.x265-KONTRAST").mkdir() assert _find_existing_tvshow_folders(tmp_path, "The X Files", 1993) == [ "The.X.Files.1993.x265-KONTRAST" ] # --------------------------------------------------------------------------- # # _resolve_series_folder # # --------------------------------------------------------------------------- # class TestResolveSeriesFolderInternal: def test_confirmed_folder_when_exists(self, tmp_path): (tmp_path / "Oz.1997.X-GRP").mkdir() out = _resolve_series_folder( tmp_path, "Oz", "Oz", 1997, "Oz.1997.WEBRip-KONTRAST", confirmed_folder="Oz.1997.X-GRP", ) assert out == ("Oz.1997.X-GRP", False) def test_confirmed_folder_when_new(self, tmp_path): out = _resolve_series_folder( tmp_path, "Oz", "Oz", 1997, "Oz.1997.WEBRip-KONTRAST", confirmed_folder="Oz.1997.New-X", ) assert out == ("Oz.1997.New-X", True) def test_no_existing_returns_computed_as_new(self, tmp_path): out = _resolve_series_folder( tmp_path, "Oz", "Oz", 1997, "Oz.1997.WEBRip-KONTRAST", None ) assert out == ("Oz.1997.WEBRip-KONTRAST", True) def test_single_existing_matching_computed_returns_existing(self, tmp_path): (tmp_path / "Oz.1997.WEBRip-KONTRAST").mkdir() out = _resolve_series_folder( tmp_path, "Oz", "Oz", 1997, "Oz.1997.WEBRip-KONTRAST", None ) assert out == ("Oz.1997.WEBRip-KONTRAST", False) def test_single_existing_different_name_returns_clarification(self, tmp_path): (tmp_path / "Oz.1997.BluRay-OTHER").mkdir() out = _resolve_series_folder( tmp_path, "Oz", "Oz", 1997, "Oz.1997.WEBRip-KONTRAST", None ) assert isinstance(out, _Clarification) assert "Oz" in out.question assert "Oz.1997.BluRay-OTHER" in out.options assert "Oz.1997.WEBRip-KONTRAST" in out.options def test_multiple_existing_returns_clarification(self, tmp_path): (tmp_path / "Oz.1997.A-GRP").mkdir() (tmp_path / "Oz.1997.B-GRP").mkdir() out = _resolve_series_folder(tmp_path, "Oz", "Oz", 1997, "Oz.1997.A-GRP", None) assert isinstance(out, _Clarification) # Computed already in existing → not duplicated. assert out.options.count("Oz.1997.A-GRP") == 1 # --------------------------------------------------------------------------- # # Season # # --------------------------------------------------------------------------- # @pytest.fixture def cfg_memory(tmp_path): """Memory with tv_show + movie roots inside tmp_path. Roots NOT auto-created.""" storage = tmp_path / "_mem" storage.mkdir() tv = tmp_path / "tv" mv = tmp_path / "mv" tv.mkdir() mv.mkdir() mem = Memory(storage_dir=str(storage)) set_memory(mem) mem.ltm.library_paths.set("tv_show", str(tv)) mem.ltm.library_paths.set("movie", str(mv)) mem.save() return mem, tv, mv @pytest.fixture def empty_memory(tmp_path): """Memory with no library_paths configured.""" storage = tmp_path / "_mem_empty" storage.mkdir() mem = Memory(storage_dir=str(storage)) set_memory(mem) return mem class TestSeason: def test_library_not_set(self, empty_memory): out = resolve_season_destination(REL_SEASON, "Oz", 1997) assert out.status == "error" assert out.error == "library_not_set" def test_ok_path_new_series(self, cfg_memory): _, tv, _ = cfg_memory out = resolve_season_destination(REL_SEASON, "Oz", 1997) assert out.status == "ok" assert out.is_new_series_folder is True assert out.series_folder_name.startswith("Oz.1997") assert out.season_folder_name.startswith("Oz.S03") assert out.series_folder == str(tv / out.series_folder_name) assert out.season_folder == str( tv / out.series_folder_name / out.season_folder_name ) def test_clarification_path(self, cfg_memory): _, tv, _ = cfg_memory (tv / "Oz.1997.BluRay-OTHER").mkdir() out = resolve_season_destination(REL_SEASON, "Oz", 1997) assert out.status == "needs_clarification" assert out.options assert any("Oz" in o for o in out.options) # --------------------------------------------------------------------------- # # Episode # # --------------------------------------------------------------------------- # class TestEpisode: def test_library_not_set(self, empty_memory): out = resolve_episode_destination(REL_EPISODE, "/in/x.mkv", "Oz", 1997) assert out.status == "error" assert out.error == "library_not_set" def test_ok_path_with_episode_title(self, cfg_memory): _, tv, _ = cfg_memory out = resolve_episode_destination( REL_EPISODE, "/dl/source.mkv", "Oz", 1997, tmdb_episode_title="The Routine" ) assert out.status == "ok" assert out.filename.endswith(".mkv") assert "S01E01" in out.filename assert "The.Routine" in out.filename # library_file is series/season/file assert out.library_file == str( tv / out.series_folder_name / out.season_folder_name / out.filename ) def test_ok_path_without_episode_title(self, cfg_memory): out = resolve_episode_destination(REL_EPISODE, "/dl/source.mkv", "Oz", 1997) assert out.status == "ok" # No '..' from blank ep title. assert ".." not in out.filename def test_extension_taken_from_source_file(self, cfg_memory): out = resolve_episode_destination(REL_EPISODE, "/dl/source.mp4", "Oz", 1997) assert out.filename.endswith(".mp4") def test_clarification_path(self, cfg_memory): _, tv, _ = cfg_memory (tv / "Oz.1997.BluRay-OTHER").mkdir() out = resolve_episode_destination(REL_EPISODE, "/dl/source.mkv", "Oz", 1997) assert out.status == "needs_clarification" def test_confirmed_folder_threaded_through(self, cfg_memory): _, tv, _ = cfg_memory (tv / "Oz.1997.BluRay-OTHER").mkdir() out = resolve_episode_destination( REL_EPISODE, "/dl/source.mkv", "Oz", 1997, confirmed_folder="Oz.1997.BluRay-OTHER", ) assert out.status == "ok" assert out.series_folder_name == "Oz.1997.BluRay-OTHER" assert out.is_new_series_folder is False # --------------------------------------------------------------------------- # # Movie # # --------------------------------------------------------------------------- # class TestMovie: def test_library_not_set(self, empty_memory): out = resolve_movie_destination(REL_MOVIE, "/dl/m.mkv", "Inception", 2010) assert out.status == "error" assert out.error == "library_not_set" def test_ok_path(self, cfg_memory): _, _, mv = cfg_memory out = resolve_movie_destination(REL_MOVIE, "/dl/m.mkv", "Inception", 2010) assert out.status == "ok" assert out.movie_folder_name.startswith("Inception.2010") assert out.filename.endswith(".mkv") assert out.movie_folder == str(mv / out.movie_folder_name) assert out.library_file == str(mv / out.movie_folder_name / out.filename) assert out.is_new_folder is True def test_is_new_folder_false_when_exists(self, cfg_memory): _, _, mv = cfg_memory out_first = resolve_movie_destination(REL_MOVIE, "/dl/m.mkv", "Inception", 2010) (mv / out_first.movie_folder_name).mkdir() out = resolve_movie_destination(REL_MOVIE, "/dl/m.mkv", "Inception", 2010) assert out.is_new_folder is False def test_title_sanitized(self, cfg_memory): # Title with forbidden chars should be stripped. out = resolve_movie_destination(REL_MOVIE, "/dl/m.mkv", "Foo:Bar", 2010) assert ":" not in out.movie_folder_name assert ":" not in out.filename # --------------------------------------------------------------------------- # # Series # # --------------------------------------------------------------------------- # class TestSeries: def test_library_not_set(self, empty_memory): out = resolve_series_destination(REL_SERIES, "Oz", 1997) assert out.status == "error" assert out.error == "library_not_set" def test_ok_path(self, cfg_memory): _, tv, _ = cfg_memory out = resolve_series_destination(REL_SERIES, "Oz", 1997) assert out.status == "ok" assert out.series_folder_name.startswith("Oz.1997") assert out.series_folder == str(tv / out.series_folder_name) assert out.is_new_series_folder is True def test_clarification_path(self, cfg_memory): _, tv, _ = cfg_memory (tv / "Oz.1997.X-GRP").mkdir() out = resolve_series_destination(REL_SERIES, "Oz", 1997) assert out.status == "needs_clarification" # --------------------------------------------------------------------------- # # Probe enrichment wiring # # --------------------------------------------------------------------------- # class _StubProber: """Minimal MediaProber stub used to drive enrich_from_probe.""" def __init__(self, info): self._info = info def list_subtitle_streams(self, video): # pragma: no cover - unused here return [] def probe(self, video): return self._info def _stereo_movie_info(): """A MediaInfo that fills quality+codec when the release name omits them.""" from alfred.domain.shared.media import AudioTrack, MediaInfo, VideoTrack return MediaInfo( video_tracks=(VideoTrack(index=0, codec="hevc", width=1920, height=1080),), audio_tracks=( AudioTrack( index=1, codec="aac", channels=2, channel_layout="stereo", language="eng", is_default=True, ), ), subtitle_tracks=(), ) class TestProbeEnrichmentWiring: """When source_path/source_file points to a real file, the resolver should pick up ffprobe data via inspect_release and let the enriched tech_string land in the destination name.""" def test_movie_picks_up_probe_quality(self, cfg_memory, tmp_path): # Release name parses to "movie" but is missing the quality token; # probe must supply 1080p and refresh tech_string. bare_name = "Inception.2010.BluRay.x264-GROUP" video = tmp_path / "movie.mkv" video.write_bytes(b"") out = resolve_movie_destination( bare_name, str(video), "Inception", 2010, prober=_StubProber(_stereo_movie_info()), ) assert out.status == "ok" # tech_string -> "1080p.BluRay.x264" -> "1080p" shows up in names. assert "1080p" in out.movie_folder_name assert "1080p" in out.filename def test_movie_skips_probe_when_path_missing(self, cfg_memory): # If the file doesn't exist, no probe runs (the stub would have # injected 1080p — its absence proves the skip). out = resolve_movie_destination( "Inception.2010.BluRay.x264-GROUP", "/nowhere/m.mkv", "Inception", 2010, prober=_StubProber(_stereo_movie_info()), ) assert out.status == "ok" assert "1080p" not in out.movie_folder_name def test_season_picks_up_probe_via_source_path(self, cfg_memory, tmp_path): # Season pack name missing quality token; probe must add it. bare_name = "Oz.S03.BluRay.x265-KONTRAST" release_dir = tmp_path / bare_name release_dir.mkdir() (release_dir / "episode.mkv").write_bytes(b"") out = resolve_season_destination( bare_name, "Oz", 1997, source_path=str(release_dir), prober=_StubProber(_stereo_movie_info()), ) assert out.status == "ok" # Series folder name embeds tech_string -> "1080p" surfaced by probe. assert "1080p" in out.series_folder_name # --------------------------------------------------------------------------- # # DTO to_dict() # # --------------------------------------------------------------------------- # class TestDTOToDict: def test_season_ok(self): d = ResolvedSeasonDestination( status="ok", series_folder="/tv/S", season_folder="/tv/S/Season", series_folder_name="S", season_folder_name="Season", is_new_series_folder=True, ).to_dict() assert d["status"] == "ok" assert d["series_folder"] == "/tv/S" assert d["season_folder"] == "/tv/S/Season" assert d["is_new_series_folder"] is True def test_season_error(self): d = ResolvedSeasonDestination( status="error", error="library_not_set", message="missing" ).to_dict() assert d == { "status": "error", "error": "library_not_set", "message": "missing", } def test_season_clarification(self): d = ResolvedSeasonDestination( status="needs_clarification", question="which?", options=["a", "b"] ).to_dict() assert d == { "status": "needs_clarification", "question": "which?", "options": ["a", "b"], } def test_episode_ok(self): d = ResolvedEpisodeDestination( status="ok", series_folder="/tv/S", season_folder="/tv/S/Season", library_file="/tv/S/Season/X.mkv", series_folder_name="S", season_folder_name="Season", filename="X.mkv", is_new_series_folder=False, ).to_dict() assert d["library_file"] == "/tv/S/Season/X.mkv" assert d["filename"] == "X.mkv" def test_movie_ok(self): d = ResolvedMovieDestination( status="ok", movie_folder="/mv/X", library_file="/mv/X/X.mkv", movie_folder_name="X", filename="X.mkv", is_new_folder=True, ).to_dict() assert d["movie_folder"] == "/mv/X" assert d["library_file"] == "/mv/X/X.mkv" assert d["is_new_folder"] is True def test_series_ok(self): d = ResolvedSeriesDestination( status="ok", series_folder="/tv/S", series_folder_name="S", is_new_series_folder=False, ).to_dict() assert d == { "status": "ok", "series_folder": "/tv/S", "series_folder_name": "S", "is_new_series_folder": False, } def test_clarification_options_none_yields_empty_list(self): d = ResolvedSeasonDestination( status="needs_clarification", question="q", options=None ).to_dict() assert d["options"] == []