github
/
streamlink
mirror of https://github.com/streamlink/streamlink
1
Fork 0
Code Releases Activity

plugin: move options to the plugin instance 

Plugin options are currently shared between all plugin instances via the
`Plugin.options` class attribute, unless it gets overridden by
Streamlink's CLI via `setup_plugin_args()` after reading the plugin
arguments, updating the argument parser and reading the default values.

This allows for accidental pollution of the plugin options when not
overriding the `options` class attribute with a new `Options` instance.
Since this is only done by `streamlink_cli`, this is an issue in tests
and third party python projects using the Streamlink API.

- Remove `Plugin.options` class attribute and store an `Options`
  instance on the `Plugin` instance instead, and allow passing an
  already initialized options instance to the `Plugin` constructor
- Turn `Plugin.{g,s}et_option()` from class methods to regular methods
- Update CLI and initialize the resolved plugin with an options instance
  that gets built by `setup_plugin_options()`, with default values read
  from the CLI arguments
- Remove `Streamlink.{g,s}et_plugin_option()` and fix usages:
  - Session and Options tests
  - Twitch plugin, TwitchAPI, TwitchHLSStream, and their tests
- Add `keywords` mapping to `HLSStream.parse_variant_playlist`,
  to be able to pass custom keywords to the `{,Muxed}HLSStream`
- Move and rewrite CLI plugin args+options integration tests
This commit is contained in:
bastimeyer 2022-12-16 09:25:45 +01:00 committed by Sebastian Meyer
parent 5c8c0b213c
commit d314c5b895
10 changed files with 227 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/streamlink/plugin/plugin.py
View File

@ -270,7 +270,6 @@ class Plugin:
category: Optional[str] = None
"""Metadata 'category' attribute: name of a game being played, a music genre, etc."""
options = Options()
_url: str = ""
# deprecated
@ -317,15 +316,17 @@ class Plugin:
return cls.__new__(PluginWrapperBack, *args, **kwargs)
def __init__(self, session: "Streamlink", url: str):
def __init__(self, session: "Streamlink", url: str, options: Optional[Options] = None):
"""
:param session: The Streamlink session instance
:param url: The input URL used for finding and resolving streams
:param options: An optional :class:`Options` instance
"""
modulename = self.__class__.__module__
self.module = modulename.split(".")[-1]
self.logger = logging.getLogger(modulename)
self.options = Options() if options is None else options
self.cache = Cache(
filename="plugin-cache.json",
key_prefix=self.module,
@ -353,13 +354,11 @@ class Plugin:
if self.matchers:
self.matcher, self.match = self.matches.update(self.matchers, value)
@classmethod
def set_option(cls, key, value):
cls.options.set(key, value)
def set_option(self, key, value):
self.options.set(key, value)
@classmethod
def get_option(cls, key):
return cls.options.get(key)
def get_option(self, key):
return self.options.get(key)
@classmethod
def get_argument(cls, key):

29
src/streamlink/plugins/twitch.py
View File

@ -202,10 +202,10 @@ class TwitchHLSStreamReader(HLSStreamReader):
class TwitchHLSStream(HLSStream):
__reader__ = TwitchHLSStreamReader
def __init__(self, *args, **kwargs):
def __init__(self, *args, disable_ads: bool = False, low_latency: bool = False, **kwargs):
super().__init__(*args, **kwargs)
self.disable_ads = self.session.get_plugin_option("twitch", "disable-ads")
self.low_latency = self.session.get_plugin_option("twitch", "low-latency")
self.disable_ads = disable_ads
self.low_latency = low_latency
class UsherService:
@ -253,13 +253,13 @@ class UsherService:
class TwitchAPI:
CLIENT_ID = "kimne78kx3ncx6brgo4mv6wki5h1ko"
def __init__(self, session):
def __init__(self, session, api_header=None, access_token_param=None):
self.session = session
self.headers = {
"Client-ID": self.CLIENT_ID,
}
self.headers.update(**dict(session.get_plugin_option("twitch", "api-header") or []))
self.access_token_params = dict(session.get_plugin_option("twitch", "access-token-param") or [])
self.headers.update(**dict(api_header or []))
self.access_token_params = dict(access_token_param or [])
self.access_token_params.setdefault("playerType", "embed")
def call(self, data, schema=None, **kwargs):
@ -604,7 +604,11 @@ class Twitch(Plugin):
self.video_id = match.get("video_id") or match.get("videos_id")
self.clip_name = match.get("clip_name")
self.api = TwitchAPI(session=self.session)
self.api = TwitchAPI(
session=self.session,
api_header=self.get_option("api-header"),
access_token_param=self.get_option("access-token-param"),
)
self.usher = UsherService(session=self.session)
def method_factory(parent_method):
@ -698,7 +702,16 @@ class Twitch(Plugin):
time_offset = 0
try:
streams = TwitchHLSStream.parse_variant_playlist(self.session, url, start_offset=time_offset, **extra_params)
streams = TwitchHLSStream.parse_variant_playlist(
self.session,
url,
start_offset=time_offset,
keywords={
"disable_ads": self.get_option("disable-ads"),
"low_latency": self.get_option("low-latency"),
},
**extra_params,
)
except OSError as err:
err = str(err)
if "404 Client Error" in err or "Failed to parse playlist" in err:

25
src/streamlink/session.py
View File

@ -501,31 +501,6 @@ class Streamlink:
return self.options.get(key)
def set_plugin_option(self, plugin: str, key: str, value: Any) -> None:
"""
Sets plugin specific options used by plugins originating from this session object.
:param plugin: name of the plugin
:param key: key of the option
:param value: value to set the option to
"""
if plugin in self.plugins:
plugincls = self.plugins[plugin]
plugincls.set_option(key, value)
def get_plugin_option(self, plugin: str, key: str) -> Optional[Any]:
"""
Returns the current value of the plugin specific option.
:param plugin: name of the plugin
:param key: key of the option
"""
if plugin in self.plugins:
plugincls = self.plugins[plugin]
return plugincls.get_option(key)
@lru_cache(maxsize=128) # noqa: B019
def resolve_url(
self,

9
src/streamlink/stream/hls.py
View File

@ -3,7 +3,7 @@ import re
import struct
from concurrent.futures import Future
from datetime import datetime, timedelta
from typing import Any, Dict, List, NamedTuple, Optional, Tuple, Union
from typing import Any, Dict, List, Mapping, NamedTuple, Optional, Tuple, Union
from urllib.parse import urlparse
# noinspection PyPackageRequirements
@ -601,6 +601,7 @@ class HLSStream(HTTPStream):
force_restart: bool = False,
start_offset: float = 0,
duration: Optional[float] = None,
# TODO: turn args into dedicated keyword
**args,
):
"""
@ -680,6 +681,8 @@ class HLSStream(HTTPStream):
name_fmt: Optional[str] = None,
start_offset: float = 0,
duration: Optional[float] = None,
keywords: Optional[Mapping] = None,
# TODO: turn request_params into a dedicated keyword
**request_params,
) -> Dict[str, Union["HLSStream", "MuxedHLSStream"]]:
"""
@ -694,6 +697,7 @@ class HLSStream(HTTPStream):
:param name_fmt: A format string for the name, allowed format keys are: name, pixels, bitrate
:param start_offset: Number of seconds to be skipped from the beginning
:param duration: Number of second until ending the stream
:param keywords: Optional keywords to be passed to the :class:`HLSStream` or :class:`MuxedHLSStream`
:param request_params: Additional keyword arguments passed to :class:`HLSStream`, :class:`MuxedHLSStream`,
or :py:meth:`requests.Session.request`
"""
@ -711,6 +715,7 @@ class HLSStream(HTTPStream):
stream_name: Optional[str]
stream: Union["HLSStream", "MuxedHLSStream"]
streams: Dict[str, Union["HLSStream", "MuxedHLSStream"]] = {}
keywords = keywords or {}
for playlist in filter(lambda p: not p.is_iframe, multivariant.playlists):
names: Dict[str, Optional[str]] = dict(name=None, pixels=None, bitrate=None)
@ -819,6 +824,7 @@ class HLSStream(HTTPStream):
force_restart=force_restart,
start_offset=start_offset,
duration=duration,
**keywords,
**request_params,
)
else:
@ -829,6 +835,7 @@ class HLSStream(HTTPStream):
force_restart=force_restart,
start_offset=start_offset,
duration=duration,
**keywords,
**request_params,
)

49
src/streamlink_cli/main.py
View File

@ -16,7 +16,8 @@ from typing import Any, Dict, List, Optional, Type, Union
import streamlink.logger as logger
from streamlink import NoPluginError, PluginError, StreamError, Streamlink, __version__ as streamlink_version
from streamlink.exceptions import FatalPluginError, StreamlinkDeprecationWarning
from streamlink.plugin import Plugin, PluginOptions
from streamlink.options import Options
from streamlink.plugin import Plugin
from streamlink.stream.stream import Stream, StreamIO
from streamlink.utils.named_pipe import NamedPipe
from streamlink.utils.times import LOCAL as LOCALTIMEZONE
@ -530,8 +531,8 @@ def handle_url():
try:
pluginname, pluginclass, resolved_url = streamlink.resolve_url(args.url)
setup_plugin_options(streamlink, pluginname, pluginclass)
plugin = pluginclass(streamlink, resolved_url)
options = setup_plugin_options(pluginname, pluginclass)
plugin = pluginclass(streamlink, resolved_url, options)
log.info(f"Found matching plugin {pluginname} for URL {args.url}")
if args.retry_max or args.retry_streams:
@ -711,33 +712,34 @@ def setup_streamlink():
def setup_plugin_args(session: Streamlink, parser: ArgumentParser):
"""Adds plugin argument data to the argument parser and sets default plugin options."""
"""Adds plugin argument data to the argument parser."""
plugin_args = parser.add_argument_group("Plugin options")
for pname, plugin in session.plugins.items():
defaults = {}
group = parser.add_argument_group(pname.capitalize(), parent=plugin_args)
for parg in plugin.arguments or []:
group.add_argument(parg.argument_name(pname), **parg.options)
defaults[parg.dest] = parg.default
plugin.options = PluginOptions(defaults)
def setup_plugin_options(session: Streamlink, pluginname: str, pluginclass: Type[Plugin]):
"""Sets Streamlink plugin options."""
if pluginclass.arguments is None:
return
def setup_plugin_options(pluginname: str, pluginclass: Type[Plugin]) -> Options:
"""Initializes plugin options from argument values."""
if not pluginclass.arguments:
return Options()
defaults = {}
values = {}
required = {}
for parg in pluginclass.arguments:
defaults[parg.dest] = parg.default
if parg.options.get("help") == argparse.SUPPRESS:
continue
value = getattr(args, parg.namespace_dest(pluginname))
session.set_plugin_option(pluginname, parg.dest, value)
values[parg.dest] = value
if parg.required:
required[parg.name] = parg
@ -746,19 +748,20 @@ def setup_plugin_options(session: Streamlink, pluginname: str, pluginclass: Type
try:
for rparg in pluginclass.arguments.requires(parg.name):
required[rparg.name] = rparg
except RuntimeError:
except RuntimeError: # pragma: no cover
log.error(f"{pluginname} plugin has a configuration error and the arguments cannot be parsed")
break
if required:
for req in required.values():
if not session.get_plugin_option(pluginname, req.dest):
prompt = f"{req.prompt or f'Enter {pluginname} {req.name}'}: "
session.set_plugin_option(
pluginname,
req.dest,
console.askpass(prompt) if req.sensitive else console.ask(prompt),
)
for req in required.values():
if not values.get(req.dest):
prompt = f"{req.prompt or f'Enter {pluginname} {req.name}'}: "
value = console.askpass(prompt) if req.sensitive else console.ask(prompt)
values[req.dest] = value
options = Options(defaults)
options.update(values)
return options
def log_root_warning():

129
tests/cli/test_plugin_args_and_options.py Normal file
View File

@ -0,0 +1,129 @@
import argparse
from typing import Type
from unittest.mock import Mock, call
import pytest
from streamlink.plugin import Plugin, pluginargument
from streamlink.session import Streamlink
from streamlink_cli.argparser import ArgumentParser
from streamlink_cli.main import setup_plugin_args, setup_plugin_options
@pytest.fixture()
def parser():
return ArgumentParser(add_help=False)
@pytest.fixture(autouse=True)
def _args(monkeypatch: pytest.MonkeyPatch):
args = argparse.Namespace(
mock_foo_bar=123,
mock_baz=654,
# mock_qux wouldn't be set by the parser if the argument is suppressed
# its value will be ignored
mock_qux=987,
mock_user="username",
mock_pass=None,
mock_captcha=None,
)
monkeypatch.setattr("streamlink_cli.main.args", args)
@pytest.fixture()
def plugin():
# simple argument which requires namespace-name normalization
@pluginargument("foo-bar")
# argument with default value
@pluginargument("baz", default=456)
# suppressed argument
@pluginargument("qux", default=789, help=argparse.SUPPRESS)
# required argument with dependencies
@pluginargument("user", required=True, requires=["pass", "captcha"])
# sensitive argument (using console.askpass if unset)
@pluginargument("pass", sensitive=True)
# argument with custom prompt (using console.ask if unset)
@pluginargument("captcha", prompt="CAPTCHA code")
class FakePlugin(Plugin):
def _get_streams(self): # pragma: no cover
pass
return FakePlugin
@pytest.fixture(autouse=True)
def session(monkeypatch: pytest.MonkeyPatch, parser: ArgumentParser, plugin: Type[Plugin]):
monkeypatch.setattr("streamlink.session.Streamlink.load_builtin_plugins", Mock())
session = Streamlink()
session.plugins["mock"] = plugin
setup_plugin_args(session, parser)
return session
@pytest.fixture()
def console(monkeypatch: pytest.MonkeyPatch):
console = Mock()
monkeypatch.setattr("streamlink_cli.main.console", console)
return console
class TestPluginArgs:
def test_arguments(self, parser: ArgumentParser, plugin: Type[Plugin]):
group_plugins = next((grp for grp in parser._action_groups if grp.title == "Plugin options"), None) # pragma: no branch
assert group_plugins is not None, "Adds the 'Plugin options' arguments group"
assert group_plugins in parser.NESTED_ARGUMENT_GROUPS[None], "Adds the 'Plugin options' arguments group"
group_plugin = next((grp for grp in parser._action_groups if grp.title == "Mock"), None) # pragma: no branch
assert group_plugin is not None, "Adds the 'Mock' arguments group"
assert group_plugin in parser.NESTED_ARGUMENT_GROUPS[group_plugins], "Adds the 'Mock' arguments group"
assert [
item
for action in parser._actions
for item in action.option_strings
if action.help != argparse.SUPPRESS
] == [
"--mock-foo-bar",
"--mock-baz",
"--mock-user",
"--mock-pass",
"--mock-captcha",
], "Parser has all arguments registered"
class TestPluginOptions:
def test_empty(self, console: Mock):
options = setup_plugin_options("mock", Plugin)
assert not options.defaults
assert not options.options
assert not console.ask.called
assert not console.askpass.called
def test_options(self, plugin: Type[Plugin], console: Mock):
options = setup_plugin_options("mock", plugin)
assert console.ask.call_args_list == [call("CAPTCHA code: ")]
assert console.askpass.call_args_list == [call("Enter mock pass: ")]
assert plugin.arguments
arg_foo = plugin.arguments.get("foo-bar")
arg_baz = plugin.arguments.get("baz")
arg_qux = plugin.arguments.get("qux")
assert arg_foo
assert arg_baz
assert arg_qux
assert arg_foo.default is None
assert arg_baz.default == 456
assert arg_qux.default == 789
assert options.get("foo-bar") == 123, "Overrides the default plugin-argument value"
assert options.get("baz") == 654, "Uses the plugin-argument default value"
assert options.get("qux") == 789, "Ignores values of suppressed plugin-arguments"
options.clear()
assert options.get("foo-bar") == arg_foo.default
assert options.get("baz") == arg_baz.default
assert options.get("qux") == arg_qux.default

49
tests/plugins/test_twitch.py
View File

@ -7,6 +7,7 @@ import requests_mock as rm
from streamlink import Streamlink
from streamlink.exceptions import NoStreamsError
from streamlink.options import Options
from streamlink.plugins.twitch import Twitch, TwitchAPI, TwitchHLSStream, TwitchHLSStreamReader, TwitchHLSStreamWriter
from tests.mixins.stream_hls import EventedHLSStreamWriter, Playlist, Segment as _Segment, Tag, TestMixinStreamHLS
from tests.plugins import PluginCanHandleUrl
@ -119,12 +120,11 @@ def test_stream_weight(requests_mock: rm.Mocker):
@patch("streamlink.stream.hls.HLSStreamWorker.wait", MagicMock(return_value=True))
class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
__stream__ = _TwitchHLSStream
stream: TwitchHLSStream
def get_session(self, options=None, disable_ads=False, low_latency=False):
session = super().get_session(options)
def get_session(self, *args, **kwargs):
session = super().get_session(*args, **kwargs)
session.set_option("hls-live-edge", 4)
session.set_plugin_option("twitch", "disable-ads", disable_ads)
session.set_plugin_option("twitch", "low-latency", low_latency)
return session
@ -139,7 +139,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [daterange, Segment(0), Segment(1)], end=True),
], disable_ads=True, low_latency=False)
], streamoptions={"disable_ads": True, "low_latency": False})
self.await_write(2)
data = self.await_read(read_all=True)
@ -157,7 +157,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [daterange, Segment(0), Segment(1)], end=True),
], disable_ads=True, low_latency=False)
], streamoptions={"disable_ads": True, "low_latency": False})
self.await_write(2)
data = self.await_read(read_all=True)
@ -175,7 +175,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [daterange, Segment(0), Segment(1)], end=True),
], disable_ads=True, low_latency=False)
], streamoptions={"disable_ads": True, "low_latency": False})
self.await_write(2)
data = self.await_read(read_all=True)
@ -193,7 +193,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [daterange, Segment(0), Segment(1)], end=True),
], disable_ads=True, low_latency=False)
], streamoptions={"disable_ads": True, "low_latency": False})
self.await_write(2)
data = self.await_read(read_all=True)
@ -207,7 +207,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
Playlist(0, [daterange, Segment(0), Segment(1)]),
Playlist(2, [daterange, Segment(2), Segment(3)]),
Playlist(4, [Segment(4), Segment(5)], end=True),
], disable_ads=True, low_latency=False)
], streamoptions={"disable_ads": True, "low_latency": False})
self.await_write(6)
data = self.await_read(read_all=True)
@ -225,7 +225,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
Playlist(0, [Segment(0), Segment(1)]),
Playlist(2, [daterange, Segment(2), Segment(3)]),
Playlist(4, [Segment(4), Segment(5)], end=True),
], disable_ads=True, low_latency=False)
], streamoptions={"disable_ads": True, "low_latency": False})
self.await_write(6)
data = self.await_read(read_all=True)
@ -241,7 +241,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [daterange, Segment(0), Segment(1)]),
Playlist(2, [Segment(2), Segment(3)], end=True),
], disable_ads=False, low_latency=False)
], streamoptions={"disable_ads": False, "low_latency": False})
self.await_write(4)
data = self.await_read(read_all=True)
@ -254,7 +254,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [Segment(0), Segment(1), Segment(2), Segment(3), SegmentPrefetch(4), SegmentPrefetch(5)]),
Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True),
], disable_ads=False, low_latency=True)
], streamoptions={"disable_ads": False, "low_latency": True})
assert self.session.options.get("hls-live-edge") == 2
assert self.session.options.get("hls-segment-stream-data")
@ -273,7 +273,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [Segment(0), Segment(1), Segment(2), Segment(3), SegmentPrefetch(4), SegmentPrefetch(5)]),
Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True),
], disable_ads=False, low_latency=False)
], streamoptions={"disable_ads": False, "low_latency": False})
assert self.session.options.get("hls-live-edge") == 4
assert not self.session.options.get("hls-segment-stream-data")
@ -290,10 +290,10 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
self.subject([
Playlist(0, [Segment(0), Segment(1), Segment(2), Segment(3)]),
Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7)], end=True),
], disable_ads=False, low_latency=True)
], streamoptions={"disable_ads": False, "low_latency": True})
assert self.session.get_plugin_option("twitch", "low-latency")
assert not self.session.get_plugin_option("twitch", "disable-ads")
assert not self.stream.disable_ads
assert self.stream.low_latency
self.await_write(6)
self.await_read(read_all=True)
@ -308,7 +308,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
thread, segments = self.subject([
Playlist(0, [daterange, Segment(0), Segment(1), Segment(2), Segment(3)]),
Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True),
], disable_ads=False, low_latency=True)
], streamoptions={"disable_ads": False, "low_latency": True})
self.await_write(8)
data = self.await_read(read_all=True)
@ -323,7 +323,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
self.subject([
Playlist(0, [daterange, Segment(0), Segment(1), Segment(2), Segment(3)]),
Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True),
], disable_ads=True, low_latency=True)
], streamoptions={"disable_ads": True, "low_latency": True})
self.await_write(8)
self.await_read(read_all=True)
@ -357,7 +357,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
Playlist(5, ads + [Seg(5), Seg(6), Seg(7), Seg(8), Pre(9), Pre(10)]),
Playlist(6, ads + [Seg(6), Seg(7), Seg(8), Seg(9), Pre(10), Pre(11)]),
Playlist(7, [Seg(7), Seg(8), Seg(9), Seg(10), Pre(11), Pre(12)], end=True),
], disable_ads=True, low_latency=True)
], streamoptions={"disable_ads": True, "low_latency": True})
self.await_write(11)
data = self.await_read(read_all=True)
@ -373,7 +373,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
self.subject([
Playlist(0, [daterange, Segment(0), Segment(1), Segment(2), Segment(3)]),
Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7)], end=True),
], disable_ads=True, low_latency=True)
], streamoptions={"disable_ads": True, "low_latency": True})
self.await_write(6)
self.await_read(read_all=True)
@ -388,7 +388,7 @@ class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
Seg, SegPre = Segment, SegmentPrefetch
self.subject([
Playlist(0, [Seg(0, duration=5), Seg(1, duration=7), Seg(2, duration=11), SegPre(3)], end=True),
], low_latency=True)
], streamoptions={"low_latency": True})
self.await_write(4)
self.await_read(read_all=True)
@ -399,10 +399,11 @@ class TestTwitchAPIAccessToken:
@pytest.fixture()
def plugin(self, request: pytest.FixtureRequest):
session = Streamlink()
options = Options()
for param in getattr(request, "param", {}):
session.set_plugin_option("twitch", *param)
yield Twitch(session, "https://twitch.tv/channelname")
Twitch.options.clear()
options.set(*param)
return Twitch(session, "https://twitch.tv/channelname", options)
@pytest.fixture()
def mock(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker):

67
tests/test_options.py
View File

@ -1,13 +1,8 @@
import argparse
import unittest
from unittest.mock import Mock, patch
import pytest
from streamlink.options import Argument, Arguments, Options
from streamlink.plugin import Plugin, pluginargument
from streamlink_cli.argparser import ArgumentParser
from streamlink_cli.main import setup_plugin_args, setup_plugin_options
class TestOptions(unittest.TestCase):
@ -190,65 +185,3 @@ class TestArguments(unittest.TestCase):
with pytest.raises(RuntimeError):
list(args.requires("test1"))
class TestSetupOptions:
def test_setup_plugin_args(self):
session = Mock()
plugin = Mock()
parser = ArgumentParser(add_help=False)
session.plugins = {"mock": plugin}
plugin.arguments = Arguments(
Argument("test1", default="default1"),
Argument("test2", default="default2"),
Argument("test3"),
)
setup_plugin_args(session, parser)
group_plugins = next((grp for grp in parser._action_groups if grp.title == "Plugin options"), None) # pragma: no branch
assert group_plugins is not None, "Adds the 'Plugin options' arguments group"
assert group_plugins in parser.NESTED_ARGUMENT_GROUPS[None], "Adds the 'Plugin options' arguments group"
group_plugin = next((grp for grp in parser._action_groups if grp.title == "Mock"), None) # pragma: no branch
assert group_plugin is not None, "Adds the 'Mock' arguments group"
assert group_plugin in parser.NESTED_ARGUMENT_GROUPS[group_plugins], "Adds the 'Mock' arguments group"
assert [item for action in group_plugin._group_actions for item in action.option_strings] \
== ["--mock-test1", "--mock-test2", "--mock-test3"], \
"Only adds plugin arguments and ignores global argument references"
assert [item for action in parser._actions for item in action.option_strings] \
== ["--mock-test1", "--mock-test2", "--mock-test3"], \
"Parser has all arguments registered"
assert plugin.options.get("test1") == "default1"
assert plugin.options.get("test2") == "default2"
assert plugin.options.get("test3") is None
def test_setup_plugin_options(self):
@pluginargument("foo-foo")
@pluginargument("bar-bar", default=456)
@pluginargument("baz-baz", default=789, help=argparse.SUPPRESS)
class FakePlugin(Plugin):
def _get_streams(self): # pragma: no cover
pass
session = Mock()
parser = ArgumentParser()
session.plugins = {"plugin": FakePlugin}
session.set_plugin_option = lambda name, key, value: session.plugins[name].options.update({key: value})
with patch("streamlink_cli.main.args") as args:
args.plugin_foo_foo = 123
args.plugin_bar_bar = 654
args.plugin_baz_baz = 987 # this wouldn't be set by the parser if the argument is suppressed
setup_plugin_args(session, parser)
assert FakePlugin.options.get("foo_foo") is None, "No default value"
assert FakePlugin.options.get("bar_bar") == 456, "Sets the plugin-argument's default value"
assert FakePlugin.options.get("baz_baz") == 789, "Sets the suppressed plugin-argument's default value"
setup_plugin_options(session, "plugin", FakePlugin)
assert FakePlugin.options.get("foo_foo") == 123, "Overrides the default plugin-argument value"
assert FakePlugin.options.get("bar_bar") == 654, "Sets the provided plugin-argument value"
assert FakePlugin.options.get("baz_baz") == 789, "Doesn't set values of suppressed plugin-arguments"

11
tests/test_plugin.py
View File

@ -9,6 +9,7 @@ import freezegun
import pytest
import requests.cookies
from streamlink.options import Options
from streamlink.plugin import (
HIGH_PRIORITY,
NORMAL_PRIORITY,
@ -107,6 +108,16 @@ class TestPlugin:
assert mock_load_cookies.call_args_list == [call()]
def test_constructor_options(self):
one = FakePlugin(Mock(), "https://mocked", Options({"key": "val"}))
two = FakePlugin(Mock(), "https://mocked")
assert one.get_option("key") == "val"
assert two.get_option("key") is None
one.set_option("key", "other")
assert one.get_option("key") == "other"
assert two.get_option("key") is None
class TestPluginMatcher:
# noinspection PyUnusedLocal

12
tests/test_session.py
View File

@ -340,18 +340,6 @@ class TestStreams:
assert "vod_alt2" in streams
def test_pluginoptions(session: Streamlink):
assert session.get_plugin_option("testplugin", "a_option") is None
session.load_plugins(str(PATH_TESTPLUGINS))
assert session.get_plugin_option("testplugin", "a_option") == "default"
session.set_plugin_option("testplugin", "another_option", "test")
assert session.get_plugin_option("testplugin", "another_option") == "test"
assert session.get_plugin_option("non_existing", "non_existing") is None
assert session.get_plugin_option("testplugin", "non_existing") is None
def test_options(session: Streamlink):
session.set_option("test_option", "option")
assert session.get_option("test_option") == "option"