From 86222d95d1f2cdc8463ce1ab7e4f93594a728560 Mon Sep 17 00:00:00 2001 From: Francwa Date: Mon, 25 May 2026 21:10:32 +0200 Subject: [PATCH] =?UTF-8?q?refactor(persistence):=20Phase=204=20Step=203?= =?UTF-8?q?=20=E2=80=94=20delete=20v1=20dot=5Falfred=20+=20ports?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Now that rescan_show + rescan_movie run on the v2 release repositories (Phase 4 Steps 1-2), the v1 dot_alfred stack and its abstract domain ports have zero callers. Delete them and lift the Phase 3 quarantines. Deleted * alfred/infrastructure/persistence/dot_alfred/bridge.py * alfred/infrastructure/persistence/dot_alfred/repository.py (v1) * alfred/infrastructure/persistence/dot_alfred/serializer.py (v1) * alfred/infrastructure/persistence/dot_alfred/sidecar.py (v1) * alfred/domain/tv_shows/repositories.py (TVShowRepository ABC) * alfred/domain/movies/repositories.py (MovieRepository ABC) * tests/infrastructure/persistence/dot_alfred/test_repository.py * tests/infrastructure/persistence/dot_alfred/test_serializer.py Rewrite alfred/infrastructure/persistence/dot_alfred/__init__.py now re- exports only the v2 surface: the four concrete repositories (DotAlfredSeriesReleaseRepository, DotAlfredMovieReleaseRepository, DotAlfredTVShowLibraryIndex, DotAlfredMovieLibraryIndex) plus ShowFolderUnknown. DTO-level imports go through alfred.infrastructure.persistence.dot_alfred.v2 directly. No backwards-compat shims (per CLAUDE.md): the v1 names are gone, not aliased. Test suite drops from 10 → 8 skips (the two Phase 3 module-level skips disappear with the quarantined files). Full suite: 1233 passed / 8 skipped / 4 xfailed. The MediaWithTracks mixin in alfred.domain.shared.media is now orphaned (Episode lost its tracks in Phase 3, MovieRelease doesn't inherit it). Parked for Phase 5, which will either mount it on MovieRelease / SeasonRelease or delete it for good. --- alfred/domain/movies/repositories.py | 73 --- alfred/domain/tv_shows/repositories.py | 40 -- .../persistence/dot_alfred/__init__.py | 52 +-- .../persistence/dot_alfred/bridge.py | 184 -------- .../persistence/dot_alfred/repository.py | 198 -------- .../persistence/dot_alfred/serializer.py | 294 ------------ .../persistence/dot_alfred/sidecar.py | 87 ---- .../persistence/dot_alfred/test_repository.py | 316 ------------- .../persistence/dot_alfred/test_serializer.py | 433 ------------------ 9 files changed, 20 insertions(+), 1657 deletions(-) delete mode 100644 alfred/domain/movies/repositories.py delete mode 100644 alfred/domain/tv_shows/repositories.py delete mode 100644 alfred/infrastructure/persistence/dot_alfred/bridge.py delete mode 100644 alfred/infrastructure/persistence/dot_alfred/repository.py delete mode 100644 alfred/infrastructure/persistence/dot_alfred/serializer.py delete mode 100644 alfred/infrastructure/persistence/dot_alfred/sidecar.py delete mode 100644 tests/infrastructure/persistence/dot_alfred/test_repository.py delete mode 100644 tests/infrastructure/persistence/dot_alfred/test_serializer.py diff --git a/alfred/domain/movies/repositories.py b/alfred/domain/movies/repositories.py deleted file mode 100644 index 601c126..0000000 --- a/alfred/domain/movies/repositories.py +++ /dev/null @@ -1,73 +0,0 @@ -"""Movie repository interfaces (abstract).""" - -from abc import ABC, abstractmethod - -from ..shared.value_objects import ImdbId -from .entities import Movie - - -class MovieRepository(ABC): - """ - Abstract repository for movie persistence. - - This defines the interface that infrastructure implementations must follow. - """ - - @abstractmethod - def save(self, movie: Movie) -> None: - """ - Save a movie to the repository. - - Args: - movie: Movie entity to save - """ - pass - - @abstractmethod - def find_by_imdb_id(self, imdb_id: ImdbId) -> Movie | None: - """ - Find a movie by its IMDb ID. - - Args: - imdb_id: IMDb ID to search for - - Returns: - Movie if found, None otherwise - """ - pass - - @abstractmethod - def find_all(self) -> list[Movie]: - """ - Get all movies in the repository. - - Returns: - List of all movies - """ - pass - - @abstractmethod - def delete(self, imdb_id: ImdbId) -> bool: - """ - Delete a movie from the repository. - - Args: - imdb_id: IMDb ID of the movie to delete - - Returns: - True if deleted, False if not found - """ - pass - - @abstractmethod - def exists(self, imdb_id: ImdbId) -> bool: - """ - Check if a movie exists in the repository. - - Args: - imdb_id: IMDb ID to check - - Returns: - True if exists, False otherwise - """ - pass diff --git a/alfred/domain/tv_shows/repositories.py b/alfred/domain/tv_shows/repositories.py deleted file mode 100644 index f6fd954..0000000 --- a/alfred/domain/tv_shows/repositories.py +++ /dev/null @@ -1,40 +0,0 @@ -"""TV Show repository interface. - -A single repository for the aggregate root only — Season and Episode are -**inside** the TVShow aggregate and are never persisted independently. The -aggregate is always loaded and saved as a whole. -""" - -from abc import ABC, abstractmethod - -from ..shared.value_objects import ImdbId -from .entities import TVShow - - -class TVShowRepository(ABC): - """ - Abstract repository for the TVShow aggregate. - - Implementations are responsible for persisting the full aggregate graph - (TVShow + all its Seasons + all their Episodes) atomically. - """ - - @abstractmethod - def save(self, show: TVShow) -> None: - """Persist the full TVShow aggregate.""" - - @abstractmethod - def find_by_imdb_id(self, imdb_id: ImdbId) -> TVShow | None: - """Load the full TVShow aggregate by IMDb ID, or None if absent.""" - - @abstractmethod - def find_all(self) -> list[TVShow]: - """Load all TVShow aggregates.""" - - @abstractmethod - def delete(self, imdb_id: ImdbId) -> bool: - """Remove the aggregate. Returns True if it existed and was deleted.""" - - @abstractmethod - def exists(self, imdb_id: ImdbId) -> bool: - """True if the aggregate exists in the store.""" diff --git a/alfred/infrastructure/persistence/dot_alfred/__init__.py b/alfred/infrastructure/persistence/dot_alfred/__init__.py index c93ba4d..08b8f7c 100644 --- a/alfred/infrastructure/persistence/dot_alfred/__init__.py +++ b/alfred/infrastructure/persistence/dot_alfred/__init__.py @@ -1,42 +1,30 @@ -"""`.alfred` sidecar persistence layer. +"""`.alfred` sidecar persistence (v2-only surface). -Implements the per-show YAML sidecar described in -``specs/dot_alfred.md``. The sidecar is a single file named ``.alfred`` -posed at the root of a show's directory, containing the full aggregate -in a factual-only schema. +v1 was deleted in Phase 4 of the v2 migration (see +``.claude/specs/dot_alfred_v2_phase4.md``). This package now exposes +only the v2 stack: -Public surface: +* Per-item release sidecars (one ``.alfred`` per show / per movie). +* Library-root indexes (one ``.alfred.index`` per library root). -* :mod:`.sidecar` — DTOs (``ShowSidecar``, ``SeasonSidecar``, - ``EpisodeSidecar``, ``SubtitleEntry``) that mirror the YAML schema. -* :mod:`.serializer` — ``serialize`` / ``deserialize`` functions - converting between DTOs and plain dicts (YAML-ready). -* :mod:`.bridge` — ``to_sidecar`` / ``from_sidecar`` translating - between the domain :class:`TVShow` aggregate and the sidecar DTOs. -* :mod:`.repository` — :class:`DotAlfredTVShowRepository`, the - concrete filesystem-backed implementation of the abstract - :class:`TVShowRepository` port. +For schema-level DTOs (``SeriesReleaseSidecar`` etc.) import directly +from :mod:`alfred.infrastructure.persistence.dot_alfred.v2`. The +top-level package re-exports the four concrete repositories that +application orchestrators talk to. """ -from .bridge import from_sidecar, to_sidecar -from .repository import DotAlfredTVShowRepository, ShowFolderUnknown -from .serializer import deserialize, serialize -from .sidecar import ( - EpisodeSidecar, - SeasonSidecar, - ShowSidecar, - SubtitleEntry, +from .v2.repository import ( + DotAlfredMovieLibraryIndex, + DotAlfredMovieReleaseRepository, + DotAlfredSeriesReleaseRepository, + DotAlfredTVShowLibraryIndex, + ShowFolderUnknown, ) __all__ = [ - "deserialize", - "serialize", - "from_sidecar", - "to_sidecar", - "DotAlfredTVShowRepository", + "DotAlfredMovieLibraryIndex", + "DotAlfredMovieReleaseRepository", + "DotAlfredSeriesReleaseRepository", + "DotAlfredTVShowLibraryIndex", "ShowFolderUnknown", - "EpisodeSidecar", - "SeasonSidecar", - "ShowSidecar", - "SubtitleEntry", ] diff --git a/alfred/infrastructure/persistence/dot_alfred/bridge.py b/alfred/infrastructure/persistence/dot_alfred/bridge.py deleted file mode 100644 index 6a23d6c..0000000 --- a/alfred/infrastructure/persistence/dot_alfred/bridge.py +++ /dev/null @@ -1,184 +0,0 @@ -"""Bridge between the ``.alfred`` sidecar DTOs and the TVShow aggregate. - -The sidecar stores a **summary** of the probe (audio languages, -subtitle entries with source + type) — not the full ffprobe output. -Going back to the domain we synthesize ``AudioTrack`` and -``SubtitleTrack`` objects with only the fields the sidecar preserved: - -* ``AudioTrack`` — one per language, ``codec`` / ``channels`` / - ``channel_layout`` / ``is_default`` left as ``None`` / ``False``. -* ``SubtitleTrack`` — one per :class:`SubtitleEntry`, ``codec`` / - ``is_default`` left as ``None`` / ``False``; ``is_forced`` derived - from ``entry.type == "forced"``. - -The reverse path (TVShow → sidecar) summarizes the rich tracks back to -the sidecar shape: unique audio languages preserving track order, -subtitle entries built from the track flags (``is_forced`` → -``type="forced"``, otherwise ``type="standard"``; ``source="embedded"`` -since the domain track represents an in-container stream). - -External subtitles (``source="adjacent"``) are not currently surfaced -by the domain track model — they exist on disk next to the file but -the aggregate has no slot for them today. They will be lossless once -the subtitle scan layer feeds them in; for now the bridge round-trips -the embedded subs only. -""" - -from __future__ import annotations - -from ....domain.shared.media import AudioTrack, SubtitleTrack -from ....domain.shared.value_objects import FilePath -from ....domain.tv_shows.builders import SeasonBuilder, TVShowBuilder -from ....domain.tv_shows.entities import Episode, Season, TVShow -from ....domain.tv_shows.value_objects import SeasonNumber -from .sidecar import ( - EpisodeSidecar, - SeasonSidecar, - ShowSidecar, - SubtitleEntry, -) - -# ════════════════════════════════════════════════════════════════════════════ -# TVShow → ShowSidecar -# ════════════════════════════════════════════════════════════════════════════ - - -def to_sidecar(show: TVShow, *, folder_paths: dict[int, str]) -> ShowSidecar: - """Build a :class:`ShowSidecar` from a domain :class:`TVShow`. - - ``folder_paths`` maps season numbers to the on-disk folder name - (relative to the show root). Required because the domain does not - carry the source folder name; the caller (repository) knows it. - """ - seasons = tuple( - _season_to_sidecar(s, folder_paths[s.season_number.value]) - for s in show.seasons - ) - return ShowSidecar( - imdb_id=show.imdb_id, - tmdb_id=show.tmdb_id, - seasons=seasons, - ) - - -def _season_to_sidecar(season: Season, path: str) -> SeasonSidecar: - if season.episodes: - # EPISODIC mode — tracks live on each episode. - return SeasonSidecar( - number=season.season_number, - path=path, - episodes=tuple(_episode_to_sidecar(ep) for ep in season.episodes), - ) - - # PACK mode — season-scoped tracks (single release covering the whole - # season). Summarize the same way as for episodes. - return SeasonSidecar( - number=season.season_number, - path=path, - audio_languages=tuple(season.audio_languages()), - subtitles=tuple( - _subtitle_track_to_entry(t) for t in season.subtitle_tracks - ), - ) - - -def _episode_to_sidecar(episode: Episode) -> EpisodeSidecar: - if episode.file_path is None: - raise ValueError( - f"cannot serialize episode {episode!r} without a file_path" - ) - audio_languages = tuple(episode.audio_languages()) - subtitles = tuple(_subtitle_track_to_entry(t) for t in episode.subtitle_tracks) - return EpisodeSidecar( - number=episode.episode_number, - path=str(episode.file_path), - audio_languages=audio_languages, - subtitles=subtitles, - ) - - -def _subtitle_track_to_entry(track: SubtitleTrack) -> SubtitleEntry: - return SubtitleEntry( - language=track.language or "und", - source="embedded", - type="forced" if track.is_forced else "standard", - ) - - -# ════════════════════════════════════════════════════════════════════════════ -# ShowSidecar → TVShow -# ════════════════════════════════════════════════════════════════════════════ - - -def from_sidecar(sidecar: ShowSidecar, *, title: str) -> TVShow: - """Reconstruct a :class:`TVShow` from a sidecar. - - ``title`` must be supplied by the caller — the sidecar stores - identity (``imdb_id`` / ``tmdb_id``) but not the display title; the - repository derives it from the folder name on disk. - """ - builder = TVShowBuilder( - imdb_id=sidecar.imdb_id, - title=title, - tmdb_id=sidecar.tmdb_id, - ) - for season in sidecar.seasons: - builder.add_season(_season_from_sidecar(season)) - return builder.build() - - -def _season_from_sidecar(season: SeasonSidecar) -> Season: - sb = SeasonBuilder(season.number) - for ep in season.episodes: - sb.add_episode(_episode_from_sidecar(ep, season.number)) - if not season.episodes: - # PACK mode — populate season-scoped tracks from the sidecar. - sb.set_audio_tracks(_synth_audio_tracks(season.audio_languages)) - sb.set_subtitle_tracks(_synth_subtitle_tracks(season.subtitles)) - return sb.build() - - -def _synth_audio_tracks( - languages: tuple[str, ...], -) -> tuple[AudioTrack, ...]: - return tuple( - AudioTrack( - index=i, - codec=None, - channels=None, - channel_layout=None, - language=lang, - ) - for i, lang in enumerate(languages) - ) - - -def _synth_subtitle_tracks( - entries: tuple[SubtitleEntry, ...], -) -> tuple[SubtitleTrack, ...]: - return tuple( - SubtitleTrack( - index=i, - codec=None, - language=entry.language, - is_default=False, - is_forced=(entry.type == "forced"), - ) - for i, entry in enumerate(entries) - ) - - -def _episode_from_sidecar( - episode: EpisodeSidecar, season_number: SeasonNumber -) -> Episode: - return Episode( - season_number=season_number, - episode_number=episode.number, - title="", - file_path=FilePath(episode.path), - audio_tracks=_synth_audio_tracks(episode.audio_languages), - subtitle_tracks=_synth_subtitle_tracks(episode.subtitles), - ) - - -__all__ = ["from_sidecar", "to_sidecar"] diff --git a/alfred/infrastructure/persistence/dot_alfred/repository.py b/alfred/infrastructure/persistence/dot_alfred/repository.py deleted file mode 100644 index 6c2ba6a..0000000 --- a/alfred/infrastructure/persistence/dot_alfred/repository.py +++ /dev/null @@ -1,198 +0,0 @@ -"""Filesystem-backed implementation of :class:`TVShowRepository`. - -The repository keeps no in-memory cache of aggregates: every read goes -back to the filesystem. It does keep a tiny mapping ``imdb_id → -folder_name`` populated as folders are discovered, so subsequent saves -can find the right destination without re-walking ``library_root/``. - -Atomic writes: the YAML is dumped to ``.alfred.tmp`` and then renamed -to ``.alfred`` via ``os.replace`` — atomic on POSIX and NTFS. No half- -written file ever becomes visible. - -Cold scan: a show folder without a ``.alfred`` returns ``None`` from -``find_by_imdb_id`` and is skipped by ``find_all``. The opt-in -``rescan_show`` tool (step 4) will be responsible for rebuilding a -missing sidecar by walking the filesystem. - -The repository never invents a folder name. ``save(show)`` assumes the -target folder already exists (the upstream ``MediaOrganizer`` is in -charge of placing files); the repository writes the ``.alfred`` next -to them. -""" - -from __future__ import annotations - -import logging -import os -from pathlib import Path - -import yaml - -from ....domain.shared.value_objects import ImdbId -from ....domain.tv_shows.entities import TVShow -from ....domain.tv_shows.repositories import TVShowRepository -from .bridge import from_sidecar, to_sidecar -from .serializer import SidecarSchemaError, deserialize, serialize - -logger = logging.getLogger(__name__) - -SIDECAR_FILENAME = ".alfred" -SIDECAR_TMP_FILENAME = ".alfred.tmp" - - -class ShowFolderUnknown(LookupError): - """Raised by :meth:`DotAlfredTVShowRepository.save` when the folder - for the given show cannot be located. - - The repository never invents a folder name; the caller is expected - to have placed files there beforehand (typically via the - ``MediaOrganizer``). - """ - - -class DotAlfredTVShowRepository(TVShowRepository): - """A :class:`TVShowRepository` backed by per-show ``.alfred`` files. - - Args: - library_root: directory containing one folder per show. - """ - - def __init__(self, library_root: Path) -> None: - self._library_root = Path(library_root) - # Lazy cache: imdb_id → folder name (relative to library_root). - # Populated on every successful read or save; rebuilt on demand. - self._folder_index: dict[str, str] = {} - - # ── TVShowRepository surface ──────────────────────────────────────────── - - def save(self, show: TVShow) -> None: - folder_name = self._resolve_folder_name(show) - show_dir = self._library_root / folder_name - if not show_dir.is_dir(): - raise ShowFolderUnknown( - f"show folder does not exist on disk: {show_dir}" - ) - - folder_paths = { - s.season_number.value: s.get_folder_name() for s in show.seasons - } - sidecar = to_sidecar(show, folder_paths=folder_paths) - text = yaml.safe_dump(serialize(sidecar), sort_keys=False) - self._atomic_write(show_dir, text) - self._folder_index[str(show.imdb_id)] = folder_name - - def find_by_imdb_id(self, imdb_id: ImdbId) -> TVShow | None: - for folder_name, show in self._iter_library(): - if show.imdb_id == imdb_id: - self._folder_index[str(imdb_id)] = folder_name - return show - return None - - def find_all(self) -> list[TVShow]: - result: list[TVShow] = [] - for folder_name, show in self._iter_library(): - self._folder_index[str(show.imdb_id)] = folder_name - result.append(show) - return result - - def delete(self, imdb_id: ImdbId) -> bool: - folder_name = self._lookup_folder(imdb_id) - if folder_name is None: - return False - sidecar_path = self._library_root / folder_name / SIDECAR_FILENAME - if not sidecar_path.is_file(): - return False - sidecar_path.unlink() - self._folder_index.pop(str(imdb_id), None) - return True - - def exists(self, imdb_id: ImdbId) -> bool: - return self.find_by_imdb_id(imdb_id) is not None - - # ── Internals ─────────────────────────────────────────────────────────── - - def _iter_library(self): - """Yield ``(folder_name, TVShow)`` for every readable sidecar. - - Folders without a sidecar, or with an unreadable / invalid one, - are skipped (with a warning logged). The repository never - cold-scans here — that is the job of the upcoming - ``rescan_show`` tool. - """ - if not self._library_root.is_dir(): - return - for entry in sorted(self._library_root.iterdir()): - if not entry.is_dir(): - continue - sidecar_path = entry / SIDECAR_FILENAME - if not sidecar_path.is_file(): - continue - show = self._read_sidecar(entry, sidecar_path) - if show is not None: - yield entry.name, show - - def _read_sidecar(self, show_dir: Path, sidecar_path: Path) -> TVShow | None: - try: - raw = yaml.safe_load(sidecar_path.read_text()) - except (OSError, yaml.YAMLError) as exc: - logger.warning( - "skipping %s — sidecar unreadable: %s", sidecar_path, exc - ) - return None - try: - sidecar = deserialize(raw) - except SidecarSchemaError as exc: - logger.warning( - "skipping %s — invalid sidecar schema: %s", sidecar_path, exc - ) - return None - return from_sidecar(sidecar, title=show_dir.name) - - def _resolve_folder_name(self, show: TVShow) -> str: - """Return the folder name to write ``show``'s sidecar into. - - Order of resolution: - - 1. Cache hit on ``imdb_id``. - 2. Folder ``show.get_folder_name()`` exists on disk. - 3. Full ``find_all`` scan as a last resort to refresh the index. - """ - key = str(show.imdb_id) - cached = self._folder_index.get(key) - if cached is not None and (self._library_root / cached).is_dir(): - return cached - - guess = show.get_folder_name() - if (self._library_root / guess).is_dir(): - return guess - - # Last resort — refresh the index in case the folder was renamed. - for folder_name, found in self._iter_library(): - self._folder_index[str(found.imdb_id)] = folder_name - if found.imdb_id == show.imdb_id: - return folder_name - - raise ShowFolderUnknown( - f"no folder found for show {show.imdb_id} under {self._library_root}" - ) - - def _lookup_folder(self, imdb_id: ImdbId) -> str | None: - key = str(imdb_id) - cached = self._folder_index.get(key) - if cached is not None and (self._library_root / cached).is_dir(): - return cached - for folder_name, found in self._iter_library(): - self._folder_index[str(found.imdb_id)] = folder_name - if found.imdb_id == imdb_id: - return folder_name - return None - - @staticmethod - def _atomic_write(show_dir: Path, text: str) -> None: - tmp = show_dir / SIDECAR_TMP_FILENAME - final = show_dir / SIDECAR_FILENAME - tmp.write_text(text) - os.replace(tmp, final) - - -__all__ = ["DotAlfredTVShowRepository", "ShowFolderUnknown"] diff --git a/alfred/infrastructure/persistence/dot_alfred/serializer.py b/alfred/infrastructure/persistence/dot_alfred/serializer.py deleted file mode 100644 index 6f018cc..0000000 --- a/alfred/infrastructure/persistence/dot_alfred/serializer.py +++ /dev/null @@ -1,294 +0,0 @@ -"""Serialize / deserialize ``.alfred`` sidecar DTOs to plain dicts. - -The functions here operate strictly on Python dicts — no YAML I/O. The -repository layer is responsible for ``yaml.safe_dump`` / ``yaml.safe_load`` -and atomic file writes. Keeping I/O out of the serializer makes it -trivially testable without touching the filesystem. - -The output dict layout matches the schema in ``specs/dot_alfred.md``: - -* Top level: ``schema_version``, ``imdb_id``, ``tmdb_id``, ``seasons``. -* Each season carries ``number``, ``path``, and either pack-mode probed - metadata (``audio`` / ``subtitles``) **or** an ``episodes`` list - (episodic mode, each episode carrying its own probed metadata). -* Subtitles are written as inline-style dicts (handled by the YAML - writer, not here) — at the DTO level they are just regular keys. - -Conventions: - -* Fields that are ``None`` or empty tuples are **omitted** from the - output dict (cleaner YAML, no ``null`` / ``[]`` noise). -* Identity fields (``imdb_id``, ``tmdb_id``) are required; empty - ``seasons`` is allowed (a show with no season is legitimate during - initial population). -* Deserialization is **strict on unknown keys** — a stray field is a - bug, not a feature; raising early prevents silent drift. -* Release identifiers (group/source/quality/codec) are intentionally - absent: they are derived from folder/file names by the parser. -""" - -from __future__ import annotations - -from typing import Any - -from ....domain.shared.value_objects import ImdbId -from ....domain.tv_shows.value_objects import EpisodeNumber, SeasonNumber -from .sidecar import ( - SCHEMA_VERSION, - EpisodeSidecar, - SeasonSidecar, - ShowSidecar, - SubtitleEntry, -) - - -class SidecarSchemaError(ValueError): - """Raised when a sidecar dict does not match the expected schema.""" - - -# ════════════════════════════════════════════════════════════════════════════ -# Serialize — DTO → dict -# ════════════════════════════════════════════════════════════════════════════ - - -def serialize(sidecar: ShowSidecar) -> dict[str, Any]: - """Render a :class:`ShowSidecar` to a plain dict ready for YAML dump.""" - out: dict[str, Any] = { - "schema_version": sidecar.schema_version, - "imdb_id": str(sidecar.imdb_id), - } - if sidecar.tmdb_id is not None: - out["tmdb_id"] = sidecar.tmdb_id - out["seasons"] = [_serialize_season(s) for s in sidecar.seasons] - return out - - -def _serialize_season(season: SeasonSidecar) -> dict[str, Any]: - out: dict[str, Any] = { - "number": season.number.value, - "path": season.path, - } - _put_tracks(out, season.audio_languages, season.subtitles) - if season.episodes: - out["episodes"] = [_serialize_episode(ep) for ep in season.episodes] - return out - - -def _serialize_episode(episode: EpisodeSidecar) -> dict[str, Any]: - out: dict[str, Any] = { - "number": episode.number.value, - "path": episode.path, - } - _put_tracks(out, episode.audio_languages, episode.subtitles) - return out - - -def _put_tracks( - out: dict[str, Any], - audio_languages: tuple[str, ...], - subtitles: tuple[SubtitleEntry, ...], -) -> None: - """Append the optional probed-track fields to ``out`` if set.""" - if audio_languages: - out["audio"] = [{"language": lang} for lang in audio_languages] - if subtitles: - out["subtitles"] = [_serialize_subtitle(sub) for sub in subtitles] - - -def _serialize_subtitle(sub: SubtitleEntry) -> dict[str, Any]: - return {"language": sub.language, "source": sub.source, "type": sub.type} - - -# ════════════════════════════════════════════════════════════════════════════ -# Deserialize — dict → DTO -# ════════════════════════════════════════════════════════════════════════════ - -_ALLOWED_ROOT = {"schema_version", "imdb_id", "tmdb_id", "seasons"} -_ALLOWED_SEASON = {"number", "path", "audio", "subtitles", "episodes"} -_ALLOWED_EPISODE = {"number", "path", "audio", "subtitles"} -_ALLOWED_SUBTITLE = {"language", "source", "type"} -_ALLOWED_AUDIO = {"language"} - - -def deserialize(data: dict[str, Any]) -> ShowSidecar: - """Parse a sidecar dict into a :class:`ShowSidecar`. - - Raises :class:`SidecarSchemaError` on schema violations (unknown - keys, missing required fields, type mismatch, unsupported - ``schema_version``). - """ - _require_dict(data, "root") - _reject_unknown(data, _ALLOWED_ROOT, "root") - - version = data.get("schema_version") - if version != SCHEMA_VERSION: - raise SidecarSchemaError( - f"Unsupported schema_version: {version!r} (expected {SCHEMA_VERSION})" - ) - - imdb_id_raw = data.get("imdb_id") - if not isinstance(imdb_id_raw, str): - raise SidecarSchemaError( - f"imdb_id must be a string, got {type(imdb_id_raw).__name__}" - ) - - tmdb_id_raw = data.get("tmdb_id") - if tmdb_id_raw is not None and not isinstance(tmdb_id_raw, int): - raise SidecarSchemaError( - f"tmdb_id must be an int or absent, got {type(tmdb_id_raw).__name__}" - ) - - seasons_raw = data.get("seasons", []) - if not isinstance(seasons_raw, list): - raise SidecarSchemaError( - f"seasons must be a list, got {type(seasons_raw).__name__}" - ) - - seasons = tuple(_deserialize_season(s) for s in seasons_raw) - - return ShowSidecar( - imdb_id=ImdbId(imdb_id_raw), - tmdb_id=tmdb_id_raw, - seasons=seasons, - schema_version=version, - ) - - -def _deserialize_season(data: Any) -> SeasonSidecar: - _require_dict(data, "season") - _reject_unknown(data, _ALLOWED_SEASON, "season") - - number = _require_int(data, "number", "season") - path = _require_str(data, "path", "season") - episodes_raw = data.get("episodes") - - tracks = _read_tracks(data, "season") - if episodes_raw is not None and not isinstance(episodes_raw, list): - raise SidecarSchemaError( - f"season.episodes must be a list, got {type(episodes_raw).__name__}" - ) - episodes = ( - tuple(_deserialize_episode(e) for e in episodes_raw) - if episodes_raw - else () - ) - - return SeasonSidecar( - number=SeasonNumber(number), - path=path, - episodes=episodes, - **tracks, - ) - - -def _deserialize_episode(data: Any) -> EpisodeSidecar: - _require_dict(data, "episode") - _reject_unknown(data, _ALLOWED_EPISODE, "episode") - - number = _require_int(data, "number", "episode") - path = _require_str(data, "path", "episode") - tracks = _read_tracks(data, "episode") - - return EpisodeSidecar( - number=EpisodeNumber(number), - path=path, - **tracks, - ) - - -def _read_tracks(data: dict[str, Any], where: str) -> dict[str, Any]: - """Extract the optional probed-track fields shared between season and episode.""" - result: dict[str, Any] = {} - - audio_raw = data.get("audio") - if audio_raw is not None: - if not isinstance(audio_raw, list): - raise SidecarSchemaError( - f"{where}.audio must be a list, got {type(audio_raw).__name__}" - ) - result["audio_languages"] = tuple( - _deserialize_audio(entry, where) for entry in audio_raw - ) - - subtitles_raw = data.get("subtitles") - if subtitles_raw is not None: - if not isinstance(subtitles_raw, list): - raise SidecarSchemaError( - f"{where}.subtitles must be a list, got {type(subtitles_raw).__name__}" - ) - result["subtitles"] = tuple( - _deserialize_subtitle(entry) for entry in subtitles_raw - ) - - return result - - -def _deserialize_audio(entry: Any, where: str) -> str: - _require_dict(entry, f"{where}.audio[]") - _reject_unknown(entry, _ALLOWED_AUDIO, f"{where}.audio[]") - language = entry.get("language") - if not isinstance(language, str): - raise SidecarSchemaError( - f"{where}.audio[].language must be a string, " - f"got {type(language).__name__}" - ) - return language - - -def _deserialize_subtitle(entry: Any) -> SubtitleEntry: - _require_dict(entry, "subtitle") - _reject_unknown(entry, _ALLOWED_SUBTITLE, "subtitle") - language = entry.get("language") - source = entry.get("source") - type_ = entry.get("type") - if not isinstance(language, str): - raise SidecarSchemaError( - f"subtitle.language must be a string, got {type(language).__name__}" - ) - if not isinstance(source, str): - raise SidecarSchemaError( - f"subtitle.source must be a string, got {type(source).__name__}" - ) - if not isinstance(type_, str): - raise SidecarSchemaError( - f"subtitle.type must be a string, got {type(type_).__name__}" - ) - return SubtitleEntry(language=language, source=source, type=type_) - - -# ════════════════════════════════════════════════════════════════════════════ -# Schema-checking helpers -# ════════════════════════════════════════════════════════════════════════════ - - -def _require_dict(value: Any, where: str) -> None: - if not isinstance(value, dict): - raise SidecarSchemaError( - f"{where} must be a mapping, got {type(value).__name__}" - ) - - -def _reject_unknown(data: dict[str, Any], allowed: set[str], where: str) -> None: - extra = set(data) - allowed - if extra: - raise SidecarSchemaError( - f"{where} has unknown keys: {sorted(extra)}" - ) - - -def _require_str(data: dict[str, Any], key: str, where: str) -> str: - value = data.get(key) - if not isinstance(value, str): - raise SidecarSchemaError( - f"{where}.{key} must be a string, got {type(value).__name__}" - ) - return value - - -def _require_int(data: dict[str, Any], key: str, where: str) -> int: - value = data.get(key) - if not isinstance(value, int) or isinstance(value, bool): - raise SidecarSchemaError( - f"{where}.{key} must be an int, got {type(value).__name__}" - ) - return value diff --git a/alfred/infrastructure/persistence/dot_alfred/sidecar.py b/alfred/infrastructure/persistence/dot_alfred/sidecar.py deleted file mode 100644 index bf44cab..0000000 --- a/alfred/infrastructure/persistence/dot_alfred/sidecar.py +++ /dev/null @@ -1,87 +0,0 @@ -"""DTOs mirroring the `.alfred` YAML schema. - -These dataclasses are the **in-memory representation** of a single -``.alfred`` file. They mirror the YAML schema described in -``specs/dot_alfred.md`` field-for-field. - -Philosophy: the sidecar exists to avoid two costly operations on every -read — re-walking the show directory and re-probing the media tracks. -Parser-derivable fields (release group, source, quality, codec) are -**not stored**: they live in folder and file names and the parser -reconstructs them on demand. The sidecar only caches what is not -otherwise free — folder/file paths (to skip the walk) and probed track -metadata (audio languages, subtitles — to skip ffprobe). - -Schema version: 1. -""" - -from __future__ import annotations - -from dataclasses import dataclass, field - -from ....domain.shared.value_objects import ImdbId -from ....domain.tv_shows.value_objects import EpisodeNumber, SeasonNumber - -SCHEMA_VERSION = 1 - - -@dataclass(frozen=True) -class SubtitleEntry: - """One subtitle row, as it appears under ``subtitles:`` in YAML.""" - - language: str - source: str # "embedded" | "adjacent" - type: str # "standard" | "sdh" | "forced" - - -@dataclass(frozen=True) -class EpisodeSidecar: - """One episode entry under ``episodes:`` in episodic mode. - - Carries only probed track metadata — release identifiers - (group/source/quality/codec) are derived from the filename by the - parser, not duplicated here. - """ - - number: EpisodeNumber - path: str - audio_languages: tuple[str, ...] = () - subtitles: tuple[SubtitleEntry, ...] = () - - -@dataclass(frozen=True) -class SeasonSidecar: - """One season block in the sidecar. - - Two storage modes are encoded structurally: - - * **PACK** — ``episodes`` is empty; ``audio_languages`` / - ``subtitles`` describe the season as a whole (VO-only policy means - all episodes share the same audio set). - * **EPISODIC** — ``episodes`` is populated; per-episode track data - lives on each :class:`EpisodeSidecar`. - - Release identifiers (group/source/quality/codec) come from parsing - the season folder name and are not stored. - """ - - number: SeasonNumber - path: str - audio_languages: tuple[str, ...] = () - subtitles: tuple[SubtitleEntry, ...] = () - episodes: tuple[EpisodeSidecar, ...] = () - - -@dataclass(frozen=True) -class ShowSidecar: - """Root DTO — one ``.alfred`` file maps to one ``ShowSidecar``. - - Identity-only at the root (``imdb_id`` / ``tmdb_id``). The show's - display title is the parent directory name on disk, not stored - here. - """ - - imdb_id: ImdbId - tmdb_id: int | None = None - seasons: tuple[SeasonSidecar, ...] = field(default_factory=tuple) - schema_version: int = SCHEMA_VERSION diff --git a/tests/infrastructure/persistence/dot_alfred/test_repository.py b/tests/infrastructure/persistence/dot_alfred/test_repository.py deleted file mode 100644 index f467bf9..0000000 --- a/tests/infrastructure/persistence/dot_alfred/test_repository.py +++ /dev/null @@ -1,316 +0,0 @@ -"""Tests for the filesystem-backed ``.alfred`` repository.""" - -from __future__ import annotations - -import pytest -import yaml - -# Phase 3 (refactor/dot-alfred-v2): v1 repository is intentionally -# left in tree as a frozen reference until Phase 4 deletes both v1 -# and this test module in one swing. -pytest.skip( - "v1 dot_alfred repository — replaced in Phase 4", - allow_module_level=True, -) - -from alfred.domain.shared.media import AudioTrack, SubtitleTrack -from alfred.domain.shared.value_objects import FilePath, ImdbId -from alfred.domain.tv_shows.builders import TVShowBuilder -from alfred.domain.tv_shows.entities import Episode -from alfred.infrastructure.persistence.dot_alfred import ( - DotAlfredTVShowRepository, - ShowFolderUnknown, -) -from alfred.infrastructure.persistence.dot_alfred.repository import ( - SIDECAR_FILENAME, -) - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - - -def _make_breaking_bad_episodic(): - """Breaking Bad with one EPISODIC season carrying two episodes.""" - return ( - TVShowBuilder( - imdb_id="tt0903747", - title="Breaking Bad", - tmdb_id=1396, - ) - .add_episode( - Episode( - season_number=5, - episode_number=1, - title="Live Free or Die", - file_path=FilePath("Breaking.Bad.S05E01.mkv"), - audio_tracks=( - AudioTrack( - index=0, - codec=None, - channels=None, - channel_layout=None, - language="eng", - ), - ), - subtitle_tracks=( - SubtitleTrack( - index=0, - codec=None, - language="eng", - is_default=False, - is_forced=False, - ), - ), - ) - ) - .add_episode( - Episode( - season_number=5, - episode_number=2, - title="Madrigal", - file_path=FilePath("Breaking.Bad.S05E02.mkv"), - audio_tracks=( - AudioTrack( - index=0, - codec=None, - channels=None, - channel_layout=None, - language="eng", - ), - ), - ) - ) - .build() - ) - - -def _make_foundation_pack(): - """Foundation as a PACK season (no episodes in the aggregate).""" - return TVShowBuilder( - imdb_id="tt0804484", - title="Foundation", - tmdb_id=84958, - ).build() - - -# --------------------------------------------------------------------------- -# save + find_by_imdb_id round-trip -# --------------------------------------------------------------------------- - - -class TestSaveAndRead: - def test_save_then_find_by_imdb_id(self, tmp_path): - # Folder must exist before save (the repo never invents one). - (tmp_path / "Breaking.Bad").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - show = _make_breaking_bad_episodic() - repo.save(show) - - recovered = repo.find_by_imdb_id(ImdbId("tt0903747")) - assert recovered is not None - assert recovered.imdb_id == show.imdb_id - assert recovered.tmdb_id == show.tmdb_id - assert recovered.seasons_count == 1 - assert recovered.episode_count == 2 - - def test_find_by_imdb_id_uses_folder_name_as_title(self, tmp_path): - (tmp_path / "Breaking.Bad").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - - recovered = repo.find_by_imdb_id(ImdbId("tt0903747")) - assert recovered is not None - assert recovered.title == "Breaking.Bad" - - def test_find_returns_none_for_unknown(self, tmp_path): - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.find_by_imdb_id(ImdbId("tt9999999")) is None - - def test_find_returns_none_for_cold_folder(self, tmp_path): - # Folder exists but has no .alfred — cold scan returns None. - (tmp_path / "Breaking.Bad").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.find_by_imdb_id(ImdbId("tt0903747")) is None - - def test_pack_season_round_trip(self, tmp_path): - (tmp_path / "Foundation").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_foundation_pack()) - recovered = repo.find_by_imdb_id(ImdbId("tt0804484")) - assert recovered is not None - assert recovered.seasons_count == 0 - - -# --------------------------------------------------------------------------- -# find_all -# --------------------------------------------------------------------------- - - -class TestFindAll: - def test_returns_every_sidecar(self, tmp_path): - (tmp_path / "Breaking.Bad").mkdir() - (tmp_path / "Foundation").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - repo.save(_make_foundation_pack()) - - all_shows = repo.find_all() - ids = {str(s.imdb_id) for s in all_shows} - assert ids == {"tt0903747", "tt0804484"} - - def test_skips_folders_without_sidecar(self, tmp_path): - (tmp_path / "Breaking.Bad").mkdir() - (tmp_path / "ColdFolder").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - assert len(repo.find_all()) == 1 - - def test_skips_corrupted_sidecar(self, tmp_path, caplog): - cold = tmp_path / "Garbage" - cold.mkdir() - (cold / SIDECAR_FILENAME).write_text("not: valid: yaml: :{[") - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.find_all() == [] - - def test_skips_schema_violation(self, tmp_path): - bad = tmp_path / "WrongSchema" - bad.mkdir() - (bad / SIDECAR_FILENAME).write_text( - yaml.safe_dump({"schema_version": 99, "imdb_id": "tt0", "seasons": []}) - ) - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.find_all() == [] - - def test_empty_library(self, tmp_path): - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.find_all() == [] - - -# --------------------------------------------------------------------------- -# delete + exists -# --------------------------------------------------------------------------- - - -class TestDeleteAndExists: - def test_exists_after_save(self, tmp_path): - (tmp_path / "Breaking.Bad").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - assert repo.exists(ImdbId("tt0903747")) is True - - def test_exists_false_before_save(self, tmp_path): - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.exists(ImdbId("tt0903747")) is False - - def test_delete_removes_sidecar(self, tmp_path): - show_dir = tmp_path / "Breaking.Bad" - show_dir.mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - assert (show_dir / SIDECAR_FILENAME).is_file() - - assert repo.delete(ImdbId("tt0903747")) is True - assert not (show_dir / SIDECAR_FILENAME).exists() - # The show folder itself stays — the repo only owns the sidecar. - assert show_dir.is_dir() - - def test_delete_returns_false_when_unknown(self, tmp_path): - repo = DotAlfredTVShowRepository(tmp_path) - assert repo.delete(ImdbId("tt9999999")) is False - - -# --------------------------------------------------------------------------- -# save edge cases -# --------------------------------------------------------------------------- - - -class TestSaveEdgeCases: - def test_save_raises_when_folder_missing(self, tmp_path): - repo = DotAlfredTVShowRepository(tmp_path) - with pytest.raises(ShowFolderUnknown): - repo.save(_make_breaking_bad_episodic()) - - def test_save_atomic_no_tmp_file_left_behind(self, tmp_path): - show_dir = tmp_path / "Breaking.Bad" - show_dir.mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - # Only the final .alfred should remain. - leftovers = [p.name for p in show_dir.iterdir()] - assert leftovers == [SIDECAR_FILENAME] - - def test_save_overwrites_existing(self, tmp_path): - show_dir = tmp_path / "Breaking.Bad" - show_dir.mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - # Save a second time with a different aggregate — sidecar must - # carry the new content. - updated = TVShowBuilder( - imdb_id="tt0903747", - title="Breaking Bad", - tmdb_id=1396, - ).build() - repo.save(updated) - - recovered = repo.find_by_imdb_id(ImdbId("tt0903747")) - assert recovered is not None - assert recovered.seasons_count == 0 # the second save dropped the season - - def test_save_finds_folder_via_get_folder_name(self, tmp_path): - # show.get_folder_name() returns "Breaking.Bad" for title "Breaking Bad" - (tmp_path / "Breaking.Bad").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - # No prior save → cache empty → falls back to get_folder_name(). - repo.save(_make_breaking_bad_episodic()) - assert (tmp_path / "Breaking.Bad" / SIDECAR_FILENAME).is_file() - - def test_save_falls_back_to_scan_when_folder_renamed(self, tmp_path): - # Create a custom folder name (not the show's default). - custom = tmp_path / "Breaking.Bad.1080p" - custom.mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - # Pre-populate the index by reading the sidecar — first we need - # to put one there. Write a minimal one by hand. - (custom / SIDECAR_FILENAME).write_text( - yaml.safe_dump( - { - "schema_version": 1, - "imdb_id": "tt0903747", - "tmdb_id": 1396, - "seasons": [], - } - ) - ) - # find_all primes the folder index. - repo.find_all() - # Now save through the repo — it must reuse the custom folder. - repo.save(_make_breaking_bad_episodic()) - assert (custom / SIDECAR_FILENAME).is_file() - # Default folder was never created. - assert not (tmp_path / "Breaking.Bad").exists() - - -# --------------------------------------------------------------------------- -# Sidecar content sanity -# --------------------------------------------------------------------------- - - -class TestSidecarContent: - def test_written_yaml_matches_schema(self, tmp_path): - (tmp_path / "Breaking.Bad").mkdir() - repo = DotAlfredTVShowRepository(tmp_path) - repo.save(_make_breaking_bad_episodic()) - text = (tmp_path / "Breaking.Bad" / SIDECAR_FILENAME).read_text() - data = yaml.safe_load(text) - assert data["schema_version"] == 1 - assert data["imdb_id"] == "tt0903747" - assert data["tmdb_id"] == 1396 - # One EPISODIC season with two episodes. - assert len(data["seasons"]) == 1 - season = data["seasons"][0] - assert season["number"] == 5 - assert len(season["episodes"]) == 2 - # Episode paths come from FilePath. - assert season["episodes"][0]["path"] == "Breaking.Bad.S05E01.mkv" diff --git a/tests/infrastructure/persistence/dot_alfred/test_serializer.py b/tests/infrastructure/persistence/dot_alfred/test_serializer.py deleted file mode 100644 index 14f90b9..0000000 --- a/tests/infrastructure/persistence/dot_alfred/test_serializer.py +++ /dev/null @@ -1,433 +0,0 @@ -"""Tests for the ``.alfred`` sidecar serializer. - -Covers: - -* Round-trip equivalence (``serialize`` → ``deserialize`` → equal DTO). -* Field omission rules (``None`` / empty tuples never make it to dict). -* Strict schema (unknown keys rejected, missing keys raise clearly). -* The Foundation fixture (real-world PACK season with mixed subtitles) - to exercise the full surface on a realistic case. - -The serializer is pure-dict in/out; YAML I/O lives in the repository -layer and is tested separately. - -Note: release identifiers (group/source/quality/codec) live in folder -and file names — the parser derives them on demand. They are -deliberately absent from the sidecar schema. -""" - -from __future__ import annotations - -import pytest -import yaml - -# Phase 3 (refactor/dot-alfred-v2): v1 serializer is intentionally -# left in tree as a frozen reference until Phase 4 deletes both v1 -# and this test module in one swing. -pytest.skip( - "v1 dot_alfred serializer — replaced in Phase 4", - allow_module_level=True, -) - -from alfred.domain.shared.value_objects import ImdbId # noqa: E402 -from alfred.domain.tv_shows.value_objects import EpisodeNumber, SeasonNumber -from alfred.infrastructure.persistence.dot_alfred import ( - EpisodeSidecar, - SeasonSidecar, - ShowSidecar, - SubtitleEntry, - deserialize, - serialize, -) -from alfred.infrastructure.persistence.dot_alfred.serializer import ( - SidecarSchemaError, -) - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - - -def _foundation_sidecar() -> ShowSidecar: - """The Foundation S01 PACK season — real-world fixture data. - - Mirrors the layout seen in - ``/mnt/testipool/tv_shows/Foundation.2021.1080p.WEBRip.x265-RARBG/`` — - superset audio/subs at season level (some episodes have a forced - English sub, captured at season scope). - """ - return ShowSidecar( - imdb_id=ImdbId("tt0804484"), - tmdb_id=84958, - seasons=( - SeasonSidecar( - number=SeasonNumber(1), - path="Foundation.2021.S01.1080p.WEBRip.x265-RARBG", - audio_languages=("eng",), - subtitles=( - SubtitleEntry(language="eng", source="adjacent", type="standard"), - SubtitleEntry(language="eng", source="adjacent", type="sdh"), - SubtitleEntry(language="eng", source="adjacent", type="forced"), - SubtitleEntry(language="fra", source="adjacent", type="standard"), - SubtitleEntry(language="fra", source="adjacent", type="sdh"), - ), - ), - ), - ) - - -def _minimal_sidecar() -> ShowSidecar: - """Identity-only sidecar — no seasons, no track data.""" - return ShowSidecar(imdb_id=ImdbId("tt0903747")) - - -def _episodic_sidecar() -> ShowSidecar: - """A season in EPISODIC mode (per-episode track metadata).""" - return ShowSidecar( - imdb_id=ImdbId("tt0903747"), - tmdb_id=1396, - seasons=( - SeasonSidecar( - number=SeasonNumber(5), - path="Breaking.Bad.S05", - episodes=( - EpisodeSidecar( - number=EpisodeNumber(1), - path="Breaking.Bad.S05E01.Live.Free.or.Die-MeGusta/Breaking.Bad.S05E01.mkv", - audio_languages=("eng",), - subtitles=( - SubtitleEntry( - language="eng", source="embedded", type="standard" - ), - ), - ), - EpisodeSidecar( - number=EpisodeNumber(2), - path="Breaking.Bad.S05E02.Madrigal-CtrlHD/Breaking.Bad.S05E02.mkv", - audio_languages=("eng",), - ), - ), - ), - ), - ) - - -# --------------------------------------------------------------------------- -# Round-trip -# --------------------------------------------------------------------------- - - -class TestRoundTrip: - def test_minimal(self): - original = _minimal_sidecar() - assert deserialize(serialize(original)) == original - - def test_foundation_pack_season(self): - original = _foundation_sidecar() - assert deserialize(serialize(original)) == original - - def test_episodic_breaking_bad(self): - original = _episodic_sidecar() - assert deserialize(serialize(original)) == original - - def test_round_trip_through_yaml(self): - """Full pipeline: DTO → dict → YAML text → dict → DTO.""" - original = _foundation_sidecar() - text = yaml.safe_dump(serialize(original), sort_keys=False) - recovered = deserialize(yaml.safe_load(text)) - assert recovered == original - - -# --------------------------------------------------------------------------- -# Serialize — field omission -# --------------------------------------------------------------------------- - - -class TestSerializeOmission: - def test_tmdb_id_omitted_when_none(self): - out = serialize(_minimal_sidecar()) - assert "tmdb_id" not in out - - def test_empty_seasons_is_empty_list_not_omitted(self): - # We always emit `seasons:` even if empty — the key documents the - # show "has no season recorded yet" vs being entirely missing. - out = serialize(_minimal_sidecar()) - assert out["seasons"] == [] - - def test_no_audio_when_empty(self): - sidecar = ShowSidecar( - imdb_id=ImdbId("tt0903747"), - seasons=(SeasonSidecar(number=SeasonNumber(1), path="X.S01"),), - ) - out = serialize(sidecar) - assert "audio" not in out["seasons"][0] - - def test_no_subtitles_when_empty(self): - sidecar = ShowSidecar( - imdb_id=ImdbId("tt0903747"), - seasons=(SeasonSidecar(number=SeasonNumber(1), path="X.S01"),), - ) - out = serialize(sidecar) - assert "subtitles" not in out["seasons"][0] - - def test_no_episodes_when_pack(self): - sidecar = ShowSidecar( - imdb_id=ImdbId("tt0903747"), - seasons=(SeasonSidecar(number=SeasonNumber(1), path="X.S01"),), - ) - out = serialize(sidecar) - assert "episodes" not in out["seasons"][0] - - def test_parser_derivable_fields_never_emitted(self): - """group/source/quality/codec must never appear in the YAML.""" - out = serialize(_foundation_sidecar()) - season = out["seasons"][0] - for forbidden in ("group", "source", "quality", "codec"): - assert forbidden not in season - - -# --------------------------------------------------------------------------- -# Serialize — shape -# --------------------------------------------------------------------------- - - -class TestSerializeShape: - def test_root_keys(self): - out = serialize(_foundation_sidecar()) - assert out["schema_version"] == 1 - assert out["imdb_id"] == "tt0804484" - assert out["tmdb_id"] == 84958 - assert isinstance(out["seasons"], list) - - def test_season_number_is_int(self): - out = serialize(_foundation_sidecar()) - assert out["seasons"][0]["number"] == 1 - assert isinstance(out["seasons"][0]["number"], int) - - def test_audio_as_list_of_dicts(self): - out = serialize(_foundation_sidecar()) - assert out["seasons"][0]["audio"] == [{"language": "eng"}] - - def test_subtitle_structure(self): - out = serialize(_foundation_sidecar()) - subs = out["seasons"][0]["subtitles"] - assert subs[0] == { - "language": "eng", - "source": "adjacent", - "type": "standard", - } - - -# --------------------------------------------------------------------------- -# Deserialize — strict schema -# --------------------------------------------------------------------------- - - -class TestDeserializeStrict: - def _valid_minimal(self) -> dict: - return { - "schema_version": 1, - "imdb_id": "tt0903747", - "seasons": [], - } - - def test_unknown_root_key_raises(self): - data = self._valid_minimal() - data["bogus"] = "x" - with pytest.raises(SidecarSchemaError, match="root has unknown keys"): - deserialize(data) - - def test_unknown_season_key_raises(self): - data = self._valid_minimal() - data["seasons"] = [{"number": 1, "path": "X", "weird": True}] - with pytest.raises(SidecarSchemaError, match="season has unknown keys"): - deserialize(data) - - def test_parser_derivable_season_key_raises(self): - """A stray group/source/quality/codec key must be rejected.""" - data = self._valid_minimal() - data["seasons"] = [{"number": 1, "path": "X", "group": "RARBG"}] - with pytest.raises(SidecarSchemaError, match="season has unknown keys"): - deserialize(data) - - def test_unknown_episode_key_raises(self): - data = self._valid_minimal() - data["seasons"] = [ - { - "number": 1, - "path": "X", - "episodes": [{"number": 1, "path": "p", "huh": 1}], - } - ] - with pytest.raises(SidecarSchemaError, match="episode has unknown keys"): - deserialize(data) - - def test_unknown_subtitle_key_raises(self): - data = self._valid_minimal() - data["seasons"] = [ - { - "number": 1, - "path": "X", - "subtitles": [ - {"language": "eng", "source": "adjacent", "type": "sdh", "x": 1} - ], - } - ] - with pytest.raises(SidecarSchemaError, match="subtitle has unknown keys"): - deserialize(data) - - def test_unknown_audio_key_raises(self): - data = self._valid_minimal() - data["seasons"] = [ - { - "number": 1, - "path": "X", - "audio": [{"language": "eng", "channels": 6}], - } - ] - with pytest.raises(SidecarSchemaError, match=r"audio\[\] has unknown keys"): - deserialize(data) - - def test_wrong_schema_version_raises(self): - data = self._valid_minimal() - data["schema_version"] = 2 - with pytest.raises(SidecarSchemaError, match="schema_version"): - deserialize(data) - - def test_missing_schema_version_raises(self): - data = self._valid_minimal() - del data["schema_version"] - with pytest.raises(SidecarSchemaError, match="schema_version"): - deserialize(data) - - def test_imdb_id_must_be_string(self): - data = self._valid_minimal() - data["imdb_id"] = 12345 - with pytest.raises(SidecarSchemaError, match="imdb_id must be a string"): - deserialize(data) - - def test_tmdb_id_must_be_int_when_present(self): - data = self._valid_minimal() - data["tmdb_id"] = "1396" - with pytest.raises(SidecarSchemaError, match="tmdb_id"): - deserialize(data) - - def test_seasons_must_be_list(self): - data = self._valid_minimal() - data["seasons"] = {"1": {}} - with pytest.raises(SidecarSchemaError, match="seasons must be a list"): - deserialize(data) - - def test_season_number_must_be_int(self): - data = self._valid_minimal() - data["seasons"] = [{"number": "1", "path": "X"}] - with pytest.raises(SidecarSchemaError, match="season.number must be an int"): - deserialize(data) - - def test_season_number_bool_rejected(self): - # bool is a subclass of int but should not pass — guards against - # YAML quirks where `True` could sneak in as a season number. - data = self._valid_minimal() - data["seasons"] = [{"number": True, "path": "X"}] - with pytest.raises(SidecarSchemaError, match="season.number must be an int"): - deserialize(data) - - def test_season_path_must_be_string(self): - data = self._valid_minimal() - data["seasons"] = [{"number": 1, "path": 1}] - with pytest.raises(SidecarSchemaError, match="season.path"): - deserialize(data) - - def test_subtitle_missing_field_raises(self): - data = self._valid_minimal() - data["seasons"] = [ - { - "number": 1, - "path": "X", - "subtitles": [{"language": "eng", "source": "adjacent"}], - } - ] - with pytest.raises(SidecarSchemaError, match="subtitle.type"): - deserialize(data) - - -# --------------------------------------------------------------------------- -# Foundation fixture — golden YAML -# --------------------------------------------------------------------------- - - -class TestFoundationGolden: - """Use the Foundation case to validate the produced YAML reads well.""" - - def test_yaml_dump_shape(self): - text = yaml.safe_dump(serialize(_foundation_sidecar()), sort_keys=False) - # Sanity-check that the human-readable layout matches the spec. - assert "schema_version: 1" in text - assert "imdb_id: tt0804484" in text - assert "tmdb_id: 84958" in text - assert "- number: 1" in text - assert "path: Foundation.2021.S01.1080p.WEBRip.x265-RARBG" in text - # No episodes block (PACK mode). - assert "episodes:" not in text - # No release identifiers at season scope — those live in folder - # names. (We can't check ``source:`` here because the subtitle - # entries legitimately carry their own ``source`` key.) - for forbidden in ("group:", "quality:", "codec:"): - assert forbidden not in text - - -# --------------------------------------------------------------------------- -# Foundation on-disk fixture (real folder structure, no real .mkv) -# --------------------------------------------------------------------------- - - -@pytest.fixture -def foundation_tree(tmp_path): - """Recreate the Foundation S01 layout in a tmp directory. - - Mirrors the on-disk structure of - ``/mnt/testipool/tv_shows/Foundation.2021.1080p.WEBRip.x265-RARBG/`` - using empty placeholder files — sufficient for tests that need a - realistic show folder without dragging in real media. - """ - show = tmp_path / "Foundation.2021.1080p.WEBRip.x265-RARBG" - season = show / "Foundation.2021.S01.1080p.WEBRip.x265-RARBG" - season.mkdir(parents=True) - base = "Foundation.2021.S01E{n:02d}.1080p.WEBRip.x265-RARBG" - for ep in range(1, 11): - stem = base.format(n=ep) - (season / f"{stem}.mp4").touch() - (season / f"{stem}.eng.srt").touch() - (season / f"{stem}.eng.sdh.srt").touch() - (season / f"{stem}.fra.srt").touch() - (season / f"{stem}.fra.sdh.srt").touch() - if 4 <= ep <= 9: - (season / f"{stem}.eng.forced.srt").touch() - return show - - -class TestFoundationOnDisk: - """The on-disk fixture is mostly for future tests (repository walk). - - For now we exercise the basic shape — a placeholder for richer - walk-and-build tests landing in step 3 (repository). - """ - - def test_fixture_has_expected_episode_count(self, foundation_tree): - season = foundation_tree / "Foundation.2021.S01.1080p.WEBRip.x265-RARBG" - mkvs = sorted(season.glob("*.mp4")) - assert len(mkvs) == 10 - - def test_fixture_has_forced_subs_only_on_some_episodes(self, foundation_tree): - season = foundation_tree / "Foundation.2021.S01.1080p.WEBRip.x265-RARBG" - forced = sorted(season.glob("*.eng.forced.srt")) - assert len(forced) == 6 # E04 through E09 - - def test_serialize_yaml_can_be_written_alongside(self, foundation_tree): - """Write the sidecar next to the show folder and read it back.""" - sidecar_path = foundation_tree / ".alfred" - sidecar_path.write_text( - yaml.safe_dump(serialize(_foundation_sidecar()), sort_keys=False) - ) - recovered = deserialize(yaml.safe_load(sidecar_path.read_text())) - assert recovered == _foundation_sidecar()