feat(library): add rescan_show orchestrator + walker (Step 4)
Step 4 of specs/dot_alfred.md — rebuild a TVShow aggregate from disk
by reusing the existing release pipeline (inspect_release) on every
video file in a show folder, then persist via the .alfred repository.
- alfred/application/library/walker.py — pure structural walk
(season folders detected via \bS\d{1,2}\b regex, video files
filtered against kb.video_extensions, no recursion).
- alfred/application/library/rescan.py — orchestrator that ingests
each season folder, infers PACK vs EPISODIC from on-disk file
count + parser output, and assembles via TVShowBuilder. Episode
paths stored relative to show_root. Logs + skips corrupt input
(no season parsed, mixed season numbers, unparseable episodes).
- Season now inherits MediaWithTracks: PACK seasons carry
season-level audio_tracks / subtitle_tracks; EPISODIC seasons
leave them empty (tracks live per-episode). SeasonBuilder gains
set_audio_tracks / set_subtitle_tracks; bridge writes/reads them
in the PACK branch via shared _synth_* helpers.
Out of scope, tracked as tech debt: adjacent .srt capture, multi-
episode (episode_end), TMDB-driven PACK detection (the current
heuristic '1 file == PACK' is a placeholder until ShowTracker lands).
18 new tests (11 walker + 7 rescan integration) on tmp_path with
the Foundation layout. Full suite: 1149 passed.
This commit is contained in:
@@ -0,0 +1,277 @@
|
||||
"""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
|
||||
|
||||
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 == ()
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Tests for the show-tree walker."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from alfred.application.library.walker import walk_show
|
||||
from alfred.infrastructure.filesystem.scanner import PathlibFilesystemScanner
|
||||
from alfred.infrastructure.knowledge.release_kb import YamlReleaseKnowledge
|
||||
|
||||
_KB = YamlReleaseKnowledge()
|
||||
_SCANNER = PathlibFilesystemScanner()
|
||||
|
||||
|
||||
def _episode(dir_path, name: str) -> None:
|
||||
(dir_path / name).write_bytes(b"")
|
||||
|
||||
|
||||
class TestSeasonFolderDetection:
|
||||
def test_keeps_folders_with_season_token(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
(show / "Foundation.S01.1080p.WEB-DL.x265-GROUP").mkdir()
|
||||
(show / "Foundation.S02.Complete.1080p.WEB-DL-GROUP").mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
names = sorted(s.season_dir.name for s in tree.season_folders)
|
||||
assert names == [
|
||||
"Foundation.S01.1080p.WEB-DL.x265-GROUP",
|
||||
"Foundation.S02.Complete.1080p.WEB-DL-GROUP",
|
||||
]
|
||||
|
||||
def test_skips_folders_without_season_token(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
(show / "Sample").mkdir()
|
||||
(show / "Soundtrack").mkdir()
|
||||
(show / "Foundation.S01.WEB").mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
assert [s.season_dir.name for s in tree.season_folders] == [
|
||||
"Foundation.S01.WEB"
|
||||
]
|
||||
|
||||
def test_season_token_is_case_insensitive(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
(show / "Foundation.s01.WEB").mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
assert len(tree.season_folders) == 1
|
||||
|
||||
def test_season_token_rejects_in_word(self, tmp_path):
|
||||
# "Season01" without separator must not match (we want a real
|
||||
# Sxx token, not any letter S followed by digits).
|
||||
show = tmp_path / "WeirdShow"
|
||||
show.mkdir()
|
||||
(show / "Pass100").mkdir()
|
||||
(show / "ABS01XYZ").mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
assert tree.season_folders == ()
|
||||
|
||||
def test_ignores_files_at_show_root(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
(show / "Foundation.S01.mkv").write_bytes(b"") # file, not a folder
|
||||
(show / "Foundation.S01.WEB").mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
assert [s.season_dir.name for s in tree.season_folders] == [
|
||||
"Foundation.S01.WEB"
|
||||
]
|
||||
|
||||
|
||||
class TestVideoFileCollection:
|
||||
def test_keeps_only_video_extensions(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
season = show / "Foundation.S01.WEB"
|
||||
season.mkdir()
|
||||
_episode(season, "Foundation.S01E01.mkv")
|
||||
_episode(season, "Foundation.S01E02.mp4")
|
||||
_episode(season, "Foundation.S01E01.eng.srt") # not a video
|
||||
_episode(season, "info.nfo") # not a video
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
(folder,) = tree.season_folders
|
||||
suffixes = sorted(p.suffix for p in folder.video_files)
|
||||
assert suffixes == [".mkv", ".mp4"]
|
||||
|
||||
def test_empty_season_folder_surfaces(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
(show / "Foundation.S01.WEB").mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
(folder,) = tree.season_folders
|
||||
assert folder.video_files == ()
|
||||
|
||||
def test_single_pack_video_is_kept(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
season = show / "Foundation.S01.Complete.WEB"
|
||||
season.mkdir()
|
||||
_episode(season, "Foundation.S01.Complete.WEB.mkv")
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
(folder,) = tree.season_folders
|
||||
assert [p.name for p in folder.video_files] == [
|
||||
"Foundation.S01.Complete.WEB.mkv"
|
||||
]
|
||||
|
||||
def test_does_not_recurse_into_sub_subfolders(self, tmp_path):
|
||||
show = tmp_path / "Foundation"
|
||||
show.mkdir()
|
||||
season = show / "Foundation.S01.WEB"
|
||||
season.mkdir()
|
||||
nested = season / "Extras"
|
||||
nested.mkdir()
|
||||
_episode(nested, "Foundation.S01E01.mkv")
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
(folder,) = tree.season_folders
|
||||
assert folder.video_files == ()
|
||||
|
||||
|
||||
class TestEdgeCases:
|
||||
def test_empty_show_root(self, tmp_path):
|
||||
show = tmp_path / "Empty"
|
||||
show.mkdir()
|
||||
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
|
||||
assert tree.show_root == show
|
||||
assert tree.season_folders == ()
|
||||
|
||||
def test_missing_show_root_returns_empty_tree(self, tmp_path):
|
||||
# Scanner returns [] for non-existent paths; walker must not raise.
|
||||
tree = walk_show(tmp_path / "Nope", scanner=_SCANNER, kb=_KB)
|
||||
assert tree.season_folders == ()
|
||||
Reference in New Issue
Block a user