mirror of
https://github.com/streamlink/streamlink
synced 2024-11-01 01:19:33 +01:00
stream.dash: refactor DASHStream.parse_manifest()
- Move MPD manifest fetching+parsing logic to separate methods - Rewrite tests based on pytest
This commit is contained in:
bastimeyer
Forrest
parent
ff654873bb
commit
4bd1bf87cb
@ -3,12 +3,13 @@ import datetime
|
||||
import itertools
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from contextlib import contextmanager
|
||||
from contextlib import contextmanager, suppress
|
||||
from time import time
|
||||
from typing import Dict, List, Optional
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
|
||||
from streamlink import PluginError, StreamError
|
||||
from streamlink.exceptions import PluginError, StreamError
|
||||
from streamlink.session import Streamlink
|
||||
from streamlink.stream.dash_manifest import MPD, Representation, Segment, freeze_timeline
|
||||
from streamlink.stream.ffmpegmux import FFMPEGMuxer
|
||||
from streamlink.stream.segmented import SegmentedStreamReader, SegmentedStreamWorker, SegmentedStreamWriter
|
||||
@ -186,15 +187,15 @@ class DASHStream(Stream):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
session,
|
||||
session: Streamlink,
|
||||
mpd: MPD,
|
||||
video_representation: Optional[Representation] = None,
|
||||
audio_representation: Optional[Representation] = None,
|
||||
period: float = 0,
|
||||
period: int = 0,
|
||||
**args,
|
||||
):
|
||||
"""
|
||||
:param streamlink.Streamlink session: Streamlink session instance
|
||||
:param session: Streamlink session instance
|
||||
:param mpd: Parsed MPD manifest
|
||||
:param video_representation: Video representation
|
||||
:param audio_representation: Audio representation
|
||||
@ -231,57 +232,71 @@ class DASHStream(Stream):
|
||||
# the MPD URL has already been prepared by the initial request in `parse_manifest`
|
||||
return self.mpd.url
|
||||
|
||||
@staticmethod
|
||||
def fetch_manifest(session: Streamlink, url_or_manifest: str, **request_args) -> Tuple[str, Dict[str, Any]]:
|
||||
if url_or_manifest.startswith("<?xml"):
|
||||
return url_or_manifest, {}
|
||||
|
||||
retries = session.options.get("dash-manifest-reload-attempts")
|
||||
args = session.http.valid_request_args(**request_args)
|
||||
res = session.http.get(url_or_manifest, retries=retries, **args)
|
||||
manifest: str = res.text
|
||||
url: str = res.url
|
||||
|
||||
urlp = list(urlparse(url))
|
||||
urlp[2], _ = urlp[2].rsplit("/", 1)
|
||||
base_url: str = urlunparse(urlp)
|
||||
|
||||
return manifest, dict(url=url, base_url=base_url)
|
||||
|
||||
@staticmethod
|
||||
def parse_mpd(manifest: str, mpd_params: Dict[str, Any]) -> MPD:
|
||||
node = parse_xml(manifest, ignore_ns=True)
|
||||
|
||||
return MPD(node, **mpd_params)
|
||||
|
||||
@classmethod
|
||||
def parse_manifest(
|
||||
cls,
|
||||
session,
|
||||
session: Streamlink,
|
||||
url_or_manifest: str,
|
||||
**args,
|
||||
) -> Dict[str, "DASHStream"]:
|
||||
"""
|
||||
Parse a DASH manifest file and return its streams.
|
||||
|
||||
:param streamlink.Streamlink session: Streamlink session instance
|
||||
:param session: Streamlink session instance
|
||||
:param url_or_manifest: URL of the manifest file or an XML manifest string
|
||||
:param args: Additional keyword arguments passed to :meth:`requests.Session.request`
|
||||
"""
|
||||
|
||||
if url_or_manifest.startswith("<?xml"):
|
||||
mpd = MPD(parse_xml(url_or_manifest, ignore_ns=True))
|
||||
else:
|
||||
retries = session.options.get("dash-manifest-reload-attempts")
|
||||
res = session.http.get(
|
||||
url_or_manifest,
|
||||
retries=retries,
|
||||
**session.http.valid_request_args(**args),
|
||||
)
|
||||
url = res.url
|
||||
manifest, mpd_params = cls.fetch_manifest(session, url_or_manifest)
|
||||
|
||||
urlp = list(urlparse(url))
|
||||
urlp[2], _ = urlp[2].rsplit("/", 1)
|
||||
|
||||
mpd = MPD(session.http.xml(res, ignore_ns=True), base_url=urlunparse(urlp), url=url)
|
||||
try:
|
||||
mpd = cls.parse_mpd(manifest, mpd_params)
|
||||
except Exception as err:
|
||||
raise PluginError(f"Failed to parse MPD manifest: {err}") from err
|
||||
|
||||
source = mpd_params.get("url", "MPD manifest")
|
||||
video: List[Optional[Representation]] = []
|
||||
audio: List[Optional[Representation]] = []
|
||||
|
||||
# Search for suitable video and audio representations
|
||||
for aset in mpd.periods[0].adaptationSets:
|
||||
if aset.contentProtection:
|
||||
raise PluginError(f"{url} is protected by DRM")
|
||||
raise PluginError(f"{source} is protected by DRM")
|
||||
for rep in aset.representations:
|
||||
if rep.contentProtection:
|
||||
raise PluginError(f"{url} is protected by DRM")
|
||||
raise PluginError(f"{source} is protected by DRM")
|
||||
if rep.mimeType.startswith("video"):
|
||||
video.append(rep)
|
||||
elif rep.mimeType.startswith("audio"):
|
||||
elif rep.mimeType.startswith("audio"): # pragma: no branch
|
||||
audio.append(rep)
|
||||
|
||||
if not video:
|
||||
video = [None]
|
||||
|
||||
video.append(None)
|
||||
if not audio:
|
||||
audio = [None]
|
||||
audio.append(None)
|
||||
|
||||
locale = session.localization
|
||||
locale_lang = locale.language
|
||||
@ -292,20 +307,17 @@ class DASHStream(Stream):
|
||||
for aud in audio:
|
||||
if aud and aud.lang:
|
||||
available_languages.add(aud.lang)
|
||||
try:
|
||||
with suppress(LookupError):
|
||||
if locale.explicit and aud.lang and Language.get(aud.lang) == locale_lang:
|
||||
lang = aud.lang
|
||||
except LookupError:
|
||||
continue
|
||||
|
||||
if not lang:
|
||||
# filter by the first language that appears
|
||||
lang = audio[0].lang if audio[0] else None
|
||||
|
||||
log.debug("Available languages for DASH audio streams: {0} (using: {1})".format(
|
||||
", ".join(available_languages) or "NONE",
|
||||
lang or "n/a",
|
||||
))
|
||||
log.debug(
|
||||
f"Available languages for DASH audio streams: {', '.join(available_languages) or 'NONE'} (using: {lang or 'n/a'})",
|
||||
)
|
||||
|
||||
# if the language is given by the stream, filter out other languages that do not match
|
||||
if len(available_languages) > 1:
|
||||
@ -317,9 +329,9 @@ class DASHStream(Stream):
|
||||
stream_name = []
|
||||
|
||||
if vid:
|
||||
stream_name.append("{:0.0f}{}".format(vid.height or vid.bandwidth_rounded, "p" if vid.height else "k"))
|
||||
stream_name.append(f"{vid.height or vid.bandwidth_rounded:0.0f}{'p' if vid.height else 'k'}")
|
||||
if aud and len(audio) > 1:
|
||||
stream_name.append("a{:0.0f}k".format(aud.bandwidth))
|
||||
stream_name.append(f"a{aud.bandwidth:0.0f}k")
|
||||
ret.append(("+".join(stream_name), stream))
|
||||
|
||||
# rename duplicate streams
|
||||
@ -338,10 +350,8 @@ class DASHStream(Stream):
|
||||
for q in dict_value_list:
|
||||
items = dict_value_list[q]
|
||||
|
||||
try:
|
||||
with suppress(AttributeError):
|
||||
items = sorted(items, key=sortby_bandwidth, reverse=True)
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
for n in range(len(items)):
|
||||
if n == 0:
|
||||
@ -350,6 +360,7 @@ class DASHStream(Stream):
|
||||
ret_new[f"{q}_alt"] = items[n]
|
||||
else:
|
||||
ret_new[f"{q}_alt{n}"] = items[n]
|
||||
|
||||
return ret_new
|
||||
|
||||
def open(self):
|
||||
|
||||
@ -29,3 +29,7 @@ def pytest_collection_modifyitems(items: List[pytest.Item]): # pragma: no cover
|
||||
for item in items
|
||||
}
|
||||
items.sort(key=lambda item: priorities.get(item, default))
|
||||
|
||||
|
||||
def pytest_configure(config: pytest.Config):
|
||||
config.addinivalue_line("markers", "nomockedhttprequest: tests where no mocked HTTP request will be made")
|
||||
|
||||
@ -1,23 +1,59 @@
|
||||
import unittest
|
||||
from typing import List
|
||||
from unittest.mock import ANY, MagicMock, Mock, call, patch
|
||||
from unittest.mock import ANY, Mock, call
|
||||
|
||||
import pytest
|
||||
import requests_mock as rm
|
||||
from lxml.etree import ParseError
|
||||
|
||||
from streamlink import PluginError
|
||||
from streamlink.exceptions import PluginError
|
||||
from streamlink.session import Streamlink
|
||||
from streamlink.stream.dash import DASHStream, DASHStreamWorker
|
||||
from streamlink.stream.dash_manifest import MPD
|
||||
from streamlink.stream.dash_manifest import MPD, MPDParsingError
|
||||
from streamlink.utils.parse import parse_xml as original_parse_xml
|
||||
from tests.resources import text, xml
|
||||
|
||||
|
||||
class TestDASHStream(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.session = MagicMock()
|
||||
self.test_url = "http://test.bar/foo.mpd"
|
||||
self.session.http.get.return_value = Mock(url=self.test_url)
|
||||
@pytest.fixture()
|
||||
def session(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(Streamlink, "load_builtin_plugins", Mock())
|
||||
return Streamlink()
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_video_only(self, mpdClass):
|
||||
|
||||
class TestDASHStreamParseManifest:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _response(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker):
|
||||
invalid = requests_mock.register_uri(rm.ANY, rm.ANY, exc=rm.exceptions.InvalidRequest("Invalid request"))
|
||||
response = requests_mock.register_uri("GET", "http://test/manifest.mpd", **getattr(request, "param", {}))
|
||||
called_once = "nomockedhttprequest" not in request.keywords
|
||||
yield
|
||||
assert not invalid.called
|
||||
assert response.called_once is called_once
|
||||
|
||||
@pytest.fixture()
|
||||
def parse_xml(self, monkeypatch: pytest.MonkeyPatch):
|
||||
parse_xml = Mock(return_value=Mock())
|
||||
monkeypatch.setattr("streamlink.stream.dash.parse_xml", parse_xml)
|
||||
return parse_xml
|
||||
|
||||
@pytest.fixture()
|
||||
def mpd(self, monkeypatch: pytest.MonkeyPatch, parse_xml: Mock):
|
||||
mpd = Mock()
|
||||
monkeypatch.setattr("streamlink.stream.dash.MPD", mpd)
|
||||
return mpd
|
||||
|
||||
@pytest.mark.parametrize(("se_parse_xml", "se_mpd"), [
|
||||
(ParseError, None),
|
||||
(None, MPDParsingError),
|
||||
])
|
||||
def test_parse_fail(self, session: Streamlink, mpd: Mock, parse_xml: Mock, se_parse_xml, se_mpd):
|
||||
parse_xml.side_effect = se_parse_xml
|
||||
mpd.side_effect = se_mpd
|
||||
|
||||
with pytest.raises(PluginError) as cm:
|
||||
DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert str(cm.value).startswith("Failed to parse MPD manifest: ")
|
||||
|
||||
def test_video_only(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -25,15 +61,13 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="2", contentProtection=None, mimeType="video/mp4", height=1080),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p"])
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_only(self, mpdClass):
|
||||
def test_audio_only(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -41,15 +75,13 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="2", contentProtection=None, mimeType="audio/mp4", bandwidth=256.0, lang="en"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["a128k", "a256k"])
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_single(self, mpdClass):
|
||||
def test_audio_single(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -58,15 +90,13 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="3", contentProtection=None, mimeType="audio/aac", bandwidth=128.0, lang="en"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p"])
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_multi(self, mpdClass):
|
||||
def test_audio_multi(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -76,15 +106,13 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="4", contentProtection=None, mimeType="audio/aac", bandwidth=256.0, lang="en"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p+a128k", "1080p+a128k", "720p+a256k", "1080p+a256k"])
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_multi_lang(self, mpdClass):
|
||||
def test_audio_multi_lang(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -94,18 +122,15 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="4", contentProtection=None, mimeType="audio/aac", bandwidth=128.0, lang="es"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p"])
|
||||
assert getattr(streams["720p"].audio_representation, "lang") == "en"
|
||||
assert getattr(streams["1080p"].audio_representation, "lang") == "en"
|
||||
|
||||
assert streams["720p"].audio_representation.lang == "en"
|
||||
assert streams["1080p"].audio_representation.lang == "en"
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_multi_lang_alpha3(self, mpdClass):
|
||||
def test_audio_multi_lang_alpha3(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -115,18 +140,15 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="4", contentProtection=None, mimeType="audio/aac", bandwidth=128.0, lang="spa"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p"])
|
||||
assert getattr(streams["720p"].audio_representation, "lang") == "eng"
|
||||
assert getattr(streams["1080p"].audio_representation, "lang") == "eng"
|
||||
|
||||
assert streams["720p"].audio_representation.lang == "eng"
|
||||
assert streams["1080p"].audio_representation.lang == "eng"
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_invalid_lang(self, mpdClass):
|
||||
def test_audio_invalid_lang(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -135,20 +157,16 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="3", contentProtection=None, mimeType="audio/aac", bandwidth=128.0, lang="en_no_voice"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p"])
|
||||
assert getattr(streams["720p"].audio_representation, "lang") == "en_no_voice"
|
||||
assert getattr(streams["1080p"].audio_representation, "lang") == "en_no_voice"
|
||||
|
||||
assert streams["720p"].audio_representation.lang == "en_no_voice"
|
||||
assert streams["1080p"].audio_representation.lang == "en_no_voice"
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_audio_multi_lang_locale(self, mpdClass):
|
||||
self.session.localization.language.alpha2 = "es"
|
||||
self.session.localization.explicit = True
|
||||
def test_audio_multi_lang_locale(self, monkeypatch: pytest.MonkeyPatch, session: Streamlink, mpd: Mock):
|
||||
session.set_option("locale", "es_ES")
|
||||
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
@ -159,91 +177,16 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="4", contentProtection=None, mimeType="audio/aac", bandwidth=128.0, lang="es"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p"])
|
||||
assert getattr(streams["720p"].audio_representation, "lang") == "es"
|
||||
assert getattr(streams["1080p"].audio_representation, "lang") == "es"
|
||||
|
||||
assert streams["720p"].audio_representation.lang == "es"
|
||||
assert streams["1080p"].audio_representation.lang == "es"
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_drm_adaptationset(self, mpdClass):
|
||||
adaptationset = Mock(
|
||||
contentProtection="DRM",
|
||||
representations=[],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
with pytest.raises(PluginError):
|
||||
DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_drm_representation(self, mpdClass):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
Mock(id="1", contentProtection="DRM"),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
with pytest.raises(PluginError):
|
||||
DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
|
||||
def test_parse_manifest_string(self):
|
||||
with text("dash/test_9.mpd") as mpd_txt:
|
||||
test_manifest = mpd_txt.read()
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, test_manifest)
|
||||
assert list(streams.keys()) == ["2500k"]
|
||||
|
||||
@patch("streamlink.stream.dash.DASHStreamReader")
|
||||
@patch("streamlink.stream.dash.FFMPEGMuxer")
|
||||
def test_stream_open_video_only(self, muxer: Mock, reader: Mock):
|
||||
rep_video = Mock(ident=(None, None, "1"), mimeType="video/mp4")
|
||||
stream = DASHStream(self.session, Mock(), rep_video)
|
||||
stream.open()
|
||||
|
||||
assert reader.call_args_list == [call(stream, rep_video)]
|
||||
reader_video = reader(stream, rep_video)
|
||||
assert reader_video.open.called_once
|
||||
assert muxer.call_args_list == []
|
||||
|
||||
@patch("streamlink.stream.dash.DASHStreamReader")
|
||||
@patch("streamlink.stream.dash.FFMPEGMuxer")
|
||||
def test_stream_open_video_audio(self, muxer: Mock, reader: Mock):
|
||||
rep_video = Mock(ident=(None, None, "1"), mimeType="video/mp4")
|
||||
rep_audio = Mock(ident=(None, None, "2"), mimeType="audio/mp3", lang="en")
|
||||
stream = DASHStream(self.session, Mock(), rep_video, rep_audio)
|
||||
stream.open()
|
||||
|
||||
assert reader.call_args_list == [call(stream, rep_video), call(stream, rep_audio)]
|
||||
reader_video = reader(stream, rep_video)
|
||||
reader_audio = reader(stream, rep_audio)
|
||||
assert reader_video.open.called_once
|
||||
assert reader_audio.open.called_once
|
||||
assert muxer.call_args_list == [call(self.session, reader_video, reader_audio, copyts=True)]
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_segments_number_time(self, mpdClass):
|
||||
with xml("dash/test_9.mpd") as mpd_xml:
|
||||
mpdClass.return_value = MPD(mpd_xml, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
|
||||
assert list(streams.keys()) == ["2500k"]
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_with_duplicated_resolutions(self, mpdClass):
|
||||
"""
|
||||
Verify the fix for https://github.com/streamlink/streamlink/issues/3365
|
||||
"""
|
||||
# Verify the fix for https://github.com/streamlink/streamlink/issues/3365
|
||||
def test_duplicated_resolutions(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -253,18 +196,14 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="4", contentProtection=None, mimeType="video/mp4", height=720),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert sorted(streams.keys()) == sorted(["720p", "1080p", "1080p_alt", "1080p_alt2"])
|
||||
|
||||
@patch("streamlink.stream.dash.MPD")
|
||||
def test_parse_manifest_with_duplicated_resolutions_sorted_bandwidth(self, mpdClass):
|
||||
"""
|
||||
Verify the fix for https://github.com/streamlink/streamlink/issues/4217
|
||||
"""
|
||||
# Verify the fix for https://github.com/streamlink/streamlink/issues/4217
|
||||
def test_duplicated_resolutions_sorted_bandwidth(self, session: Streamlink, mpd: Mock):
|
||||
adaptationset = Mock(
|
||||
contentProtection=None,
|
||||
representations=[
|
||||
@ -273,14 +212,87 @@ class TestDASHStream(unittest.TestCase):
|
||||
Mock(id="3", contentProtection=None, mimeType="video/mp4", height=1080, bandwidth=32.0),
|
||||
],
|
||||
)
|
||||
mpdClass.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
streams = DASHStream.parse_manifest(self.session, self.test_url)
|
||||
mpdClass.assert_called_with(ANY, base_url="http://test.bar", url="http://test.bar/foo.mpd")
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert getattr(streams["1080p"].video_representation, "bandwidth") == pytest.approx(128.0)
|
||||
assert getattr(streams["1080p_alt"].video_representation, "bandwidth") == pytest.approx(64.0)
|
||||
assert getattr(streams["1080p_alt2"].video_representation, "bandwidth") == pytest.approx(32.0)
|
||||
|
||||
assert streams["1080p"].video_representation.bandwidth == pytest.approx(128.0)
|
||||
assert streams["1080p_alt"].video_representation.bandwidth == pytest.approx(64.0)
|
||||
assert streams["1080p_alt2"].video_representation.bandwidth == pytest.approx(32.0)
|
||||
@pytest.mark.parametrize("adaptationset", [
|
||||
pytest.param(
|
||||
Mock(contentProtection="DRM", representations=[]),
|
||||
id="ContentProtection on AdaptationSet",
|
||||
),
|
||||
pytest.param(
|
||||
Mock(contentProtection=None, representations=[Mock(id="1", contentProtection="DRM")]),
|
||||
id="ContentProtection on Representation",
|
||||
),
|
||||
])
|
||||
def test_contentprotection(self, session: Streamlink, mpd: Mock, adaptationset: Mock):
|
||||
mpd.return_value = Mock(periods=[Mock(adaptationSets=[adaptationset])])
|
||||
|
||||
with pytest.raises(PluginError):
|
||||
DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
|
||||
@pytest.mark.nomockedhttprequest()
|
||||
def test_string(self, session: Streamlink, mpd: Mock, parse_xml: Mock):
|
||||
with text("dash/test_9.mpd") as mpd_txt:
|
||||
test_manifest = mpd_txt.read()
|
||||
parse_xml.side_effect = original_parse_xml
|
||||
mpd.side_effect = MPD
|
||||
|
||||
streams = DASHStream.parse_manifest(session, test_manifest)
|
||||
assert mpd.call_args_list == [call(ANY)]
|
||||
assert list(streams.keys()) == ["2500k"]
|
||||
|
||||
# TODO: Move this test to test_dash_parser and properly test segment URLs.
|
||||
# This test currently achieves nothing... (manifest fixture added in 7aada92)
|
||||
def test_segments_number_time(self, session: Streamlink, mpd: Mock):
|
||||
with xml("dash/test_9.mpd") as mpd_xml:
|
||||
mpd.return_value = MPD(mpd_xml, base_url="http://test", url="http://test/manifest.mpd")
|
||||
|
||||
streams = DASHStream.parse_manifest(session, "http://test/manifest.mpd")
|
||||
assert mpd.call_args_list == [call(ANY, url="http://test/manifest.mpd", base_url="http://test")]
|
||||
assert list(streams.keys()) == ["2500k"]
|
||||
|
||||
|
||||
class TestDASHStreamOpen:
|
||||
@pytest.fixture()
|
||||
def reader(self, monkeypatch: pytest.MonkeyPatch):
|
||||
reader = Mock()
|
||||
monkeypatch.setattr("streamlink.stream.dash.DASHStreamReader", reader)
|
||||
return reader
|
||||
|
||||
@pytest.fixture()
|
||||
def muxer(self, monkeypatch: pytest.MonkeyPatch):
|
||||
muxer = Mock()
|
||||
monkeypatch.setattr("streamlink.stream.dash.FFMPEGMuxer", muxer)
|
||||
return muxer
|
||||
|
||||
def test_stream_open_video_only(self, session: Streamlink, muxer: Mock, reader: Mock):
|
||||
rep_video = Mock(ident=(None, None, "1"), mimeType="video/mp4")
|
||||
stream = DASHStream(session, Mock(), rep_video)
|
||||
stream.open()
|
||||
|
||||
assert reader.call_args_list == [call(stream, rep_video)]
|
||||
reader_video = reader(stream, rep_video)
|
||||
assert reader_video.open.called_once
|
||||
assert muxer.call_args_list == []
|
||||
|
||||
def test_stream_open_video_audio(self, session: Streamlink, muxer: Mock, reader: Mock):
|
||||
rep_video = Mock(ident=(None, None, "1"), mimeType="video/mp4")
|
||||
rep_audio = Mock(ident=(None, None, "2"), mimeType="audio/mp3", lang="en")
|
||||
stream = DASHStream(session, Mock(), rep_video, rep_audio)
|
||||
stream.open()
|
||||
|
||||
assert reader.call_args_list == [call(stream, rep_video), call(stream, rep_audio)]
|
||||
reader_video = reader(stream, rep_video)
|
||||
reader_audio = reader(stream, rep_audio)
|
||||
assert reader_video.open.called_once
|
||||
assert reader_audio.open.called_once
|
||||
assert muxer.call_args_list == [call(session, reader_video, reader_audio, copyts=True)]
|
||||
|
||||
|
||||
class TestDASHStreamWorker:
|
||||
@ -373,6 +385,7 @@ class TestDASHStreamWorker:
|
||||
assert representation.segments.call_args_list == [call(init=True)]
|
||||
assert worker._wait.is_set()
|
||||
|
||||
# Verify the fix for https://github.com/streamlink/streamlink/issues/2873
|
||||
@pytest.mark.parametrize("duration", [
|
||||
0,
|
||||
204.32,
|
||||
@ -387,9 +400,6 @@ class TestDASHStreamWorker:
|
||||
segments: List[Mock],
|
||||
mpd: Mock,
|
||||
):
|
||||
"""
|
||||
Verify the fix for https://github.com/streamlink/streamlink/issues/2873
|
||||
"""
|
||||
mpd.dynamic = False
|
||||
mpd.type = "static"
|
||||
mpd.periods[0].duration.total_seconds.return_value = duration
|
||||
|
||||
Loading…
Reference in New Issue
Block a user