francwa/alfred
1
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor(persistence): Phase 4 Step 3 — delete v1 dot_alfred + ports

Browse Source 
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.
This commit is contained in:
Francwa
2026-05-25 21:10:32 +02:00
parent 9e48c70b8a
commit 86222d95d1
9 changed files with 20 additions and 1657 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alfred/domain/movies/repositories.py
-73
View File
@@ -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
alfred/domain/tv_shows/repositories.py
-40
View File
@@ -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."""
alfred/infrastructure/persistence/dot_alfred/__init__.py
+20 -32
View File
@@ -1,42 +1,30 @@
"""`.alfred` sidecar persistence layer. """`.alfred` sidecar persistence (v2-only surface).
Implements the per-show YAML sidecar described in v1 was deleted in Phase 4 of the v2 migration (see
``specs/dot_alfred.md``. The sidecar is a single file named ``.alfred`` ``.claude/specs/dot_alfred_v2_phase4.md``). This package now exposes
posed at the root of a show's directory, containing the full aggregate only the v2 stack:
in a factual-only schema.
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``, For schema-level DTOs (``SeriesReleaseSidecar`` etc.) import directly
``EpisodeSidecar``, ``SubtitleEntry``) that mirror the YAML schema. from :mod:`alfred.infrastructure.persistence.dot_alfred.v2`. The
* :mod:`.serializer` — ``serialize`` / ``deserialize`` functions top-level package re-exports the four concrete repositories that
converting between DTOs and plain dicts (YAML-ready). application orchestrators talk to.
* :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.
""" """
from .bridge import from_sidecar, to_sidecar from .v2.repository import (
from .repository import DotAlfredTVShowRepository, ShowFolderUnknown DotAlfredMovieLibraryIndex,
from .serializer import deserialize, serialize DotAlfredMovieReleaseRepository,
from .sidecar import ( DotAlfredSeriesReleaseRepository,
EpisodeSidecar, DotAlfredTVShowLibraryIndex,
SeasonSidecar, ShowFolderUnknown,
ShowSidecar,
SubtitleEntry,
) )
__all__ = [ __all__ = [
"deserialize", "DotAlfredMovieLibraryIndex",
"serialize", "DotAlfredMovieReleaseRepository",
"from_sidecar", "DotAlfredSeriesReleaseRepository",
"to_sidecar", "DotAlfredTVShowLibraryIndex",
"DotAlfredTVShowRepository",
"ShowFolderUnknown", "ShowFolderUnknown",
"EpisodeSidecar",
"SeasonSidecar",
"ShowSidecar",
"SubtitleEntry",
] ]
alfred/infrastructure/persistence/dot_alfred/bridge.py
-184
View File
@@ -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"]
alfred/infrastructure/persistence/dot_alfred/repository.py
-198
View File
@@ -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"]
alfred/infrastructure/persistence/dot_alfred/serializer.py
-294
View File
@@ -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
alfred/infrastructure/persistence/dot_alfred/sidecar.py
-87
View File
@@ -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
tests/infrastructure/persistence/dot_alfred/test_repository.py
-316
View File
@@ -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"
tests/infrastructure/persistence/dot_alfred/test_serializer.py
-433
View File
@@ -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()