mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-12 12:05:07 +00:00
241 lines
9.0 KiB
Python
241 lines
9.0 KiB
Python
"""Tests for the ``--export`` file name and the messages a run with ``--export`` logs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
import mediaexport
|
|
import pytest
|
|
|
|
from unshackle.commands.dl import dl
|
|
from unshackle.core.export_name import ExportNamer, export_name, move_export, season_episodes
|
|
from unshackle.core.titles import Episode, Movie, Song
|
|
from unshackle.core.tracks import Audio, Subtitle, Video
|
|
|
|
KID = UUID(hex="00000000000000000000000000000001")
|
|
|
|
|
|
class StubService:
|
|
"""Stands in for the service class slot on a title; never instantiated."""
|
|
|
|
|
|
class StubDRM:
|
|
def __init__(self) -> None:
|
|
self.content_keys = {KID: "aa" * 16}
|
|
|
|
def to_dict(self) -> dict:
|
|
return {"system": "Widevine", "pssh_b64": "AAAA"}
|
|
|
|
|
|
def video(
|
|
id_: str = "v1",
|
|
height: int = 1080,
|
|
width: int = 1920,
|
|
codec: Video.Codec = Video.Codec.AVC,
|
|
range_: Video.Range = Video.Range.SDR,
|
|
descriptor: Video.Descriptor = Video.Descriptor.DASH,
|
|
) -> Video:
|
|
return Video(
|
|
id_=id_,
|
|
url=f"https://example.test/{id_}.mp4",
|
|
language="en",
|
|
codec=codec,
|
|
range_=range_,
|
|
width=width,
|
|
height=height,
|
|
descriptor=descriptor,
|
|
)
|
|
|
|
|
|
def audio(
|
|
id_: str = "a1",
|
|
codec: Audio.Codec = Audio.Codec.AAC,
|
|
channels: str = "2.0",
|
|
descriptor: Audio.Descriptor = Audio.Descriptor.DASH,
|
|
joc: int = 0,
|
|
) -> Audio:
|
|
return Audio(
|
|
id_=id_,
|
|
url=f"https://example.test/{id_}.mp4",
|
|
language="en",
|
|
codec=codec,
|
|
channels=channels,
|
|
descriptor=descriptor,
|
|
joc=joc,
|
|
)
|
|
|
|
|
|
def episode(season: int, number: int, name: str = "Pilot") -> Episode:
|
|
return Episode(
|
|
id_=f"ep-{season}-{number}", service=StubService, title="Example Show", season=season, number=number, name=name
|
|
)
|
|
|
|
|
|
def movie(name: str = "Example Movie", year: int | None = 2024) -> Movie:
|
|
return Movie(id_="movie-1", service=StubService, name=name, year=year)
|
|
|
|
|
|
def seasons_of(*shape: tuple[int, int]) -> dict[int, set[int]]:
|
|
return season_episodes([episode(s, n) for s, count in shape for n in range(1, count + 1)])
|
|
|
|
|
|
def test_movie_name_has_year_and_no_scope() -> None:
|
|
name = export_name([movie()], [video(), audio(descriptor=Audio.Descriptor.HLS)], "EXAMPLE")
|
|
assert name == "Example Movie 2024_1080p_h264_sdr_dash_hls_aac2.0-EXAMPLE"
|
|
|
|
|
|
def test_movie_without_year() -> None:
|
|
assert export_name([movie(year=None)], [video()], "EXAMPLE") == "Example Movie_1080p_h264_sdr_dash-EXAMPLE"
|
|
|
|
|
|
def test_full_season_is_the_season_number() -> None:
|
|
titles = [episode(1, n) for n in range(1, 4)]
|
|
name = export_name(titles, [video(), audio(codec=Audio.Codec.AC3, channels="5.1")], "EXAMPLE", seasons_of((1, 3)))
|
|
assert name == "Example Show S01_1080p_h264_sdr_dash_dd5.1-EXAMPLE"
|
|
|
|
|
|
def test_contiguous_subset_is_one_range() -> None:
|
|
titles = [episode(1, n) for n in (5, 6, 7, 8, 9)]
|
|
assert export_name(titles, [], "EXAMPLE", seasons_of((1, 10))) == "Example Show S01E05-E09-EXAMPLE"
|
|
|
|
|
|
def test_non_contiguous_subset_lists_each_run() -> None:
|
|
titles = [episode(1, n) for n in (1, 2, 5, 9)]
|
|
assert export_name(titles, [], "EXAMPLE", seasons_of((1, 10))) == "Example Show S01E01-E02_S01E05_S01E09-EXAMPLE"
|
|
|
|
|
|
def test_many_runs_are_cut_short() -> None:
|
|
titles = [episode(1, n) for n in (1, 3, 5, 7, 9, 11)]
|
|
name = export_name(titles, [], "EXAMPLE", seasons_of((1, 12)))
|
|
assert name == "Example Show S01E01_S01E03_S01E05_+3-EXAMPLE"
|
|
|
|
|
|
def test_unknown_season_list_uses_ranges() -> None:
|
|
titles = [episode(1, n) for n in (1, 2, 3)]
|
|
assert export_name(titles, [], "EXAMPLE") == "Example Show S01E01-E03-EXAMPLE"
|
|
|
|
|
|
def test_whole_seasons_in_a_row_collapse() -> None:
|
|
titles = [episode(s, n) for s in (1, 2, 3) for n in (1, 2)]
|
|
assert export_name(titles, [], "EXAMPLE", seasons_of((1, 2), (2, 2), (3, 2))) == "Example Show S01-S03-EXAMPLE"
|
|
|
|
|
|
def test_whole_season_and_a_partial_season() -> None:
|
|
titles = [episode(1, 1), episode(1, 2), episode(2, 1)]
|
|
assert export_name(titles, [], "EXAMPLE", seasons_of((1, 2), (2, 4))) == "Example Show S01_S02E01-EXAMPLE"
|
|
|
|
|
|
def test_quality_parts_are_deduplicated_and_ordered() -> None:
|
|
tracks = [
|
|
video("v1", 1080, codec=Video.Codec.HEVC, range_=Video.Range.HDR10),
|
|
video("v2", 2160, 3840, codec=Video.Codec.HEVC, range_=Video.Range.SDR, descriptor=Video.Descriptor.ISM),
|
|
video("v3", 1080, codec=Video.Codec.AVC, range_=Video.Range.DV),
|
|
video("v4", 800, 1920, codec=Video.Codec.HEVC, range_=Video.Range.HDR10),
|
|
audio("a1", Audio.Codec.EC3, "5.1", joc=16),
|
|
audio("a2", Audio.Codec.AAC, "2.0"),
|
|
audio("a3", Audio.Codec.EC3, "5.1", joc=16),
|
|
audio("a4", Audio.Codec.EC3, "2.0"),
|
|
Subtitle(id_="s1", url="https://example.test/s1.vtt", language="en", codec=Subtitle.Codec.WebVTT),
|
|
]
|
|
name = export_name([movie()], tracks, "EXAMPLE")
|
|
assert name == "Example Movie 2024_2160p_1080p_h264_h265_sdr_hdr10_dv_dash_ism_aac2.0_ddp5.1atmos_ddp2.0-EXAMPLE"
|
|
|
|
|
|
def test_title_is_filename_safe_and_keeps_letters() -> None:
|
|
name = export_name([movie('Café: "Ünïcode"/Kanji 映画?', year=None)], [], "EXAMPLE")
|
|
assert name == "Café Ünïcode & Kanji 映画-EXAMPLE"
|
|
|
|
|
|
def test_song_uses_artist_and_album() -> None:
|
|
song = Song(
|
|
id_="track-1", service=StubService, name="Track", artist="Example Artist", album="Example Album", track=1
|
|
)
|
|
assert export_name([song], [audio(codec=Audio.Codec.FLAC)], "EXAMPLE") == (
|
|
"Example Artist - Example Album_dash_flac2.0-EXAMPLE"
|
|
)
|
|
|
|
|
|
def test_collision_gets_a_suffix(tmp_path: Path) -> None:
|
|
other = tmp_path / "Example Movie 2024-EXAMPLE.json"
|
|
other.write_text("{}")
|
|
current = tmp_path / "export_EXAMPLE_1.json"
|
|
current.write_text("mine")
|
|
|
|
moved = move_export(current, "Example Movie 2024-EXAMPLE")
|
|
|
|
assert moved.name == "Example Movie 2024-EXAMPLE-2.json"
|
|
assert other.read_text() == "{}"
|
|
assert moved.read_text() == "mine"
|
|
assert not current.exists()
|
|
# the run's own file keeps its suffix instead of moving on to -3
|
|
assert move_export(moved, "Example Movie 2024-EXAMPLE") == moved
|
|
|
|
|
|
def make_dl(tmp_path: Path, seasons: dict[int, set[int]] | None = None) -> dl:
|
|
instance = dl.__new__(dl)
|
|
instance.service = "EXAMPLE"
|
|
instance.log = logging.getLogger("download")
|
|
instance.export_namer = ExportNamer("EXAMPLE", seasons or {})
|
|
return instance
|
|
|
|
|
|
def test_export_is_renamed_as_titles_arrive_and_stays_valid(tmp_path: Path) -> None:
|
|
instance = make_dl(tmp_path, seasons_of((1, 2)))
|
|
work = tmp_path / "export_EXAMPLE_1.json"
|
|
first, second = episode(1, 1), episode(1, 2)
|
|
for title in (first, second):
|
|
title.tracks.add(video(f"v{title.number}"))
|
|
|
|
instance.write_export(work, first, first.tracks.videos[0], StubDRM())
|
|
partial = tmp_path / "Example Show S01E01_1080p_h264_sdr_dash-EXAMPLE.json"
|
|
assert [p.name for p in tmp_path.iterdir()] == [partial.name]
|
|
assert [e.id for e in mediaexport.read(partial).titles] == ["ep-1-1"]
|
|
|
|
instance.write_export(work, second, second.tracks.videos[0])
|
|
final = tmp_path / "Example Show S01_1080p_h264_sdr_dash-EXAMPLE.json"
|
|
assert [p.name for p in tmp_path.iterdir()] == [final.name]
|
|
assert [e.id for e in mediaexport.read(final).titles] == ["ep-1-1", "ep-1-2"]
|
|
|
|
|
|
def test_export_messages(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
instance = make_dl(tmp_path)
|
|
title = movie()
|
|
title.tracks.add(video())
|
|
title.tracks.add(audio())
|
|
title.tracks.add(audio("a2"))
|
|
work = tmp_path / "export_EXAMPLE_1.json"
|
|
|
|
with caplog.at_level(logging.INFO, logger="download"):
|
|
for track in title.tracks:
|
|
instance.write_export(work, title, track, StubDRM() if isinstance(track, Video) else None)
|
|
instance.log_export_title(title)
|
|
instance.log_export_summary()
|
|
|
|
lines = [r.getMessage() for r in caplog.records]
|
|
assert "Exporting to Example Movie 2024_1080p_h264_sdr_dash_aac2.0-EXAMPLE.json" in lines
|
|
assert "Exported 'Example Movie' (2024): 1 video, 2 audio, 0 subtitles, 1 key" in lines
|
|
final = tmp_path / "Example Movie 2024_1080p_h264_sdr_dash_aac2.0-EXAMPLE.json"
|
|
assert f"Saved the export to {final}: 1 title, 1 key" in lines
|
|
assert any("content keys" in line for line in lines)
|
|
|
|
|
|
def test_episode_message_names_the_episode(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
instance = make_dl(tmp_path)
|
|
title = episode(1, 5, "Example Episode")
|
|
title.tracks.add(video())
|
|
instance.write_export(tmp_path / "export_EXAMPLE_1.json", title, title.tracks.videos[0])
|
|
|
|
with caplog.at_level(logging.INFO, logger="download"):
|
|
instance.log_export_title(title)
|
|
|
|
assert "Exported S01E05 'Example Episode': 1 video, 0 audio, 0 subtitles, 0 keys" in caplog.messages
|
|
|
|
|
|
def test_summary_when_nothing_was_exported(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
instance = make_dl(tmp_path)
|
|
with caplog.at_level(logging.INFO, logger="download"):
|
|
instance.log_export_summary()
|
|
assert "Nothing was exported" in caplog.messages
|