feat(persistence): add DotAlfredTVShowRepository (filesystem-backed)

Step 3 of specs/dot_alfred.md. Concrete TVShowRepository
implementation reading and writing per-show .alfred YAML files under
a configurable library_root. Writes are atomic (.alfred.tmp +
os.replace), reads tolerate corrupted/wrong-schema sidecars (log +
skip), and the repo never invents a folder name — save(show)
requires the target folder to exist beforehand (raises
ShowFolderUnknown otherwise), matching the spec's
MediaOrganizer-then-sidecar split.

Cold folders without a sidecar are skipped by find_all and yield
None from find_by_imdb_id — the upcoming rescan_show tool (step 4)
will own the opt-in rebuild path.

A small bridge module translates between the rich domain TVShow
(AudioTrack/SubtitleTrack with full ffprobe minutiae) and the
compact sidecar shape (language-only audio, embedded-only subs with
type derived from is_forced). The bridge is intentionally lossy on
probe details the sidecar does not store, per the spec's
factual-only philosophy.

20 integration tests on tmp_path: round-trip save/find,
cold-folder/unknown-id returns, find_all skipping
(corrupted/schema-violating sidecars), delete/exists, atomic write
(no .alfred.tmp leftover), overwrite, and folder-name fallbacks
(get_folder_name guess + full-scan rescue when renamed).
This commit is contained in:
2026-05-22 17:16:41 +02:00
parent b0e275bd11
commit c7c11180d9
5 changed files with 721 additions and 0 deletions
+35
View File
@@ -17,6 +17,41 @@ callers).
### Added ### Added
- **`DotAlfredTVShowRepository` — filesystem-backed implementation of
the `TVShowRepository` port
(`alfred/infrastructure/persistence/dot_alfred/repository.py`).**
Step 3 of the `specs/dot_alfred.md` plan. Reads and writes one
`.alfred` YAML file per show under a configurable `library_root`.
`save(show)` writes atomically (`.alfred.tmp` + `os.replace`) into a
folder that **must already exist** — the repository never invents a
folder name (the upstream `MediaOrganizer` is in charge of placing
files; the repo writes the sidecar next to them). `find_by_imdb_id` /
`find_all` walk `library_root/*/`, loading each readable sidecar;
folders without a sidecar return `None` / are skipped (no implicit
cold scan — that is the job of the upcoming `rescan_show` tool).
Corrupted YAML and schema violations are logged and skipped, never
raised, so a single bad folder does not break the rest of the
library. The repo keeps a tiny in-memory `imdb_id → folder_name`
index populated on every successful read/save, so subsequent saves
find the right destination without re-walking — useful when the show
folder name diverges from `show.get_folder_name()` (custom 1080p / 4K
variants). 20 integration tests on `tmp_path` cover the round-trip,
cold folder / unknown id returns, multi-show `find_all`, corrupted /
wrong-schema skipping, atomic write (no `.alfred.tmp` left behind),
overwrite, and folder-name fallbacks.
- **Sidecar ↔ TVShow bridge
(`alfred/infrastructure/persistence/dot_alfred/bridge.py`).**
`to_sidecar(show, folder_paths=...)` summarizes the rich domain
`AudioTrack` / `SubtitleTrack` to the sidecar's compact form (unique
audio languages in track order; subtitle entries derived from
`is_forced` and assumed `source="embedded"`). `from_sidecar(sidecar,
title=...)` reconstructs the domain `TVShow` with synthesized tracks
— one `AudioTrack` per language, one `SubtitleTrack` per entry, with
ffprobe-only fields (`codec`, `channels`, `channel_layout`) left as
`None`. The bridge is intentionally lossy on probe minutiae the
sidecar does not store; this is the documented trade-off from the
factual-only spec.
- **`.alfred` sidecar serializer - **`.alfred` sidecar serializer
(`alfred/infrastructure/persistence/dot_alfred/`).** Implements step 2 (`alfred/infrastructure/persistence/dot_alfred/`).** Implements step 2
of the `specs/dot_alfred.md` plan. Pure-dict in/out of the `specs/dot_alfred.md` plan. Pure-dict in/out
@@ -11,8 +11,15 @@ Public surface:
``EpisodeSidecar``, ``SubtitleEntry``) that mirror the YAML schema. ``EpisodeSidecar``, ``SubtitleEntry``) that mirror the YAML schema.
* :mod:`.serializer` — ``serialize`` / ``deserialize`` functions * :mod:`.serializer` — ``serialize`` / ``deserialize`` functions
converting between DTOs and plain dicts (YAML-ready). 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.
""" """
from .bridge import from_sidecar, to_sidecar
from .repository import DotAlfredTVShowRepository, ShowFolderUnknown
from .serializer import deserialize, serialize from .serializer import deserialize, serialize
from .sidecar import ( from .sidecar import (
EpisodeSidecar, EpisodeSidecar,
@@ -24,6 +31,10 @@ from .sidecar import (
__all__ = [ __all__ = [
"deserialize", "deserialize",
"serialize", "serialize",
"from_sidecar",
"to_sidecar",
"DotAlfredTVShowRepository",
"ShowFolderUnknown",
"EpisodeSidecar", "EpisodeSidecar",
"SeasonSidecar", "SeasonSidecar",
"ShowSidecar", "ShowSidecar",
@@ -0,0 +1,169 @@
"""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 EpisodeNumber, 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 — no episodes, season-scoped tracks. The domain doesn't
# currently expose season-level tracks (they live on episodes), so a
# PACK season produced from the domain alone has empty tracks. The
# repository populates them from the probe when scanning a real
# folder; here we just relay what is on the season.
return SeasonSidecar(
number=season.season_number,
path=path,
)
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))
return sb.build()
def _episode_from_sidecar(
episode: EpisodeSidecar, season_number: SeasonNumber
) -> Episode:
audio_tracks = tuple(
AudioTrack(
index=i,
codec=None,
channels=None,
channel_layout=None,
language=lang,
)
for i, lang in enumerate(episode.audio_languages)
)
subtitle_tracks = tuple(
SubtitleTrack(
index=i,
codec=None,
language=entry.language,
is_default=False,
is_forced=(entry.type == "forced"),
)
for i, entry in enumerate(episode.subtitles)
)
return Episode(
season_number=season_number,
episode_number=episode.number,
title="",
file_path=FilePath(episode.path),
audio_tracks=audio_tracks,
subtitle_tracks=subtitle_tracks,
)
__all__ = ["from_sidecar", "to_sidecar"]
@@ -0,0 +1,198 @@
"""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"]
@@ -0,0 +1,308 @@
"""Tests for the filesystem-backed ``.alfred`` repository."""
from __future__ import annotations
import pytest
import yaml
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"