From c2f062610019a3681fd3ededfdae310b925eac41 Mon Sep 17 00:00:00 2001 From: bastimeyer Date: Sun, 4 Feb 2024 15:22:10 +0100 Subject: [PATCH] session: move plugin load/resolve logic - Move plugin load/resolve logic into the `streamlink.session.plugins` module with the `StreamlinkPlugins` class - Make `Streamlink.plugins` an instance of this class - Add the `plugins_builtin` keyword argument to the `Streamlink` class and set its default value to `True` - Deprecate the `Streamlink.{{get,load}_plugins,load_builtin_plugins}()` methods, which are now wrappers for the respective `StreamlinkPlugins` implementations - Update `streamlink_cli.main` and all other `Streamlink` session usages accordingly (in scripts and tests) - Update session docs - Move and update plugin load/resolve tests --- docs/api/session.rst | 4 + script/test-plugin-urls.py | 2 +- src/streamlink/session/plugins.py | 135 ++++++++++++++++++++++++++ src/streamlink/session/session.py | 86 +++++++---------- src/streamlink_cli/main.py | 17 ++-- tests/cli/test_cmdline.py | 5 +- tests/cli/test_main.py | 12 +-- tests/conftest.py | 22 +++-- tests/mixins/stream_hls.py | 3 +- tests/plugins/test_ustreamtv.py | 4 +- tests/session/test_plugins.py | 154 ++++++++++++++++++++++++++++++ tests/session/test_session.py | 152 ++++++++++------------------- tests/test_plugin_userinput.py | 2 +- tests/test_streamlink_api.py | 2 +- 14 files changed, 412 insertions(+), 188 deletions(-) create mode 100644 src/streamlink/session/plugins.py create mode 100644 tests/session/test_plugins.py diff --git a/docs/api/session.rst b/docs/api/session.rst index b03c539f..9ef0b52b 100644 --- a/docs/api/session.rst +++ b/docs/api/session.rst @@ -8,3 +8,7 @@ Session .. automethod:: get .. automethod:: set + +.. autoclass:: streamlink.session.plugins.StreamlinkPlugins + :member-order: bysource + :special-members: __getitem__, __setitem__, __delitem__, __contains__ diff --git a/script/test-plugin-urls.py b/script/test-plugin-urls.py index 5dc23d72..546cc2ac 100755 --- a/script/test-plugin-urls.py +++ b/script/test-plugin-urls.py @@ -157,7 +157,7 @@ class PluginUrlTester: for url in sorted(self.urls): self.logger.info(f"Finding streams for URL: {url}") - session = Streamlink() + session = Streamlink(plugins_builtin=True) # noinspection PyBroadException try: pluginname, Pluginclass, _resolved_url = session.resolve_url(url) diff --git a/src/streamlink/session/plugins.py b/src/streamlink/session/plugins.py new file mode 100644 index 00000000..554627a9 --- /dev/null +++ b/src/streamlink/session/plugins.py @@ -0,0 +1,135 @@ +import logging +import pkgutil +from importlib.abc import PathEntryFinder +from pathlib import Path +from types import ModuleType +from typing import Dict, Iterator, List, Optional, Tuple, Type, Union + +import streamlink.plugins +from streamlink.options import Arguments + +# noinspection PyProtectedMember +from streamlink.plugin.plugin import NO_PRIORITY, Matchers, Plugin +from streamlink.utils.module import exec_module + + +log = logging.getLogger(".".join(__name__.split(".")[:-1])) + +# The path to Streamlink's built-in plugins +_PLUGINS_PATH = Path(streamlink.plugins.__path__[0]) + + +class StreamlinkPlugins: + """ + Streamlink's session-plugins implementation. This class is responsible for loading plugins and resolving them from URLs. + + See the :attr:`Streamlink.plugins ` attribute. + """ + + def __init__(self, builtin: bool = True): + self._plugins: Dict[str, Type[Plugin]] = {} + + if builtin: + self.load_builtin() + + def __getitem__(self, item: str) -> Type[Plugin]: + """Access a loaded plugin class by name""" + return self._plugins[item] + + def __setitem__(self, key: str, value: Type[Plugin]) -> None: + """Add/override a plugin class by name""" + self._plugins[key] = value + + def __delitem__(self, key: str) -> None: + """Remove a loaded plugin by name""" + self._plugins.pop(key, None) + + def __contains__(self, item: str) -> bool: + """Check if a plugin is loaded""" + return item in self._plugins + + def get_names(self) -> List[str]: + """Get a list of the names of all loaded plugins""" + return sorted(self._plugins.keys()) + + def get_loaded(self) -> Dict[str, Type[Plugin]]: + """Get a mapping of all loaded plugins""" + return dict(self._plugins) + + def load_builtin(self) -> bool: + """Load Streamlink's built-in plugins""" + return self.load_path(_PLUGINS_PATH) + + def load_path(self, path: Union[Path, str]) -> bool: + """Load plugins from a custom directory""" + plugins = self._load_from_path(path) + self.update(plugins) + + return bool(plugins) + + def update(self, plugins: Dict[str, Type[Plugin]]): + """Add/override loaded plugins""" + self._plugins.update(plugins) + + def clear(self): + """Remove all loaded plugins from the session""" + self._plugins.clear() + + def iter_arguments(self) -> Iterator[Tuple[str, Arguments]]: + """Iterate through all plugins and their :class:`Arguments `""" + yield from ( + (name, plugin.arguments) + for name, plugin in self._plugins.items() + if plugin.arguments + ) + + def iter_matchers(self) -> Iterator[Tuple[str, Matchers]]: + """Iterate through all plugins and their :class:`Matchers `""" + yield from ( + (name, plugin.matchers) + for name, plugin in self._plugins.items() + if plugin.matchers + ) + + def match_url(self, url: str) -> Optional[Tuple[str, Type[Plugin]]]: + """Find a matching plugin by URL""" + match: Optional[str] = None + priority: int = NO_PRIORITY + + for name, matchers in self.iter_matchers(): + for matcher in matchers: + if matcher.priority > priority and matcher.pattern.match(url) is not None: + match = name + priority = matcher.priority + + if match is None: + return None + + return match, self._plugins[match] + + def _load_from_path(self, path: Union[Path, str]) -> Dict[str, Type[Plugin]]: + plugins: Dict[str, Type[Plugin]] = {} + for finder, name, _ in pkgutil.iter_modules([str(path)]): + lookup = self._load_plugin_from_finder(name, finder=finder) # type: ignore[arg-type] + if lookup is None: + continue + mod, plugin = lookup + if name in self._plugins: + log.info(f"Plugin {name} is being overridden by {mod.__file__}") + plugins[name] = plugin + + return plugins + + @staticmethod + def _load_plugin_from_finder(name: str, finder: PathEntryFinder) -> Optional[Tuple[ModuleType, Type[Plugin]]]: + try: + # set the full plugin module name, even for sideloaded plugins + mod = exec_module(finder, f"streamlink.plugins.{name}") + except ImportError as err: + log.exception(f"Failed to load plugin {name} from {err.path}\n") + return None + + if not hasattr(mod, "__plugin__") or not issubclass(mod.__plugin__, Plugin): + return None + + return mod, mod.__plugin__ diff --git a/src/streamlink/session/session.py b/src/streamlink/session/session.py index 9dcc587b..82e1fcef 100644 --- a/src/streamlink/session/session.py +++ b/src/streamlink/session/session.py @@ -1,17 +1,17 @@ import logging -import pkgutil +import warnings from functools import lru_cache from typing import Any, Dict, Optional, Tuple, Type -from streamlink import __version__, plugins -from streamlink.exceptions import NoPluginError, PluginError +from streamlink import __version__ +from streamlink.exceptions import NoPluginError, PluginError, StreamlinkDeprecationWarning from streamlink.logger import StreamlinkLogger from streamlink.options import Options -from streamlink.plugin.plugin import NO_PRIORITY, Matcher, Plugin +from streamlink.plugin.plugin import Plugin from streamlink.session.http import HTTPSession from streamlink.session.options import StreamlinkOptions +from streamlink.session.plugins import StreamlinkPlugins from streamlink.utils.l10n import Localization -from streamlink.utils.module import exec_module from streamlink.utils.url import update_scheme @@ -28,9 +28,12 @@ class Streamlink: def __init__( self, options: Optional[Dict[str, Any]] = None, + *, + plugins_builtin: bool = True, ): """ :param options: Custom options + :param plugins_builtin: Whether to load built-in plugins or not """ #: An instance of Streamlink's :class:`requests.Session` subclass. @@ -43,8 +46,9 @@ class Streamlink: self.options: StreamlinkOptions = StreamlinkOptions(self) if options: self.options.update(options) - self.plugins: Dict[str, Type[Plugin]] = {} - self.load_builtin_plugins() + + #: Plugins of this session instance. + self.plugins: StreamlinkPlugins = StreamlinkPlugins(builtin=plugins_builtin) def set_option(self, key: str, value: Any) -> None: """ @@ -92,19 +96,8 @@ class Streamlink: """ url = update_scheme("https://", url, force=False) - - matcher: Matcher - candidate: Optional[Tuple[str, Type[Plugin]]] = None - priority = NO_PRIORITY - for name, plugin in self.plugins.items(): - if plugin.matchers: - for matcher in plugin.matchers: - if matcher.priority > priority and matcher.pattern.match(url) is not None: - candidate = name, plugin - priority = matcher.priority - - if candidate: - return candidate[0], candidate[1], url + if resolved := self.plugins.match_url(url): + return resolved[0], resolved[1], url if follow_redirect: # Attempt to handle a redirect URL @@ -151,42 +144,31 @@ class Streamlink: return plugin.streams(**params) def get_plugins(self): - """Returns the loaded plugins for the session.""" - - return self.plugins + """Returns the loaded plugins of this session (deprecated)""" + warnings.warn( + "`Streamlink.get_plugins()` has been deprecated in favor of `Streamlink.plugins.get_loaded()`", + StreamlinkDeprecationWarning, + stacklevel=2, + ) + return self.plugins.get_loaded() def load_builtin_plugins(self): - self.load_plugins(plugins.__path__[0]) + """Loads Streamlink's built-in plugins (deprecated)""" + warnings.warn( + "`Streamlink.load_builtin_plugins()` has been deprecated in favor of the `plugins_builtin` keyword argument", + StreamlinkDeprecationWarning, + stacklevel=2, + ) + self.plugins.load_builtin() def load_plugins(self, path: str) -> bool: - """ - Attempt to load plugins from the path specified. - - :param path: full path to a directory where to look for plugins - :return: success - """ - - success = False - for module_info in pkgutil.iter_modules([path]): - name = module_info.name - # set the full plugin module name - # use the "streamlink.plugins." prefix even for sideloaded plugins - module_name = f"streamlink.plugins.{name}" - try: - mod = exec_module(module_info.module_finder, module_name) # type: ignore[arg-type] - except ImportError as err: - log.exception(f"Failed to load plugin {name} from {path}", exc_info=err) - continue - - if not hasattr(mod, "__plugin__") or not issubclass(mod.__plugin__, Plugin): - continue - success = True - plugin = mod.__plugin__ - if name in self.plugins: - log.debug(f"Plugin {name} is being overridden by {mod.__file__}") - self.plugins[name] = plugin - - return success + """Loads plugins from a specific path (deprecated)""" + warnings.warn( + "`Streamlink.load_plugins()` has been deprecated in favor of `Streamlink.plugins.load_path()`", + StreamlinkDeprecationWarning, + stacklevel=2, + ) + return self.plugins.load_path(path) @property def version(self): diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py index 2be41965..e485a55a 100644 --- a/src/streamlink_cli/main.py +++ b/src/streamlink_cli/main.py @@ -602,20 +602,19 @@ def handle_url(): def print_plugins(): """Outputs a list of all plugins Streamlink has loaded.""" - pluginlist = list(streamlink.get_plugins().keys()) - pluginlist_formatted = ", ".join(sorted(pluginlist)) + pluginlist = streamlink.plugins.get_names() if args.json: console.msg_json(pluginlist) else: - console.msg(f"Loaded plugins: {pluginlist_formatted}") + console.msg(f"Loaded plugins: {', '.join(pluginlist)}") def load_plugins(dirs: List[Path], showwarning: bool = True): """Attempts to load plugins from a list of directories.""" for directory in dirs: if directory.is_dir(): - success = streamlink.load_plugins(str(directory)) + success = streamlink.plugins.load_path(directory) if success and type(directory) is DeprecatedPath: warnings.warn( f"Loaded plugins from deprecated path, see CLI docs for how to migrate: {directory}", @@ -724,10 +723,10 @@ def setup_plugin_args(session: Streamlink, parser: ArgumentParser): """Adds plugin argument data to the argument parser.""" plugin_args = parser.add_argument_group("Plugin options") - for pname, plugin in session.plugins.items(): + for pname, arguments in session.plugins.iter_arguments(): group = parser.add_argument_group(pname.capitalize(), parent=plugin_args) - for parg in plugin.arguments or []: + for parg in arguments: group.add_argument(parg.argument_name(pname), **parg.options) @@ -819,8 +818,8 @@ def log_current_arguments(session: Streamlink, parser: argparse.ArgumentParser): return sensitive = set() - for pname, plugin in session.plugins.items(): - for parg in plugin.arguments or []: + for pname, arguments in session.plugins.iter_arguments(): + for parg in arguments: if parg.sensitive: sensitive.add(parg.argument_name(pname)) @@ -956,7 +955,7 @@ def main(): def parser_helper(): - session = Streamlink() + session = Streamlink(plugins_builtin=True) parser = build_parser() setup_plugin_args(session, parser) return parser diff --git a/tests/cli/test_cmdline.py b/tests/cli/test_cmdline.py index b1bf7e38..1e696cf8 100644 --- a/tests/cli/test_cmdline.py +++ b/tests/cli/test_cmdline.py @@ -21,9 +21,8 @@ class CommandLineTestCase(unittest.TestCase): return fn - with patch("streamlink.session.Streamlink.load_builtin_plugins"): - session = Streamlink() - session.load_plugins(str(Path(tests.__path__[0]) / "plugin")) + session = Streamlink(plugins_builtin=False) + session.plugins.load_path(Path(tests.__path__[0]) / "plugin") actual_exit_code = 0 with patch("sys.argv") as mock_argv, \ diff --git a/tests/cli/test_main.py b/tests/cli/test_main.py index b120e41b..11b8c49a 100644 --- a/tests/cli/test_main.py +++ b/tests/cli/test_main.py @@ -494,9 +494,8 @@ class _TestCLIMainLogging(unittest.TestCase): @classmethod def subject(cls, argv, **kwargs): - with patch("streamlink.session.Streamlink.load_builtin_plugins"): - session = Streamlink() - session.load_plugins(str(Path(tests.__path__[0]) / "plugin")) + session = Streamlink(plugins_builtin=False) + session.plugins.load_path(Path(tests.__path__[0]) / "plugin") with patch("streamlink_cli.main.os.geteuid", create=True, new=Mock(return_value=kwargs.get("euid", 1000))), \ patch("streamlink_cli.main.streamlink", session), \ @@ -835,11 +834,10 @@ class TestCLIMainLoggingLogfileWindows(_TestCLIMainLogging): class TestCLIMainPrint(unittest.TestCase): def subject(self): - with patch.object(Streamlink, "load_builtin_plugins"), \ - patch.object(Streamlink, "resolve_url") as mock_resolve_url, \ + with patch.object(Streamlink, "resolve_url") as mock_resolve_url, \ patch.object(Streamlink, "resolve_url_no_redirect") as mock_resolve_url_no_redirect: - session = Streamlink() - session.load_plugins(str(Path(tests.__path__[0]) / "plugin")) + session = Streamlink(plugins_builtin=False) + session.plugins.load_path(Path(tests.__path__[0]) / "plugin") with patch("streamlink_cli.main.os.geteuid", create=True, new=Mock(return_value=1000)), \ patch("streamlink_cli.main.streamlink", session), \ patch("streamlink_cli.main.CONFIG_FILES", []), \ diff --git a/tests/conftest.py b/tests/conftest.py index c7d4be64..df1cb252 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,6 @@ import os import sys -from typing import Dict, Iterator, List, Tuple -from unittest.mock import patch +from typing import Dict, List, Tuple import pytest import requests_mock as rm @@ -68,14 +67,19 @@ def _check_test_condition(item: pytest.Item): # pragma: no cover @pytest.fixture() -def session(request: pytest.FixtureRequest) -> Iterator[Streamlink]: - with patch.object(Streamlink, "load_builtin_plugins"): - session = Streamlink() - for key, value in getattr(request, "param", {}).items(): - session.set_option(key, value) - yield session +def session(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch): + options = getattr(request, "param", {}) + plugins_builtin = options.pop("plugins-builtin", False) - Streamlink.resolve_url.cache_clear() + session = Streamlink( + options=options, + plugins_builtin=plugins_builtin, + ) + + try: + yield session + finally: + Streamlink.resolve_url.cache_clear() @pytest.fixture() diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py index 4cb10874..e6d1b884 100644 --- a/tests/mixins/stream_hls.py +++ b/tests/mixins/stream_hls.py @@ -280,8 +280,7 @@ class TestMixinStreamHLS(unittest.TestCase): return data def get_session(self, options=None, *args, **kwargs): - with patch("streamlink.session.Streamlink.load_builtin_plugins"): - return Streamlink(options) + return Streamlink(options, plugins_builtin=False) # set up HLS responses, create the session and read thread and start it def subject(self, playlists, options=None, streamoptions=None, threadoptions=None, start=True, *args, **kwargs): diff --git a/tests/plugins/test_ustreamtv.py b/tests/plugins/test_ustreamtv.py index 411b1e19..5fd1571a 100644 --- a/tests/plugins/test_ustreamtv.py +++ b/tests/plugins/test_ustreamtv.py @@ -60,9 +60,7 @@ class TestPluginUStreamTV: plugins = parser.add_argument_group("Plugin Options") group = parser.add_argument_group("UStreamTV", parent=plugins) - session.plugins = { - "ustreamtv": UStreamTV, - } + session.plugins["ustreamtv"] = UStreamTV setup_plugin_args(session, parser) diff --git a/tests/session/test_plugins.py b/tests/session/test_plugins.py new file mode 100644 index 00000000..e713f2c8 --- /dev/null +++ b/tests/session/test_plugins.py @@ -0,0 +1,154 @@ +from pathlib import Path +from typing import Type +from unittest.mock import Mock, call + +import pytest + +import streamlink.plugins +import tests.plugin +from streamlink.plugin.plugin import Plugin, pluginargument +from streamlink.session import Streamlink +from streamlink.session.plugins import StreamlinkPlugins + + +PATH_BUILTINPLUGINS = Path(streamlink.plugins.__path__[0]) +PATH_TESTPLUGINS = Path(tests.plugin.__path__[0]) +PATH_TESTPLUGINS_OVERRIDE = PATH_TESTPLUGINS / "override" + + +@pytest.fixture(autouse=True) +def caplog(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture: + caplog.set_level(1, "streamlink") + return caplog + + +@pytest.fixture(scope="module") +def fake_plugin(): + @pluginargument("foo") + @pluginargument("bar") + class FakePlugin(Plugin): + __module__ = "streamlink.plugins.fake" + + def _get_streams(self): pass # pragma: no cover + + return FakePlugin + + +def test_empty(caplog: pytest.LogCaptureFixture, session: Streamlink): + assert session.plugins.get_names() == [] + assert session.plugins.get_loaded() == {} + assert caplog.record_tuples == [] + + +def test_set_get_del(session: Streamlink, fake_plugin: Type[Plugin]): + assert "fake" not in session.plugins + + session.plugins["fake"] = fake_plugin + assert "fake" in session.plugins + assert session.plugins["fake"] is fake_plugin + assert session.plugins.get_names() == ["fake"] + assert session.plugins.get_loaded() == {"fake": fake_plugin} + assert session.plugins.get_loaded() is not session.plugins.get_loaded() + + del session.plugins["fake"] + assert "fake" not in session.plugins + assert session.plugins.get_names() == [] + assert session.plugins.get_loaded() == {} + + +def test_update_clear(session: Streamlink, fake_plugin: Type[Plugin]): + assert "fake" not in session.plugins + + session.plugins.update({"fake": fake_plugin}) + assert "fake" in session.plugins + assert session.plugins["fake"] is fake_plugin + assert session.plugins.get_names() == ["fake"] + assert session.plugins.get_loaded() == {"fake": fake_plugin} + + session.plugins.clear() + assert "fake" not in session.plugins + assert session.plugins.get_names() == [] + assert session.plugins.get_loaded() == {} + + +def test_iter_arguments(session: Streamlink, fake_plugin: Type[Plugin]): + session.plugins.update({"fake": fake_plugin}) + assert [(name, [arg.argument_name(name) for arg in args]) for name, args in session.plugins.iter_arguments()] == [ + ("fake", ["--fake-foo", "--fake-bar"]), + ] + + +class TestLoad: + def test_load_builtin(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, fake_plugin: Type[Plugin]): + mock = Mock(return_value={"fake": fake_plugin}) + monkeypatch.setattr(StreamlinkPlugins, "_load_from_path", mock) + session = Streamlink(plugins_builtin=True) + + assert mock.call_args_list == [call(PATH_BUILTINPLUGINS)] + assert "fake" in session.plugins + assert session.plugins.get_names() == ["fake"] + assert session.plugins.get_loaded() == {"fake": fake_plugin} + assert session.plugins["fake"].__module__ == "streamlink.plugins.fake" + assert caplog.record_tuples == [] + + def test_load_path_empty(self, tmp_path: Path, caplog: pytest.LogCaptureFixture, session: Streamlink): + assert not session.plugins.load_path(tmp_path) + assert session.plugins.get_names() == [] + assert session.plugins.get_loaded() == {} + assert caplog.record_tuples == [] + + def test_load_path_testplugins(self, caplog: pytest.LogCaptureFixture, session: Streamlink): + assert session.plugins.load_path(PATH_TESTPLUGINS) + assert "testplugin" in session.plugins + assert "testplugin_invalid" not in session.plugins + assert "testplugin_missing" not in session.plugins + assert session.plugins.get_names() == ["testplugin"] + assert session.plugins["testplugin"].__name__ == "TestPlugin" + assert session.plugins["testplugin"].__module__ == "streamlink.plugins.testplugin" + assert caplog.record_tuples == [] + + assert session.plugins.load_path(PATH_TESTPLUGINS_OVERRIDE) + assert "testplugin" in session.plugins + assert session.plugins.get_names() == ["testplugin"] + assert session.plugins["testplugin"].__name__ == "TestPluginOverride" + assert session.plugins["testplugin"].__module__ == "streamlink.plugins.testplugin" + assert [(record.name, record.levelname, record.message) for record in caplog.records] == [ + ( + "streamlink.session", + "info", + f"Plugin testplugin is being overridden by {PATH_TESTPLUGINS_OVERRIDE / 'testplugin.py'}", + ), + ] + + def test_importerror(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, session: Streamlink): + monkeypatch.setattr("importlib.machinery.FileFinder.find_spec", Mock(return_value=None)) + assert not session.plugins.load_path(PATH_TESTPLUGINS) + assert "testplugin" not in session.plugins + assert session.plugins.get_names() == [] + assert [(record.name, record.levelname, record.message, bool(record.exc_info)) for record in caplog.records] == [ + ( + "streamlink.session", + "error", + f"Failed to load plugin testplugin from {PATH_TESTPLUGINS}\n", + True, + ), + ( + "streamlink.session", + "error", + f"Failed to load plugin testplugin_invalid from {PATH_TESTPLUGINS}\n", + True, + ), + ( + "streamlink.session", + "error", + f"Failed to load plugin testplugin_missing from {PATH_TESTPLUGINS}\n", + True, + ), + ] + + def test_syntaxerror(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, session: Streamlink): + monkeypatch.setattr("importlib.machinery.SourceFileLoader.exec_module", Mock(side_effect=SyntaxError)) + with pytest.raises(SyntaxError): + session.plugins.load_path(PATH_TESTPLUGINS) + assert session.plugins.get_names() == [] + assert caplog.record_tuples == [] diff --git a/tests/session/test_session.py b/tests/session/test_session.py index ed926cbe..54a9ccbd 100644 --- a/tests/session/test_session.py +++ b/tests/session/test_session.py @@ -1,5 +1,4 @@ import re -from contextlib import nullcontext from pathlib import Path from unittest.mock import Mock @@ -7,7 +6,7 @@ import pytest import requests_mock as rm import tests.plugin -from streamlink.exceptions import NoPluginError +from streamlink.exceptions import NoPluginError, StreamlinkDeprecationWarning from streamlink.options import Options from streamlink.plugin import HIGH_PRIORITY, LOW_PRIORITY, NO_PRIORITY, NORMAL_PRIORITY, Plugin, pluginmatcher from streamlink.session import Streamlink @@ -16,92 +15,47 @@ from streamlink.stream.http import HTTPStream PATH_TESTPLUGINS = Path(tests.plugin.__path__[0]) -PATH_TESTPLUGINS_OVERRIDE = PATH_TESTPLUGINS / "override" -class TestLoadPlugins: - @pytest.fixture(autouse=True) - def caplog(self, caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture: - caplog.set_level(1, "streamlink") - return caplog - - def test_load_plugins(self, caplog: pytest.LogCaptureFixture, session: Streamlink): - session.load_plugins(str(PATH_TESTPLUGINS)) - plugins = session.get_plugins() - assert list(plugins.keys()) == ["testplugin"] - assert plugins["testplugin"].__name__ == "TestPlugin" - assert plugins["testplugin"].__module__ == "streamlink.plugins.testplugin" - assert caplog.records == [] - - def test_load_plugins_override(self, caplog: pytest.LogCaptureFixture, session: Streamlink): - session.load_plugins(str(PATH_TESTPLUGINS)) - session.load_plugins(str(PATH_TESTPLUGINS_OVERRIDE)) - plugins = session.get_plugins() - assert list(plugins.keys()) == ["testplugin"] - assert plugins["testplugin"].__name__ == "TestPluginOverride" - assert plugins["testplugin"].__module__ == "streamlink.plugins.testplugin" - assert [(record.name, record.levelname, record.message) for record in caplog.records] == [ +class TestPluginsDeprecations: + def test_get_plugins(self, monkeypatch: pytest.MonkeyPatch, recwarn: pytest.WarningsRecorder, session: Streamlink): + mock = Mock(return_value={}) + monkeypatch.setattr(session.plugins, "get_loaded", mock) + assert session.get_plugins() == {} + assert mock.call_count == 1 + assert [(record.filename, record.category, str(record.message)) for record in recwarn.list] == [ ( - "streamlink.session", - "debug", - f"Plugin testplugin is being overridden by {PATH_TESTPLUGINS_OVERRIDE / 'testplugin.py'}", + __file__, + StreamlinkDeprecationWarning, + "`Streamlink.get_plugins()` has been deprecated in favor of `Streamlink.plugins.get_loaded()`", ), ] - def test_load_plugins_builtin(self): - session = Streamlink() - plugins = session.get_plugins() - assert "twitch" in plugins - assert plugins["twitch"].__module__ == "streamlink.plugins.twitch" + def test_load_builtin_plugins(self, monkeypatch: pytest.MonkeyPatch, recwarn: pytest.WarningsRecorder, session: Streamlink): + mock = Mock(return_value={}) + monkeypatch.setattr(session.plugins, "load_builtin", mock) + assert session.load_builtin_plugins() is None + assert mock.call_count == 1 + assert [(record.filename, record.category, str(record.message)) for record in recwarn.list] == [ + ( + __file__, + StreamlinkDeprecationWarning, + "`Streamlink.load_builtin_plugins()` has been deprecated in favor of the `plugins_builtin` keyword argument", + ), + ] - @pytest.mark.parametrize(("side_effect", "raises", "logs"), [ - pytest.param( - ImportError, - nullcontext(), - [ - ( - "streamlink.session", - "error", - f"Failed to load plugin testplugin from {PATH_TESTPLUGINS}", - True, - ), - ( - "streamlink.session", - "error", - f"Failed to load plugin testplugin_invalid from {PATH_TESTPLUGINS}", - True, - ), - ( - "streamlink.session", - "error", - f"Failed to load plugin testplugin_missing from {PATH_TESTPLUGINS}", - True, - ), - ], - id="ImportError", - ), - pytest.param( - SyntaxError, - pytest.raises(SyntaxError), - [], - id="SyntaxError", - ), - ]) - def test_load_plugins_failure( - self, - monkeypatch: pytest.MonkeyPatch, - caplog: pytest.LogCaptureFixture, - side_effect: Exception, - raises: nullcontext, - logs: list, - ): - monkeypatch.setattr("streamlink.session.session.Streamlink.load_builtin_plugins", Mock()) - monkeypatch.setattr("streamlink.session.session.exec_module", Mock(side_effect=side_effect)) - session = Streamlink() - with raises: - session.load_plugins(str(PATH_TESTPLUGINS)) - assert session.get_plugins() == {} - assert [(record.name, record.levelname, record.message, bool(record.exc_info)) for record in caplog.records] == logs + def test_load_plugins(self, recwarn: pytest.WarningsRecorder, session: Streamlink): + session.load_plugins(str(PATH_TESTPLUGINS)) + assert session.plugins.get_names() == ["testplugin"] + assert session.plugins["testplugin"].__name__ == "TestPlugin" + assert session.plugins["testplugin"].__module__ == "streamlink.plugins.testplugin" + assert [(record.filename, record.category, str(record.message)) for record in recwarn.list] == [ + ( + __file__, + StreamlinkDeprecationWarning, + "`Streamlink.load_plugins()` has been deprecated in favor of `Streamlink.plugins.load_path()`", + ), + ] class _EmptyPlugin(Plugin): @@ -111,24 +65,23 @@ class _EmptyPlugin(Plugin): class TestResolveURL: @pytest.fixture(autouse=True) - def _load_builtins(self, session: Streamlink): - session.load_plugins(str(PATH_TESTPLUGINS)) + def _load_plugins(self, session: Streamlink): + session.plugins.load_path(PATH_TESTPLUGINS) @pytest.fixture(autouse=True) def requests_mock(self, requests_mock: rm.Mocker): return requests_mock def test_resolve_url(self, recwarn: pytest.WarningsRecorder, session: Streamlink): - plugins = session.get_plugins() _pluginname, pluginclass, resolved_url = session.resolve_url("http://test.se/channel") assert issubclass(pluginclass, Plugin) - assert pluginclass is plugins["testplugin"] + assert pluginclass is session.plugins["testplugin"] assert resolved_url == "http://test.se/channel" assert hasattr(session.resolve_url, "cache_info"), "resolve_url has a lookup cache" assert recwarn.list == [] - def test_resolve_url__noplugin(self, requests_mock: rm.Mocker, session: Streamlink): + def test_resolve_url_noplugin(self, requests_mock: rm.Mocker, session: Streamlink): requests_mock.get("http://invalid2", status_code=301, headers={"Location": "http://invalid3"}) with pytest.raises(NoPluginError): @@ -136,26 +89,24 @@ class TestResolveURL: with pytest.raises(NoPluginError): session.resolve_url("http://invalid2") - def test_resolve_url__redirected(self, requests_mock: rm.Mocker, session: Streamlink): + def test_resolve_url_redirected(self, requests_mock: rm.Mocker, session: Streamlink): requests_mock.request("HEAD", "http://redirect1", status_code=501) requests_mock.request("GET", "http://redirect1", status_code=301, headers={"Location": "http://redirect2"}) requests_mock.request("GET", "http://redirect2", status_code=301, headers={"Location": "http://test.se/channel"}) requests_mock.request("GET", "http://test.se/channel", content=b"") - plugins = session.get_plugins() _pluginname, pluginclass, resolved_url = session.resolve_url("http://redirect1") assert issubclass(pluginclass, Plugin) - assert pluginclass is plugins["testplugin"] + assert pluginclass is session.plugins["testplugin"] assert resolved_url == "http://test.se/channel" def test_resolve_url_no_redirect(self, session: Streamlink): - plugins = session.get_plugins() _pluginname, pluginclass, resolved_url = session.resolve_url_no_redirect("http://test.se/channel") assert issubclass(pluginclass, Plugin) - assert pluginclass is plugins["testplugin"] + assert pluginclass is session.plugins["testplugin"] assert resolved_url == "http://test.se/channel" - def test_resolve_url_no_redirect__noplugin(self, session: Streamlink): + def test_resolve_url_no_redirect_noplugin(self, session: Streamlink): with pytest.raises(NoPluginError): session.resolve_url_no_redirect("http://invalid") @@ -168,10 +119,10 @@ class TestResolveURL: class PluginHttps(_EmptyPlugin): pass - session.plugins = { + session.plugins.update({ "insecure": PluginHttp, "secure": PluginHttps, - } + }) with pytest.raises(NoPluginError): session.resolve_url("insecure") @@ -209,12 +160,12 @@ class TestResolveURL: class NoPriority(_EmptyPlugin): pass - session.plugins = { + session.plugins.update({ "high": HighPriority, "normal": NormalPriority, "low": LowPriority, "no": NoPriority, - } + }) no = session.resolve_url_no_redirect("no")[1] low = session.resolve_url_no_redirect("low")[1] normal = session.resolve_url_no_redirect("normal")[1] @@ -226,17 +177,18 @@ class TestResolveURL: assert high is HighPriority session.resolve_url.cache_clear() - session.plugins = { + session.plugins.clear() + session.plugins.update({ "no": NoPriority, - } + }) with pytest.raises(NoPluginError): session.resolve_url_no_redirect("no") class TestStreams: @pytest.fixture(autouse=True) - def _load_builtins(self, session: Streamlink): - session.load_plugins(str(PATH_TESTPLUGINS)) + def _load_plugins(self, session: Streamlink): + session.plugins.load_path(PATH_TESTPLUGINS) def test_streams(self, session: Streamlink): streams = session.streams("http://test.se/channel") diff --git a/tests/test_plugin_userinput.py b/tests/test_plugin_userinput.py index 4bd935d0..bc1b3fdd 100644 --- a/tests/test_plugin_userinput.py +++ b/tests/test_plugin_userinput.py @@ -10,7 +10,7 @@ from tests.plugin.testplugin import TestPlugin as _TestPlugin def test_session(): console_input = ConsoleUserInputRequester(Mock()) - session = Streamlink({"user-input-requester": console_input}) + session = Streamlink({"user-input-requester": console_input}, plugins_builtin=False) assert session.get_option("user-input-requester") is console_input diff --git a/tests/test_streamlink_api.py b/tests/test_streamlink_api.py index c8589fdf..51480f50 100644 --- a/tests/test_streamlink_api.py +++ b/tests/test_streamlink_api.py @@ -9,7 +9,7 @@ class TestStreamlinkAPI: @pytest.fixture(autouse=True) def _session(self, monkeypatch: pytest.MonkeyPatch, session: Streamlink): monkeypatch.setattr("streamlink.api.Streamlink", lambda: session) - session.load_plugins(tests.plugin.__path__[0]) + session.plugins.load_path(tests.plugin.__path__[0]) def test_find_test_plugin(self): assert "hls" in streams("test.se")