francwa/alfred
1
0
Fork 0
Code Issues 1 Pull Requests 8 Actions Packages Projects Releases Wiki Activity
Files
main
alfred/tests/application/test_resolve_destination.py
T
francwa 9f1ce94690 refactor(application): inject kb/prober into resolve_destination use cases 
Remove the module-level _KB / _PROBER singletons from
alfred/application/filesystem/resolve_destination.py. The four
resolve_{season,episode,movie,series}_destination use cases now take
kb: ReleaseKnowledge and prober: MediaProber as required arguments,
matching the shape of inspect_release.

The singletons now live at the agent-tools frontier
(alfred/agent/tools/filesystem.py), where the LLM-facing wrappers
instantiate YamlReleaseKnowledge / FfprobeMediaProber once and thread
them through. The wrappers' Python signatures are unchanged — the
inspect-based JSON-schema generator in agent/registry.py still sees the
same LLM-passable params.

analyze_release drops the dirty 'from ... import _KB' indirection.

Tests inject their own stubs by keyword (prober=_StubProber(...)) via
thin convenience wrappers, replacing the prior
monkeypatch.setattr(rd, '_PROBER', ...) pattern.

testing/debug_release.py: instantiate YamlReleaseKnowledge() /
FfprobeMediaProber() inline at the two call sites.

Suite: 1077 passed.
2026-05-21 07:46:13 +02:00

547 lines
20 KiB
Python
Raw Permalink Blame History

"""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"] == []
Reference in New Issue View Git Blame Copy Permalink