refactor(music)!: make Song an ordinary Title on the generic path

Delete the dead grouped-download block, the planner, and the music service hooks; Song now flows through the one download loop, gets tagged via tag_file, and answers -w with {disc}x{track} keys.

BREAKING CHANGE: GROUP_AUDIO_DOWNLOADS, get_music_track_options and get_music_collection_label are gone. A music service returns one Audio track per quality and lets the framework choose.
This commit is contained in:
imSp4rky
2026-08-18 15:05:30 -06:00
parent a4a385e5aa
commit 68c7d95173
43 changed files with 1082 additions and 2132 deletions
-213
View File
@@ -1,213 +0,0 @@
"""Unit tests for the shared music rendering helpers in unshackle.core.music.display.
Rendering is visual, so these assert structure/type/no-crash rather than exact
ANSI. No network: the artwork soft-fail path is exercised with a session whose
get raises, and (when Pillow is absent) via the missing-dependency guard.
"""
from __future__ import annotations
from typing import Any, Optional
from rich.console import Console, RenderableType
from rich.panel import Panel
from unshackle.core.music.display import (
MusicHeaderInfo,
TrackRow,
format_track_detail_quality,
render_album_header,
render_artwork_preview,
render_track_panel,
)
from unshackle.core.titles.music import Song
class DummyService:
"""Stand-in service class; Song only requires a type, never an instance."""
def make_song(*, track: int = 1, name: str = "Song", album: str = "Album") -> Song:
return Song(
id_=f"{album}-{track}",
service=DummyService,
name=name,
artist="Artist",
album=album,
track=track,
disc=1,
year=2020,
)
def render_to_text(renderable: RenderableType) -> str:
"""Render a rich renderable to plain text so we can assert on its content."""
console = Console(width=120, record=True, force_terminal=False)
console.print(renderable)
return console.export_text()
def test_trackrow_constructs() -> None:
song = make_song(name="Track A")
row = TrackRow(
song=song,
quality_label="FLAC 16-bit/44.1kHz",
layout="Stereo",
duration_str="3:21",
hires=True,
cd=True,
note="",
)
assert row.song is song
assert row.quality_label == "FLAC 16-bit/44.1kHz"
assert row.layout == "Stereo"
assert row.hires is True
assert row.cd is True
assert row.note == ""
def test_trackrow_defaults() -> None:
row = TrackRow(song=make_song(), quality_label="MP3 320", layout="Stereo", duration_str="2:00")
assert row.hires is False
assert row.cd is False
assert row.note == ""
def test_musicheaderinfo_constructs() -> None:
info = MusicHeaderInfo(
artist="Some Artist",
album="Some Album",
year="2021",
track_count=12,
quality_label="FLAC 24-bit/96kHz",
duration_str="42:10",
)
assert info.artist == "Some Artist"
assert info.album == "Some Album"
assert info.year == "2021"
assert info.track_count == 12
assert info.quality_label == "FLAC 24-bit/96kHz"
assert info.duration_str == "42:10"
assert info.artist_label == "Artist"
def test_format_track_detail_quality_brackets_codec() -> None:
assert format_track_detail_quality("FLAC 16-bit/44.1kHz") == "[FLAC] | 16-bit/44.1kHz"
assert format_track_detail_quality("ogg 320kbps") == "[OGG] | 320kbps"
def test_format_track_detail_quality_passthrough() -> None:
assert format_track_detail_quality("Lossless") == "Lossless"
assert format_track_detail_quality("") == ""
def test_render_track_panel_returns_panel() -> None:
rows = [
TrackRow(
song=make_song(track=1, name="First"),
quality_label="FLAC 16-bit/44.1kHz",
layout="Stereo",
duration_str="3:00",
cd=True,
),
TrackRow(
song=make_song(track=2, name="Second"),
quality_label="OGG 320",
layout="Stereo",
duration_str="4:00",
hires=True,
),
]
panel = render_track_panel(rows, total=len(rows))
assert isinstance(panel, Panel)
text = render_to_text(panel)
assert "First" in text
assert "Second" in text
assert "2 Tracks" in text
assert "CD" in text
assert "Hi-Res" in text
def test_render_track_panel_singular_label() -> None:
rows = [
TrackRow(song=make_song(name="Only"), quality_label="FLAC 16-bit/44.1kHz", layout="Stereo", duration_str="3:00")
]
text = render_to_text(render_track_panel(rows, total=1))
# The count node reads "1 Track" (singular); the panel title is always
# "Available Tracks", so we only check the count node phrasing here.
assert "1 Track" in text
assert "1 Tracks" not in text
def test_render_track_panel_note_marks_unavailable() -> None:
rows = [
TrackRow(
song=make_song(name="Gone"),
quality_label="FLAC",
layout="Stereo",
duration_str="3:00",
cd=True,
hires=True,
note="(unavailable in region)",
)
]
text = render_to_text(render_track_panel(rows, total=1))
assert "unavailable in region" in text
# When a note is present, the badges are suppressed.
assert "CD" not in text
assert "Hi-Res" not in text
def test_render_album_header_without_artwork() -> None:
info = MusicHeaderInfo(
artist="The Artist",
album="Greatest Hits",
year="1999",
track_count=10,
quality_label="FLAC 16-bit/44.1kHz",
duration_str="55:00",
)
header = render_album_header(info)
assert header is not None
text = render_to_text(header)
assert "The Artist" in text
assert "Greatest Hits" in text
assert "1999" in text
assert "10" in text
def test_render_album_header_with_artwork() -> None:
info = MusicHeaderInfo(
artist="Owner Name",
album="A Playlist",
year="2024",
track_count=3,
quality_label="OGG 320",
artist_label="Owner",
)
artwork = render_album_header(MusicHeaderInfo("x", "y", "1", 1, "q")) # any renderable
header = render_album_header(info, artwork=artwork)
assert header is not None
text = render_to_text(header)
assert "Owner Name" in text
assert "A Playlist" in text
assert "Owner" in text
class RaisingSession:
"""Session stub whose get() always raises, to drive the soft-fail path."""
def get(self, *args: Any, **kwargs: Any) -> Any:
raise RuntimeError("network disabled in tests")
def test_render_artwork_preview_empty_url_returns_none() -> None:
assert render_artwork_preview(RaisingSession(), "") is None
def test_render_artwork_preview_soft_fails_without_network() -> None:
# Either Pillow is absent (guard returns None) or the session.get raises
# (caught and returns None). Both paths must yield None, never raise.
result: Optional[RenderableType] = render_artwork_preview(RaisingSession(), "https://example.invalid/cover.jpg")
assert result is None
-14
View File
@@ -9,7 +9,6 @@ import pytest
from unshackle.core.music.extract import (
build_music_from_songs,
classify_release_kind,
dedupe_track_options,
duration_seconds,
first_number,
first_text,
@@ -17,7 +16,6 @@ from unshackle.core.music.extract import (
format_names,
year_from_value,
)
from unshackle.core.music.models import MusicTrackOption
from unshackle.core.titles.music import Music, Song
@@ -171,18 +169,6 @@ def test_classify_release_kind(raw_kind: str, count: Optional[float], expected:
assert classify_release_kind(raw_kind, count) == expected
def test_dedupe_track_options() -> None:
a = MusicTrackOption(codec="flac", bit_depth=16, sample_rate=44100, bitrate=None, quality_label="L", explicit=False)
a_dup = MusicTrackOption(
codec="FLAC", bit_depth=16, sample_rate=44100, bitrate=None, quality_label="L", explicit=False
) # codec case-insensitive duplicate of `a`
b = MusicTrackOption(codec="flac", bit_depth=24, sample_rate=96000, bitrate=None, quality_label="H", explicit=False)
c = MusicTrackOption(codec="flac", bit_depth=16, sample_rate=44100, bitrate=None, quality_label="L", explicit=True)
result = dedupe_track_options([a, a_dup, b, c])
assert result == [a, b, c]
def test_build_music_from_songs() -> None:
songs = [
make_song(
+40 -6
View File
@@ -26,13 +26,13 @@ def make_song(**overrides) -> Song:
kwargs = dict(
id_="track-0001",
service=DummyService,
name="NUEVAYoL",
artist="Bad Bunny",
album="DeBI TiRAR MaS FOToS",
name="Example Track",
artist="Example Artist",
album="Example Album",
track=1,
disc=1,
year=2025,
album_artist="Bad Bunny",
album_artist="Example Artist",
)
kwargs.update(overrides)
return Song(**kwargs)
@@ -52,13 +52,13 @@ def reset_folder_config():
def test_folder_fallback_when_no_templates(reset_folder_config):
song = make_song()
result = song.get_filename(StubMediaInfo(), folder=True)
assert result == sanitize_filename("Bad Bunny - DeBI TiRAR MaS FOToS (2025)", " ")
assert result == sanitize_filename("Example Artist - Example Album (2025)", " ")
def test_albums_template_used(reset_folder_config):
reset_folder_config.folder_templates = {"albums": "{album_artist} - {album} ({year})"}
result = make_song().get_filename(StubMediaInfo(), folder=True)
assert result == sanitize_filename("Bad Bunny - DeBI TiRAR MaS FOToS (2025)", " ")
assert result == sanitize_filename("Example Artist - Example Album (2025)", " ")
# Album folder must NOT carry per-track info like the song file name does.
assert "01" not in result
@@ -91,3 +91,37 @@ def test_validation_warns_on_unknown_folder_kind():
with pytest.warns(UserWarning, match="Unknown folder template kind"):
# A non-folder key is required so output-template validation actually runs.
Config(output_template={"songs": "{track_number}. {title}", "folder": {"bogus": "{album}"}})
def test_undated_single_degrades_cleanly(reset_folder_config):
"""A single with no release year and no disc must not leave empty parens behind."""
song = Song(
id_="track-0001",
service=DummyService,
name="Example Track",
artist="Example Artist",
album="Example Album",
track=1,
album_artist="Example Artist",
) # no disc, no year: both are optional
assert str(song) == "Example Artist - Example Album (?) / 01. Example Track"
assert song.get_filename(StubMediaInfo(), folder=True) == sanitize_filename("Example Artist - Example Album", " ")
assert song.get_filename(StubMediaInfo()) == "01.Example.Track" # spacer applies to the title too
context = song.build_template_context(StubMediaInfo())
assert context["year"] == ""
assert context["disc"] == "" # a lone disc is never numbered
assert context["track_number"] == "01"
def test_undated_album_template_omits_the_year_group(reset_folder_config):
reset_folder_config.folder_templates = {"albums": "{album_artist} - {album} ({year?})"}
result = make_song(year=None).get_filename(StubMediaInfo(), folder=True)
assert result == "Example Artist - Example Album"
def test_songs_separator_survives_a_config_without_a_songs_template(reset_folder_config):
"""A user config with output_template but no `songs` key used to raise a KeyError."""
reset_folder_config.output_template = {"movies": "{title}.{year}"}
assert reset_folder_config.get_template_separator("songs") == "."
+198
View File
@@ -0,0 +1,198 @@
"""Round-trip ``write_music_metadata`` against real audio files.
The dispatch test pins which tagger runs; this pins what it writes. Silent audio is
generated with ffmpeg, tagged from a fully populated Song, and read back with mutagen.
Cover art goes through the same path with a fake session, so no network is touched.
"""
from __future__ import annotations
import shutil
import subprocess
from pathlib import Path
from typing import Any
import pytest
from mutagen.flac import FLAC
from mutagen.mp3 import MP3
from mutagen.mp4 import MP4
from mutagen.oggvorbis import OggVorbis
from unshackle.core.music.tagger import write_music_metadata
from unshackle.core.titles.music import Song
pytestmark = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="requires ffmpeg")
PNG_STUB = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16 # magic only; Pillow dimension probing soft-fails
class DummyService:
"""Stand-in service class; a Title only requires a type, never an instance."""
class FakeResponse:
def __init__(self, content: bytes) -> None:
self.content = content
self.headers = {"Content-Type": "image/png"}
def raise_for_status(self) -> None:
pass
def __enter__(self) -> "FakeResponse":
return self
def __exit__(self, *exc_info: Any) -> None:
pass
class FakeSession:
def __init__(self, content: bytes) -> None:
self.content = content
self.requested: list[str] = []
def get(self, url: str, timeout: int = 0) -> FakeResponse:
self.requested.append(url)
return FakeResponse(self.content)
@pytest.fixture
def flac_path(tmp_path: Path) -> Path:
path = tmp_path / "04. Example Track.flac"
subprocess.run(
["ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "anullsrc=r=44100:cl=mono", "-t", "0.2", str(path)],
check=True,
)
return path
def make_song() -> Song:
return Song(
id_="track-0001",
service=DummyService,
name="Example Track",
artist="Example Artist",
album="Example Album, Live & Remastered",
track=4,
year=1983,
album_artist="Example Artist",
total_tracks=8,
genre="Synth-pop",
isrc="ZZABC2400001",
artwork_url="https://cdn.example/cover.png",
)
def test_flac_tags_read_back(flac_path: Path) -> None:
result = write_music_metadata(flac_path, make_song())
assert result.written
assert not result.skipped
assert not result.artwork_embedded, "no session was given, so no cover may be embedded"
audio = FLAC(flac_path)
assert audio["TITLE"] == ["Example Track"]
assert audio["ARTIST"] == ["Example Artist"]
assert audio["ALBUM"] == ["Example Album, Live & Remastered"]
assert audio["ALBUMARTIST"] == ["Example Artist"]
assert audio["TRACKNUMBER"] == ["4/8"]
assert audio["TRACKTOTAL"] == ["8"]
assert audio["DATE"] == ["1983"]
assert audio["GENRE"] == ["Synth-pop"]
assert audio["ISRC"] == ["ZZABC2400001"]
def test_lyrics_land_in_the_container_field_players_read(tmp_path: Path) -> None:
"""Each container has one lyrics field. A generic custom tag is not read as lyrics.
Both the MP3 and MP4 writers fall back to writing an unmapped key as a custom tag
(TXXX / freeform atom), so a lyrics field that is not mapped explicitly is written
but never displayed.
"""
lyrics = "Line one\nLine two"
song = Song(
id_="track-0001",
service=DummyService,
name="Example Track",
artist="Example Artist",
album="Example Album",
track=1,
lyrics=lyrics,
)
for suffix, encoder in ((".flac", "flac"), (".ogg", "libvorbis"), (".mp3", "libmp3lame"), (".m4a", "aac")):
path = tmp_path / f"track{suffix}"
subprocess.run(
[
"ffmpeg",
"-nostdin",
"-v",
"error",
"-f",
"lavfi",
"-i",
"anullsrc=r=44100:cl=mono",
"-t",
"0.2",
"-c:a",
encoder,
str(path),
],
check=True,
)
assert write_music_metadata(path, song).written
if suffix == ".flac":
assert FLAC(path)["LYRICS"] == [lyrics]
elif suffix == ".ogg":
assert OggVorbis(path)["LYRICS"] == [lyrics]
elif suffix == ".mp3":
tags = MP3(path).tags
assert tags.getall("USLT")[0].text == lyrics
assert not tags.getall("TXXX:LYRICS")
else:
audio = MP4(path)
assert audio["\xa9lyr"] == [lyrics]
assert not [key for key in audio if "LYRICS" in key]
def test_a_song_without_lyrics_writes_no_lyrics_field(flac_path: Path) -> None:
write_music_metadata(flac_path, make_song())
assert "LYRICS" not in FLAC(flac_path)
def test_cover_art_is_fetched_with_the_session_and_embedded(flac_path: Path) -> None:
session = FakeSession(PNG_STUB)
result = write_music_metadata(flac_path, make_song(), session=session)
assert result.artwork_embedded
assert session.requested == ["https://cdn.example/cover.png"]
audio = FLAC(flac_path)
assert len(audio.pictures) == 1
assert audio.pictures[0].mime == "image/png"
assert audio.pictures[0].data == PNG_STUB
def test_a_nested_metadata_dict_reaches_the_tags(flac_path: Path) -> None:
"""The renderer reads a nested "metadata" sub-dict, so the tagger must read it too.
A service that nests its metadata used to lose every tag sourced from song.data,
silently and with no log line.
"""
song = Song(
id_="track-0001",
service=DummyService,
name="Example Track",
artist="Example Artist",
album="Example Album",
track=1,
data={"metadata": {"composer": "Example Composer", "release_date": "2020-01-02"}},
)
write_music_metadata(flac_path, song)
audio = FLAC(flac_path)
assert audio["COMPOSER"] == ["Example Composer"]
assert audio["RELEASEDATE"] == ["2020-01-02"]
assert audio["DATE"] == ["2020-01-02"], "a full release date must beat the bare year"
+126
View File
@@ -0,0 +1,126 @@
"""`-w`/`--wanted` track selection for music.
A song shares the episode key space, its disc reading as the season and its track as the
episode. These check the whole path a user takes: the tokens they type, the keys the
click type makes of them, and the titles the download loop keeps.
"""
from __future__ import annotations
import click
import pytest
from unshackle.commands.dl import title_wanted
from unshackle.core.titles.movie import Movie
from unshackle.core.titles.music import Song
from unshackle.core.utils.click_types import SeasonRange
class DummyService:
pass
def make_song(track: int = 1, disc: int = 1) -> Song:
return Song(
id_=f"track-{disc:02}{track:04}",
service=DummyService,
name=f"Example Track {track}",
artist="Example Artist",
album="Example Album",
track=track,
disc=disc,
year=2025,
album_artist="Example Artist",
)
def make_movie() -> Movie:
return Movie(id_="movie-0001", service=DummyService, name="Example Movie", year=2025)
def select(tokens: str, songs: list[Song]) -> list[int]:
"""Track numbers kept, given what a user types after -w."""
wanted = SeasonRange().convert(tokens)
return [song.track for song in songs if title_wanted(song, wanted)]
def single_disc(count: int = 8) -> list[Song]:
return [make_song(track=n) for n in range(1, count + 1)]
def multi_disc(per_disc: int = 4) -> list[Song]:
return [make_song(track=n, disc=d) for d in (1, 2) for n in range(1, per_disc + 1)]
def test_a_bare_number_takes_one_track():
assert select("3", single_disc()) == [3]
def test_a_range_takes_the_tracks_inside_it():
assert select("2-5", single_disc()) == [2, 3, 4, 5]
def test_a_comma_list_takes_each_track():
assert select("1,3,7", single_disc()) == [1, 3, 7]
def test_a_leading_minus_removes_a_track_from_a_range():
assert select("1-5,-3", single_disc()) == [1, 2, 4, 5]
def test_the_disc_form_takes_one_track_of_that_disc():
songs = multi_disc()
kept = [(s.disc, s.track) for s in songs if title_wanted(s, SeasonRange().convert("2x3"))]
assert kept == [(2, 3)]
def test_the_disc_form_ranges_inside_one_disc():
songs = multi_disc()
kept = [(s.disc, s.track) for s in songs if title_wanted(s, SeasonRange().convert("2x1-2x3"))]
assert kept == [(2, 1), (2, 2), (2, 3)]
def test_a_bare_number_on_a_multi_disc_release_is_disc_one():
"""An omitted disc is disc 1, the number the tracklist shows for a one-disc release."""
songs = multi_disc()
kept = [(s.disc, s.track) for s in songs if title_wanted(s, SeasonRange().convert("1-2"))]
assert kept == [(1, 1), (1, 2)]
def test_discs_mix_in_one_selection():
songs = multi_disc()
kept = [(s.disc, s.track) for s in songs if title_wanted(s, SeasonRange().convert("1,2x4"))]
assert kept == [(1, 1), (2, 4)]
def test_a_selection_that_matches_nothing_keeps_no_track():
assert select("40-45", single_disc()) == []
def test_the_whole_release_downloads_without_a_selection():
songs = single_disc()
assert [s.track for s in songs if title_wanted(s, [])] == [1, 2, 3, 4, 5, 6, 7, 8]
def test_a_movie_stays_selected_whatever_the_tokens_are():
"""A movie has no selection key, so -w must not remove it."""
assert title_wanted(make_movie(), SeasonRange().convert("1-5")) is True
def test_a_range_across_discs_takes_the_rest_of_the_first_disc():
"""Same rule as a season range: the left disc runs to its end before the right one starts."""
songs = multi_disc()
kept = [(s.disc, s.track) for s in songs if title_wanted(s, SeasonRange().convert("1x3-2x2"))]
assert kept == [(1, 3), (1, 4), (2, 1), (2, 2)]
@pytest.mark.parametrize("token", ["1x", "x1", "1x2x3", "1.2"])
def test_bad_music_tokens_fail(token):
with pytest.raises(click.UsageError):
SeasonRange().parse_tokens(token)
def test_song_matches_only_its_own_key():
song = make_song(track=3, disc=2)
assert song.matches_wanted(["2x3"]) is True
assert song.matches_wanted(["1x3", "2x4"]) is False
+120
View File
@@ -0,0 +1,120 @@
"""Which tagger ``tag_file`` picks per title type.
A Song is an audio file, so it must reach mutagen and never mkvpropedit. The bug this
pins ran ``mkvpropedit`` against a ``.flac``. Movies and Episodes keep the Matroska path.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import pytest
from unshackle.core.config import config
from unshackle.core.music.tagger import MusicMetadataResult
from unshackle.core.titles.episode import Episode
from unshackle.core.titles.movie import Movie
from unshackle.core.titles.music import Song
from unshackle.core.utils import tags as tags_module
pytestmark = pytest.mark.unit
SESSION = object() # sentinel: tag_file must hand this straight to the music tagger
class DummyService:
"""Stand-in service class; a Title only requires a type, never an instance."""
class Calls:
"""Records which tagger ran, and with what."""
def __init__(self) -> None:
self.music: list[tuple[Path, Any, Any]] = []
self.mkv: list[tuple[Path, dict[str, str]]] = []
@pytest.fixture
def calls(monkeypatch: pytest.MonkeyPatch) -> Calls:
recorded = Calls()
def fake_write_music_metadata(path: Path, song: Any, *, session: Any = None, source_md5: str = "") -> Any:
recorded.music.append((path, song, session))
return MusicMetadataResult(written=True)
def fake_apply_tags(path: Path, tags: dict[str, str]) -> None:
recorded.mkv.append((path, tags))
monkeypatch.setattr(tags_module, "write_music_metadata", fake_write_music_metadata)
monkeypatch.setattr(tags_module, "apply_tags", fake_apply_tags)
monkeypatch.setattr(config, "tag_imdb_tmdb", False) # keep provider lookups out of a unit test
return recorded
def make_song(**overrides: Any) -> Song:
kwargs: dict[str, Any] = dict(
id_="track-0001",
service=DummyService,
name="Example Track",
artist="Example Artist",
album="Example Album, Live & Remastered",
track=4,
year=1983,
)
kwargs.update(overrides)
return Song(**kwargs)
def test_a_song_is_tagged_by_the_music_tagger(calls: Calls) -> None:
path = Path("04. Example Track.flac")
song = make_song()
tags_module.tag_file(path, song, session=SESSION)
assert calls.music == [(path, song, SESSION)], "the music tagger needs the session for cover art"
assert calls.mkv == [], "mkvpropedit must never touch an audio container"
def test_a_skipped_song_warns_with_the_reason(
calls: Calls, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(
tags_module,
"write_music_metadata",
lambda path, song, **kwargs: MusicMetadataResult(skipped=True, reason="install mutagen to write FLAC tags"),
)
with caplog.at_level("WARNING", logger="TAGS"):
tags_module.tag_file(Path("04. Example Track.flac"), make_song(), session=SESSION)
assert "install mutagen to write FLAC tags" in caplog.text
def test_a_tagger_fault_warns_and_never_raises(
calls: Calls, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""The file already moved when tag_file runs, so a corrupt file must not abort the run."""
def broken_tagger(path: Path, song: Any, **kwargs: Any) -> Any:
raise OSError("not a FLAC stream")
monkeypatch.setattr(tags_module, "write_music_metadata", broken_tagger)
with caplog.at_level("WARNING", logger="TAGS"):
tags_module.tag_file(Path("04. Example Track.flac"), make_song(), session=SESSION)
assert "not a FLAC stream" in caplog.text
@pytest.mark.parametrize(
"title",
[
Movie(id_="movie-0001", service=DummyService, name="Some Movie", year=2024),
Episode(id_="episode-0001", service=DummyService, title="Some Show", season=1, number=2, name="Pilot"),
],
ids=["movie", "episode"],
)
def test_a_movie_or_episode_still_takes_the_mkv_path(calls: Calls, title: Any) -> None:
path = Path("Some Title.mkv")
tags_module.tag_file(path, title)
assert calls.music == [], "write_music_metadata must not see a Matroska file"
assert [call[0] for call in calls.mkv] == [path]
+2 -3
View File
@@ -13,8 +13,7 @@ import pytest
from unshackle.core.config import config
from unshackle.core.titles.episode import Episode
from unshackle.core.titles.movie import Movie
from unshackle.core.titles.music import Song as MusicSong
from unshackle.core.titles.song import Song
from unshackle.core.titles.music import Song
from unshackle.core.utilities import sanitize_filename
from unshackle.core.utils.template_formatter import TemplateFormatter, detect_spacer
@@ -183,7 +182,7 @@ def make_album_song(**overrides):
album_artist="Some Artist",
)
kwargs.update(overrides)
return MusicSong(**kwargs)
return Song(**kwargs)
def assert_safe_path(result: str):
+121 -88
View File
@@ -1,9 +1,10 @@
"""Music post-script hooks, which no live service was available to test against.
Music returns from ``dl.result`` before the title loop, so its hooks are dispatched inside
the music pipeline. These check the pieces that pipeline feeds them: that a Song builds a
usable context, that an Album groups the way a season does, and that the album and run
contexts blank the file-level variables.
A Song runs through the same title loop as an Episode, so its hooks come from the same
two functions: ``build_context`` for a file and ``season_context`` for the release folder.
These check what those two feed a script: that a Song builds a usable context, that an
album groups the way a season does, and that the album hook describes the album rather
than whichever track landed last.
"""
from __future__ import annotations
@@ -13,7 +14,7 @@ from types import SimpleNamespace
import pytest
from unshackle.core.utils.post_scripts import build_context, dispatch, substitute, tokenize
from unshackle.core.utils.post_scripts import build_context, dispatch, season_context, substitute, tokenize
class FakeSong(SimpleNamespace):
@@ -33,26 +34,43 @@ class FakeSong(SimpleNamespace):
def song(**kwargs):
defaults = dict(
name="Blue Monday", artist="New Order", album="Power, Corruption & Lies", track=4, year=1983, id="s1"
name="Example Track",
artist="Example Artist",
album="Example Album, Live & Remastered",
track=4,
year=1983,
id="s1",
)
return FakeSong(**{**defaults, "id": "s1", **kwargs})
class FakeEpisode(SimpleNamespace):
"""An Episode as the hook sees it: a show title, a season, and a number."""
def build_template_context(self, media_info, show_service=True):
return {"title": self.title, "season": f"{self.season:02}", "audio": "EAC3", "video": "H.265"}
def episode(**kwargs):
defaults = dict(title="Some Show", season=1, number=5, name="Pilot", year=2024, id="e1")
return FakeEpisode(**{**defaults, **kwargs})
def test_song_context_has_music_fields_and_blank_episode_fields():
context = build_context(
song(),
media_info=object(),
filepath=Path("/dl/New Order/04. Blue Monday.flac"),
filepath=Path("/dl/Example Artist/04. Example Track.flac"),
service="EXAMPLE",
)
assert context["title"] == "Blue Monday"
assert context["artist"] == "New Order"
assert context["album"] == "Power, Corruption & Lies"
assert context["title"] == "Example Track"
assert context["artist"] == "Example Artist"
assert context["album"] == "Example Album, Live & Remastered"
assert context["track_number"] == "04"
assert context["service"] == "EXAMPLE"
assert context["filename"] == "04. Blue Monday.flac"
assert context["filename"] == "04. Example Track.flac"
assert context["ext"] == ".flac"
assert context["folder"] == str(Path("/dl/New Order"))
assert context["folder"] == str(Path("/dl/Example Artist"))
assert context["season"] == ""
assert context["episode"] == ""
@@ -60,24 +78,33 @@ def test_song_context_has_music_fields_and_blank_episode_fields():
def test_song_context_survives_without_media_info():
"""The failure path and any pre-mux dispatch pass no MediaInfo."""
context = build_context(song(), None, service="EXAMPLE", error="HTTPError: 403")
assert context["title"] == "Blue Monday"
assert context["title"] == "Example Track"
assert context["error"] == "HTTPError: 403"
assert context["filepath"] == ""
def test_album_grouping_matches_the_season_grouping_rule():
"""post_script_group's music branch: albums group by (album artist, album)."""
from unshackle.commands.dl import dl # noqa: F401 - import guard, the grouping is inlined in result()
from unshackle.commands.dl import post_script_group
from unshackle.core.titles.music import Song
def group(candidate):
artist = getattr(candidate, "album_artist", None) or getattr(candidate, "artist", "")
return ("album", artist, getattr(candidate, "album", ""))
def real_song(**kwargs):
defaults = dict(
id_="track-0001",
service=type("DummyService", (), {}),
name="Example Track",
artist="Example Artist",
album="Example Album, Live & Remastered",
track=4,
)
return Song(**{**defaults, **kwargs})
a = song(track=1)
b = song(track=2)
c = song(album="Technique")
assert group(a) == group(b)
assert group(a) != group(c)
a = real_song(track=1, id_="track-0002")
b = real_song(track=2, id_="track-0003")
c = real_song(album="Technique", id_="track-0004")
assert post_script_group(a) == post_script_group(b)
assert post_script_group(a) != post_script_group(c)
assert post_script_group(a)[0] == "album"
@pytest.mark.parametrize("mode", ["season", "run"])
@@ -86,25 +113,24 @@ def test_album_and_run_contexts_blank_the_file_variables(mode):
base = build_context(
song(),
media_info=object(),
filepath=Path("/dl/New Order/04. Blue Monday.flac"),
filepath=Path("/dl/Example Artist/04. Example Track.flac"),
service="EXAMPLE",
)
context = dict(base)
folder = Path("/dl/Example Artist")
if mode == "season":
for key in ("filepath", "filename", "ext", "sidecars", "episode", "episode_name"):
context[key] = ""
context = season_context(base, folder)
else:
context = dict.fromkeys(base, "")
context["folder"] = "/dl/New Order"
context["folder"] = str(folder)
assert context["filepath"] == ""
assert context["sidecars"] == ""
assert context["folder"] == "/dl/New Order"
assert context["folder"] == str(folder)
if mode == "run":
assert context["title"] == ""
assert set(context) == set(base)
else:
assert context["album"] == "Power, Corruption & Lies"
assert context["album"] == "Example Album, Live & Remastered"
def test_a_music_hook_command_builds_the_argv_a_script_would_get(monkeypatch, tmp_path):
@@ -126,7 +152,7 @@ def test_a_music_hook_command_builds_the_argv_a_script_would_get(monkeypatch, tm
context = build_context(
song(),
media_info=object(),
filepath=Path("/dl/New Order/04. Blue Monday.flac"),
filepath=Path("/dl/Example Artist/04. Example Track.flac"),
service="EXAMPLE",
)
dispatch("success", "file", context)
@@ -139,93 +165,100 @@ def test_a_music_hook_command_builds_the_argv_a_script_would_get(monkeypatch, tm
time.sleep(0.05)
assert marker.exists(), "the hook process never ran"
argv = marker.read_text().splitlines()
assert "--album=Power, Corruption & Lies" in argv
assert "--artist=New Order" in argv
assert "--album=Example Album, Live & Remastered" in argv
assert "--artist=Example Artist" in argv
assert "--track=04" in argv
assert f"--file={Path('/dl/New Order/04. Blue Monday.flac')}" in argv
assert f"--file={Path('/dl/Example Artist/04. Example Track.flac')}" in argv
def test_album_hook_mapping_is_bound_before_any_early_branch():
"""--skip-dl skips the final-move block; the album hook still runs after it.
def test_season_context_blanks_the_file_variables_and_sets_the_folder():
"""An album hook describes the release folder, so no track's file may leak into it."""
folder = Path("/dl/Example Artist - Example Album, Live & Remastered (1983)")
context = build_context(
song(),
media_info=object(),
filepath=folder / "04. Example Track.flac",
sidecars=[folder / "04. Example Track.lrc"],
service="EXAMPLE",
)
Guards a NameError: the mapping used to be created inside that block, so a music
download with --skip-dl reached the album hook with an unbound name.
"""
import inspect
album = season_context(context, folder)
from unshackle.commands.dl import dl
source = inspect.getsource(dl.result)
body = source[source.index("def download_music_title") :]
bind = body.index("album_post_script: dict")
for use in ("album_post_script[final_path.parent]", "album_post_script.items()"):
assert bind < body.index(use), f"{use} can run before the mapping is bound"
line = body[:bind].rsplit("\n", 1)[-1]
assert len(line) == 16, f"bound at indent {len(line)}, expected the function body level"
for key in ("filepath", "filename", "ext", "sidecars", "episode", "episode_name"):
assert album[key] == "", f"{key} still describes a file"
assert album["folder"] == str(folder)
assert album["service"] == "EXAMPLE"
assert context["filepath"] != "", "the file context handed in must not be mutated"
def test_album_hook_identifies_the_album_not_the_last_track():
"""A Song's ``{title}`` is the song, so blanking only the video fields is not enough.
def test_season_context_identifies_the_album_not_the_last_track():
"""A Song's ``{title}`` is the song, so blanking only the file fields is not enough.
Guards a regression where an album hook was handed whichever track landed last:
``{title}`` was a song name and ``{track_number}``/``{isrc}`` were that track's.
"""
from unshackle.commands.dl import dl
folder = Path("/dl/Example Artist - Example Album (1983)")
context = build_context(
song(track=11, isrc="GBAAA8300001"),
song(track=11),
media_info=object(),
filepath=Path("/dl/New Order - PCL (1983)/11. Blue Monday.flac"),
filepath=folder / "11. Example Track.flac",
service="EXAMPLE",
)
context.setdefault("isrc", "GBAAA8300001")
context["isrc"] = "ZZABC2400001"
context["disc"] = "02"
album = dict(context)
for key in ("filepath", "filename", "ext", "sidecars", "episode", "episode_name"):
album[key] = ""
for key in ("track_number", "disc", "isrc"):
if key in album:
album[key] = ""
album["title"] = album["title_raw"] = album.get("album", "")
album = season_context(context, folder)
assert album["title"] == "Power, Corruption & Lies"
assert album["title_raw"] == "Power, Corruption & Lies"
assert album["title"] == "Example Album, Live & Remastered"
assert album["title_raw"] == "Example Album, Live & Remastered"
assert album["track_number"] == ""
assert album["disc"] == ""
assert album["isrc"] == ""
assert album["artist"] == "New Order" # album-level, kept
import inspect
body = inspect.getsource(dl.result)
body = body[body.index("for folder, context in album_post_script.items()") :][:900]
for key in ("track_number", "disc", "isrc"):
assert f'"{key}"' in body, f"the album hook no longer blanks {key}"
assert 'album_context["title"]' in body
assert album["artist"] == "Example Artist" # album-level, kept
assert album["year"] == "1983" # album-level, kept
def test_a_music_download_failure_dispatches_a_failure_hook():
"""Parity with the video worker, which dispatches failure/file when it raises."""
import inspect
def test_season_context_leaves_a_non_music_context_untouched():
"""The music branch is gated on ``{album}``; a season hook must not lose its own keys."""
folder = Path("/dl/Some Show/Season 1")
context = build_context(
episode(),
media_info=object(),
filepath=folder / "Some Show S01E05 Pilot.mkv",
service="EXAMPLE",
)
context["disc"] = "sentinel" # only the music branch clears this
from unshackle.commands.dl import dl
season = season_context(context, folder)
body = inspect.getsource(dl.result)
body = body[body.index("def download_music_title") :]
assert body.count("music_failure(") == 3, "expected one definition and both failure exits"
assert body.index("def music_failure") < body.index("music_failure(e)")
assert "music_failure(error)" in body
assert season["title"] == "Some Show"
assert season["title_raw"] == "Some Show"
assert season["season"] == "1"
assert season["disc"] == "sentinel"
assert season["episode"] == ""
assert season["episode_name"] == ""
assert season["folder"] == str(folder)
release = SimpleNamespace(title="Power, Corruption & Lies", artist="New Order", year=1983)
context = build_context(release, service="EXAMPLE", error="HTTPError: 403")
assert context["title"] == "Power, Corruption & Lies"
def test_a_failed_track_gets_a_track_scoped_failure_context():
"""Failure is per-track, like every other title type, and does not describe the album.
The real dispatch passes no MediaInfo and no filepath, so the context is whatever the
Song itself carries, and it is never run through ``season_context``.
"""
context = build_context(song(track=11), None, service="EXAMPLE", error="HTTPError: 403")
assert context["title"] == "Example Track"
assert context["title_raw"] == "Example Track"
assert context["error"] == "HTTPError: 403"
assert context["filepath"] == ""
assert context["season"] == ""
def test_an_album_name_with_a_comma_stays_one_argument():
"""'Power, Corruption & Lies' must not split, which is the whole tokenize-first point."""
argv = substitute(tokenize("upload.py --album={album}"), {"album": "Power, Corruption & Lies"})
assert argv == ["upload.py", "--album=Power, Corruption & Lies"]
"""'Example Album, Live & Remastered' must not split, which is the whole tokenize-first point."""
argv = substitute(tokenize("upload.py --album={album}"), {"album": "Example Album, Live & Remastered"})
assert argv == ["upload.py", "--album=Example Album, Live & Remastered"]
if __name__ == "__main__":