mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-12 12:05:07 +00:00
imSp4rky
3ffc7a9735
Reject bad quality, bitrate/range conflicts and malformed ranges with a 400, convert quality in the worker, and pass cdm_only into ctx.params.
361 lines
11 KiB
Python
361 lines
11 KiB
Python
"""Unit tests for unshackle.core.api.handlers serializers + validators."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
|
|
import pytest
|
|
import requests
|
|
from langcodes import Language
|
|
|
|
from unshackle.core.api.errors import APIError, APIErrorCode
|
|
from unshackle.core.api.handlers import (
|
|
drm_preference_name,
|
|
listed_titles,
|
|
sanitize_log,
|
|
search_handler,
|
|
serialize_audio_track,
|
|
serialize_drm,
|
|
serialize_subtitle_track,
|
|
serialize_title,
|
|
serialize_video_track,
|
|
validate_download_parameters,
|
|
validate_service,
|
|
)
|
|
from unshackle.core.titles.episode import Episode
|
|
from unshackle.core.titles.movie import Movie
|
|
from unshackle.core.tracks import Audio, Subtitle, Video
|
|
from unshackle.core.tracks.track import Track
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class _FakeSvc:
|
|
pass
|
|
|
|
|
|
def video(**overrides) -> Video:
|
|
base = dict(
|
|
url="https://example.com/v.mpd",
|
|
language=Language.get("en"),
|
|
descriptor=Track.Descriptor.URL,
|
|
codec=Video.Codec.AVC,
|
|
range_=Video.Range.SDR,
|
|
bitrate=5_000_000,
|
|
width=1920,
|
|
height=1080,
|
|
fps=24,
|
|
id_="video-001",
|
|
)
|
|
base.update(overrides)
|
|
return Video(**base)
|
|
|
|
|
|
def audio(**overrides) -> Audio:
|
|
base = dict(
|
|
url="https://example.com/a.mpd",
|
|
language=Language.get("en"),
|
|
descriptor=Track.Descriptor.URL,
|
|
codec=Audio.Codec.AAC,
|
|
bitrate=128_000,
|
|
channels=2,
|
|
joc=0,
|
|
descriptive=False,
|
|
id_="audio-001",
|
|
)
|
|
base.update(overrides)
|
|
return Audio(**base)
|
|
|
|
|
|
def subtitle(**overrides) -> Subtitle:
|
|
base = dict(
|
|
url="https://example.com/s.vtt",
|
|
language=Language.get("en"),
|
|
descriptor=Track.Descriptor.URL,
|
|
codec=Subtitle.Codec.WebVTT,
|
|
cc=False,
|
|
sdh=False,
|
|
forced=False,
|
|
id_="sub-001",
|
|
)
|
|
base.update(overrides)
|
|
return Subtitle(**base)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
("hello\nworld", "helloworld"),
|
|
("a\r\nb\x00c", "abc"),
|
|
("clean", "clean"),
|
|
(12345, "12345"),
|
|
],
|
|
)
|
|
def test_sanitize_log(raw, expected: str) -> None:
|
|
assert sanitize_log(raw) == expected
|
|
|
|
|
|
def test_serialize_title_movie() -> None:
|
|
movie = Movie(id_="movie-0001", service=_FakeSvc, name="Title X", year=2024, language=Language.get("en"))
|
|
d = serialize_title(movie)
|
|
assert d["type"] == "movie"
|
|
assert d["name"] == "Title X"
|
|
assert d["year"] == 2024
|
|
assert d["id"] == "movie-0001"
|
|
assert d["language"] == "en"
|
|
|
|
|
|
def test_serialize_title_episode_named() -> None:
|
|
ep = Episode(
|
|
id_="ep-00001",
|
|
service=_FakeSvc,
|
|
title="My Show",
|
|
season=2,
|
|
number=3,
|
|
name="Pilot",
|
|
year=2024,
|
|
language=Language.get("en"),
|
|
)
|
|
d = serialize_title(ep)
|
|
assert d["type"] == "episode"
|
|
assert d["series_title"] == "My Show"
|
|
assert d["season"] == 2
|
|
assert d["number"] == 3
|
|
assert d["name"] == "Pilot"
|
|
|
|
|
|
def test_serialize_title_episode_unnamed_falls_back_to_number() -> None:
|
|
ep = Episode(
|
|
id_="ep-00002",
|
|
service=_FakeSvc,
|
|
title="Show",
|
|
season=1,
|
|
number=5,
|
|
name=None,
|
|
year=None,
|
|
language=Language.get("en"),
|
|
)
|
|
d = serialize_title(ep)
|
|
assert d["name"] == "Episode 05"
|
|
|
|
|
|
def test_no_titles_is_a_service_error() -> None:
|
|
with pytest.raises(APIError) as exc:
|
|
listed_titles(None)
|
|
assert exc.value.error_code == APIErrorCode.SERVICE_ERROR
|
|
|
|
|
|
def test_init_segment_reports_the_http_status() -> None:
|
|
refused = requests.Response()
|
|
refused.status_code = 403
|
|
refused.raw = io.BytesIO()
|
|
|
|
class FakeSession(requests.Session):
|
|
def request(self, *args: object, **kwargs: object) -> requests.Response:
|
|
return refused
|
|
|
|
track = Video(url="https://cdn.example/v.mp4", language=Language.get("en"))
|
|
with pytest.raises(ValueError, match="HTTP 403"):
|
|
track.get_init_segment(session=FakeSession())
|
|
|
|
|
|
def test_serialize_video_track_basic() -> None:
|
|
d = serialize_video_track(video())
|
|
assert d["id"] == "video-001"
|
|
assert d["codec"] == "AVC"
|
|
assert d["bitrate"] == 5000 # kbps
|
|
assert d["resolution"] == "1920x1080"
|
|
assert d["fps"] == 24
|
|
assert d["range"] == "SDR"
|
|
assert d["language"] == "en"
|
|
assert d["drm"] is None
|
|
assert "url" not in d
|
|
|
|
|
|
def test_serialize_video_track_include_url() -> None:
|
|
d = serialize_video_track(video(), include_url=True)
|
|
assert d["url"] == "https://example.com/v.mpd"
|
|
|
|
|
|
def test_serialize_audio_track_basic() -> None:
|
|
d = serialize_audio_track(audio())
|
|
assert d["id"] == "audio-001"
|
|
assert d["codec"] == "AAC"
|
|
assert d["bitrate"] == 128
|
|
assert d["channels"] == 2
|
|
assert d["descriptive"] is False
|
|
|
|
|
|
def test_serialize_video_track_drm_preference_round_trip() -> None:
|
|
d = serialize_video_track(video())
|
|
assert d["drm_preference"] is None
|
|
v = video()
|
|
v.drm_preference = "pr"
|
|
assert serialize_video_track(v)["drm_preference"] == "pr"
|
|
|
|
|
|
def test_serialize_audio_track_drm_preference() -> None:
|
|
a = audio()
|
|
a.drm_preference = "widevine"
|
|
assert serialize_audio_track(a)["drm_preference"] == "widevine"
|
|
|
|
|
|
def test_drm_preference_name_maps_short_forms() -> None:
|
|
v = video()
|
|
assert drm_preference_name(v) is None
|
|
v.drm_preference = "pr"
|
|
assert drm_preference_name(v) == "playready"
|
|
v.drm_preference = "wv"
|
|
assert drm_preference_name(v) == "widevine"
|
|
v.drm_preference = "widevine"
|
|
assert drm_preference_name(v) == "widevine"
|
|
|
|
|
|
def test_serialize_subtitle_track_basic() -> None:
|
|
d = serialize_subtitle_track(subtitle(forced=True))
|
|
assert d["id"] == "sub-001"
|
|
assert d["codec"] == "WebVTT"
|
|
assert d["forced"] is True
|
|
assert d["sdh"] is False
|
|
assert d["cc"] is False
|
|
|
|
|
|
def test_serialize_drm_none_returns_none() -> None:
|
|
assert serialize_drm(None) is None
|
|
assert serialize_drm([]) is None
|
|
|
|
|
|
def test_serialize_drm_widevine_minimal() -> None:
|
|
class _PSSH:
|
|
def dumps(self) -> str:
|
|
return "BASE64PSSH=="
|
|
|
|
class _Widevine:
|
|
def __init__(self) -> None:
|
|
self._pssh = _PSSH()
|
|
self.kids = ["00112233445566778899aabbccddeeff"]
|
|
self.license_url = "https://lic.example.com/wv"
|
|
|
|
out = serialize_drm(_Widevine())
|
|
assert isinstance(out, list)
|
|
assert len(out) == 1
|
|
info = out[0]
|
|
assert info["type"] == "_widevine" # class name lowercased
|
|
assert info["pssh"] == "BASE64PSSH=="
|
|
assert info["kids"] == ["00112233445566778899aabbccddeeff"]
|
|
assert info["license_url"] == "https://lic.example.com/wv"
|
|
|
|
|
|
def test_serialize_drm_playready_pssh_without_dumps_is_omitted() -> None:
|
|
# pyplayready's PSSH exposes no dumps()/to_base64()/__bytes__; serialization
|
|
# must omit the pssh field rather than emit an object repr.
|
|
class _PSSH:
|
|
pass
|
|
|
|
class _PlayReady:
|
|
def __init__(self) -> None:
|
|
self._pssh = _PSSH()
|
|
self.kids = ["00112233445566778899aabbccddeeff"]
|
|
|
|
info = serialize_drm(_PlayReady())[0]
|
|
assert "pssh" not in info
|
|
assert info["kids"] == ["00112233445566778899aabbccddeeff"]
|
|
|
|
|
|
def test_validate_service_unknown_returns_none() -> None:
|
|
assert validate_service("NOPE_THIS_IS_NOT_REAL_") is None
|
|
|
|
|
|
def test_validate_download_params_accepts_defaults() -> None:
|
|
assert validate_download_parameters({}) is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"data, fragment",
|
|
[
|
|
({"vcodec": "WUT"}, "Invalid vcodec"),
|
|
({"vcodec": 123}, "vcodec must be a string or list"),
|
|
({"acodec": "MP9"}, "Invalid acodec"),
|
|
({"sub_format": "doc"}, "Invalid sub_format"),
|
|
({"vbitrate": -1}, "vbitrate"),
|
|
({"abitrate": "no"}, "abitrate"),
|
|
({"vbitrate_range": "no-dash-but-letters"}, "MIN-MAX"),
|
|
({"vbitrate": 5000, "vbitrate_range": "1-2"}, "vbitrate and vbitrate_range"),
|
|
({"abitrate": 128, "abitrate_range": "96-256"}, "abitrate and abitrate_range"),
|
|
({"abitrate_range": "1-2-3"}, "MIN-MAX"),
|
|
({"quality": "4k"}, "Invalid quality"),
|
|
({"quality": [0]}, "above 0"),
|
|
({"channels": -3}, "channels"),
|
|
({"workers": 0}, "workers"),
|
|
({"downloads": 0}, "downloads"),
|
|
({"video_only": True, "audio_only": True}, "exclusive"),
|
|
({"no_subs": True, "subs_only": True}, "no_subs and subs_only"),
|
|
({"no_audio": True, "audio_only": True}, "no_audio and audio_only"),
|
|
({"require_subs": "en"}, "require_subs must be an array"),
|
|
({"range": "UHD"}, "Invalid range"),
|
|
({"range": ["SDR", "UHD"]}, "Invalid range value"),
|
|
],
|
|
)
|
|
def test_validate_download_params_errors(data: dict, fragment) -> None:
|
|
result = validate_download_parameters(data)
|
|
if fragment is None:
|
|
assert result is None
|
|
else:
|
|
assert result is not None
|
|
assert fragment in result
|
|
|
|
|
|
def test_validate_download_params_accepts_valid_values() -> None:
|
|
assert (
|
|
validate_download_parameters(
|
|
{
|
|
"vcodec": "H264,H265",
|
|
"acodec": ["AAC", "EAC3"],
|
|
"sub_format": "VTT",
|
|
"vbitrate": 6000,
|
|
"abitrate": 128,
|
|
"channels": 5.1,
|
|
"workers": 8,
|
|
"downloads": 2,
|
|
"range": ["SDR", "HDR10"],
|
|
}
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("vrange", ["6000-7000", "7000-6000", " 6000 - 7000 "])
|
|
def test_validate_download_params_accepts_valid_bitrate_ranges(vrange: str) -> None:
|
|
assert validate_download_parameters({"vbitrate_range": vrange, "abitrate_range": "96-192"}) is None
|
|
|
|
|
|
async def test_search_missing_service_raises_invalid_input() -> None:
|
|
# MISSING_SERVICE was never an APIErrorCode member, so this raised AttributeError
|
|
# and routes.py surfaced a 500 instead of the intended 400.
|
|
with pytest.raises(APIError) as exc:
|
|
await search_handler({"query": "dune"}, None)
|
|
assert exc.value.error_code == APIErrorCode.INVALID_INPUT
|
|
assert APIError.default_http_status(exc.value.error_code) == 400
|
|
|
|
|
|
async def test_search_unknown_service_raises_invalid_service() -> None:
|
|
# get_tag echoes unknown tags back, so the old `not normalized_service` guard
|
|
# never fired and Services.get_path leaked a KeyError as a 500.
|
|
with pytest.raises(APIError) as exc:
|
|
await search_handler({"service": "NOPE_THIS_IS_NOT_REAL_", "query": "x"}, None)
|
|
assert exc.value.error_code == APIErrorCode.INVALID_SERVICE
|
|
|
|
|
|
def test_resolve_vcodec_rejects_junk_as_invalid_input() -> None:
|
|
"""Session routes skip validate_download_parameters, so a bad codec must be a 400, not a 500."""
|
|
from unshackle.core.api.errors import APIErrorCode
|
|
from unshackle.core.api.handlers import resolve_vcodec
|
|
from unshackle.core.tracks import Video
|
|
|
|
assert resolve_vcodec(["H265"]) == [Video.Codec.HEVC]
|
|
assert resolve_vcodec(None) is None
|
|
for junk in ("XVID", 5, [5]):
|
|
with pytest.raises(APIError) as exc:
|
|
resolve_vcodec(junk)
|
|
assert exc.value.error_code == APIErrorCode.INVALID_INPUT
|