mirrors/yt-dlp
1
0
Fork 0
mirror of https://github.com/yt-dlp/yt-dlp.git synced 2025-08-16 09:28:28 +00:00
Code Issues Projects Releases Wiki Activity

add jsi plugin testcase like pp test

Browse Source
This commit is contained in:
c-basalt 2025-05-02 03:35:56 -04:00
parent fbd5ce74c0
commit d74f921e37
9 changed files with 102 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
test/test_jsi_external.py
View File

@ -57,6 +57,8 @@ def inner(func: typing.Callable[[unittest.TestCase, type[JSI]], None]):
def wrapper(self: unittest.TestCase):
for key, jsi in get_included_jsi(exclude=exclude).items():
def wrapped_jsi_with_unavaliable_auto_skip(*args, **kwargs):
if getattr(jsi, 'TEST_DATA_PLUGIN', False):
self.skipTest('Testdata plugin')
instance = jsi(*args, **kwargs)
if not instance.is_available():
self.skipTest(f'{key} is not available')

39
test/test_plugins.py
View File

@ -22,9 +22,11 @@
from yt_dlp.globals import (
extractors,
postprocessors,
jsi_runtimes,
plugin_dirs,
plugin_ies,
plugin_pps,
plugin_jsis,
all_plugins_loaded,
plugin_specs,
)
@ -44,16 +46,24 @@
plugin_destination=plugin_pps,
)
JSI_PLUGIN_SPEC = PluginSpec(
module_name='jsinterp',
suffix='JSI',
destination=jsi_runtimes,
plugin_destination=plugin_jsis,
)
def reset_plugins():
plugin_ies.value = {}
plugin_pps.value = {}
plugin_jsis.value = {}
plugin_dirs.value = ['default']
plugin_specs.value = {}
all_plugins_loaded.value = False
# Clearing override plugins is probably difficult
for module_name in tuple(sys.modules):
for plugin_type in ('extractor', 'postprocessor'):
for plugin_type in ('extractor', 'postprocessor', 'jsinterp'):
if module_name.startswith(f'{PACKAGE_NAME}.{plugin_type}.'):
del sys.modules[module_name]
@ -108,6 +118,12 @@ def test_postprocessor_classes(self):
self.assertIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
self.assertIn('NormalPluginPP', plugin_pps.value)
def test_jsi_runtime_classes(self):
plugins_jsi = load_plugins(JSI_PLUGIN_SPEC)
self.assertIn('NormalPluginJSI', plugins_jsi.keys())
self.assertIn(f'{PACKAGE_NAME}.jsinterp.normal', sys.modules.keys())
self.assertIn('NormalPluginJSI', plugin_jsis.value)
def test_importing_zipped_module(self):
zip_path = TEST_DATA_DIR / 'zipped_plugins.zip'
shutil.make_archive(str(zip_path)[:-4], 'zip', str(zip_path)[:-4])
@ -125,6 +141,9 @@ def test_importing_zipped_module(self):
plugins_pp = load_plugins(POSTPROCESSOR_PLUGIN_SPEC)
self.assertIn('ZippedPluginPP', plugins_pp.keys())
plugins_jsi = load_plugins(JSI_PLUGIN_SPEC)
self.assertIn('ZippedPluginJSI', plugins_jsi.keys())
finally:
sys.path.remove(str(zip_path))
os.remove(zip_path)
@ -134,13 +153,14 @@ def test_reloading_plugins(self):
reload_plugins_path = TEST_DATA_DIR / 'reload_plugins'
load_plugins(EXTRACTOR_PLUGIN_SPEC)
load_plugins(POSTPROCESSOR_PLUGIN_SPEC)
load_plugins(JSI_PLUGIN_SPEC)
# Remove default folder and add reload_plugin path
sys.path.remove(str(TEST_DATA_DIR))
sys.path.append(str(reload_plugins_path))
importlib.invalidate_caches()
try:
for plugin_type in ('extractor', 'postprocessor'):
for plugin_type in ('extractor', 'postprocessor', 'jsinterp'):
package = importlib.import_module(f'{PACKAGE_NAME}.{plugin_type}')
self.assertIn(reload_plugins_path / PACKAGE_NAME / plugin_type, map(Path, package.__path__))
@ -161,6 +181,14 @@ def test_reloading_plugins(self):
postprocessors.value['NormalPluginPP'].REPLACED,
msg='Reloading has not replaced original postprocessor plugin globally')
plugins_jsi = load_plugins(JSI_PLUGIN_SPEC)
self.assertIn('NormalPluginJSI', plugins_jsi.keys())
self.assertTrue(plugins_jsi['NormalPluginJSI'].REPLACED,
msg='Reloading has not replaced original postprocessor plugin')
self.assertTrue(
jsi_runtimes.value['NormalPluginJSI'].REPLACED,
msg='Reloading has not replaced original postprocessor plugin globally')
finally:
sys.path.remove(str(reload_plugins_path))
sys.path.append(str(TEST_DATA_DIR))
@ -188,24 +216,29 @@ def test_load_all_plugin_types(self):
self.assertNotIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
self.assertNotIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
self.assertNotIn(f'{PACKAGE_NAME}.jsinterp.normal', sys.modules.keys())
register_plugin_spec(EXTRACTOR_PLUGIN_SPEC)
register_plugin_spec(POSTPROCESSOR_PLUGIN_SPEC)
register_plugin_spec(JSI_PLUGIN_SPEC)
load_all_plugins()
self.assertTrue(all_plugins_loaded.value)
self.assertIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
self.assertIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
self.assertIn(f'{PACKAGE_NAME}.jsinterp.normal', sys.modules.keys())
def test_no_plugin_dirs(self):
register_plugin_spec(EXTRACTOR_PLUGIN_SPEC)
register_plugin_spec(POSTPROCESSOR_PLUGIN_SPEC)
register_plugin_spec(JSI_PLUGIN_SPEC)
plugin_dirs.value = []
load_all_plugins()
self.assertNotIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
self.assertNotIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
self.assertNotIn(f'{PACKAGE_NAME}.jsinterp.normal', sys.modules.keys())
def test_set_plugin_dirs(self):
custom_plugin_dir = str(TEST_DATA_DIR / 'plugin_packages')
@ -236,9 +269,11 @@ def test_append_plugin_dirs(self):
def test_get_plugin_spec(self):
register_plugin_spec(EXTRACTOR_PLUGIN_SPEC)
register_plugin_spec(POSTPROCESSOR_PLUGIN_SPEC)
register_plugin_spec(JSI_PLUGIN_SPEC)
self.assertEqual(plugin_specs.value.get('extractor'), EXTRACTOR_PLUGIN_SPEC)
self.assertEqual(plugin_specs.value.get('postprocessor'), POSTPROCESSOR_PLUGIN_SPEC)
self.assertEqual(plugin_specs.value.get('jsinterp'), JSI_PLUGIN_SPEC)
self.assertIsNone(plugin_specs.value.get('invalid'))

5
test/testdata/reload_plugins/yt_dlp_plugins/jsinterp/normal.py vendored Normal file
View File

@ -0,0 +1,5 @@
from yt_dlp.jsinterp.common import JSI
class NormalPluginJSI(JSI):
REPLACED = True

6
test/testdata/yt_dlp_plugins/jsinterp/normal.py vendored Normal file
View File

@ -0,0 +1,6 @@
from yt_dlp.jsinterp.common import JSI
class NormalPluginJSI(JSI):
TEST_DATA_PLUGIN = True
REPLACED = False

5
test/testdata/zipped_plugins/yt_dlp_plugins/jsinterp/zipped.py vendored Normal file
View File

@ -0,0 +1,5 @@
from yt_dlp.jsinterp.common import JSI
class ZippedPluginJSI(JSI):
pass

5
yt_dlp/jsinterp/__init__.py
View File

@ -12,13 +12,12 @@
if name.endswith('JSI')
})
plugin_spec = PluginSpec(
register_plugin_spec(PluginSpec(
module_name='jsinterp',
suffix='JSI',
destination=jsi_runtimes,
plugin_destination=plugin_jsis,
)
register_plugin_spec(plugin_spec)
))
__all__ = [
JSInterpreter,

19
yt_dlp/jsinterp/_deno.py
View File

@ -3,6 +3,7 @@
import http.cookiejar
import json
import platform
import re
import subprocess
import typing
import urllib.parse
@ -59,8 +60,8 @@ def execute(self, jscode, video_id=None, note='Executing JS in Deno'):
class DenoJSDomJSI(DenoJSI):
_BASE_PREFERENCE = 4
_DENO_FLAGS = ['--cached-only', '--no-prompt', '--no-check']
_JSDOM_IMPORT_CHECKED = False
_JSDOM_URL = 'https://esm.sh/v135/jsdom' # force use esm v135, esm-dev/esm.sh#1034
_JSDOM_VERSION = None
_JSDOM_URL = 'https://esm.sh/v135/jsdom' # force use esm v135, see esm-dev/esm.sh #1034
@staticmethod
def serialize_cookie(cookiejar: YoutubeDLCookieJar | None, url: str):
@ -106,10 +107,18 @@ def apply_cookies(cookiejar: YoutubeDLCookieJar | None, cookies: list[dict]):
False, None, None, {}))
def _ensure_jsdom(self):
if self._JSDOM_IMPORT_CHECKED:
if self._JSDOM_VERSION:
return
self._run_deno([self.exe, 'cache', self._JSDOM_URL])
self._JSDOM_IMPORT_CHECKED = True
# `--allow-import` is unsupported in v1, and esm.sh:443 is default allowed remote host for v2
result = self._run_deno([self.exe, 'info', self._JSDOM_URL])
version_line = next((line for line in result.splitlines() if self._JSDOM_URL in line), '')
if m := re.search(r'@([\d\.]+)', version_line):
self._JSDOM_VERSION = m[1]
def report_version(self):
super().report_version()
self._ensure_jsdom()
self.write_debug(f'JSDOM lib version {self._JSDOM_VERSION}')
def execute(self, jscode, video_id=None, note='Executing JS in Deno with jsdom', html='', cookiejar=None):
self.report_note(video_id, note)

2
yt_dlp/jsinterp/_helper.py
View File

@ -115,7 +115,7 @@ def extract_script_tags(html: str) -> tuple[str, list[str]]:
def prepare_wasm_jsmodule(js_mod: str, wasm: bytes) -> str:
"""
Sanitize js wrapper module generated by rust wasm-pack for wasm init
removes export and import.meta and inlines wasm binary as Uint8Array
Removes export and import.meta, and inlines wasm binary as Uint8Array
See test/test_data/jsi_external/hello_wasm.js for example
@param {str} js_mod: js wrapper module generated by rust wasm-pack

45
yt_dlp/jsinterp/common.py
View File

@ -1,8 +1,8 @@
from __future__ import annotations
import abc
import typing
import inspect
import typing
from ..globals import jsi_runtimes
from ..extractor.common import InfoExtractor
@ -19,7 +19,7 @@
_JSI_PREFERENCES: set[JSIPreference] = set()
def all_handlers() -> dict[str, type[JSI]]:
def get_all_handlers() -> dict[str, type[JSI]]:
return {jsi.JSI_KEY: jsi for jsi in jsi_runtimes.value.values()}
@ -29,13 +29,14 @@ def to_jsi_keys(jsi_or_keys: typing.Iterable[str | type[JSI] | JSI]) -> list[str
def get_included_jsi(only_include=None, exclude=None):
return {
key: value for key, value in all_handlers().items()
key: value for key, value in get_all_handlers().items()
if (not only_include or key in to_jsi_keys(only_include))
and (not exclude or key not in to_jsi_keys(exclude))
}
def order_to_pref(jsi_order: typing.Iterable[str | type[JSI] | JSI], multiplier: int) -> JSIPreference:
"""convert a list of jsi keys into a preference function"""
jsi_order = reversed(to_jsi_keys(jsi_order))
pref_score = {jsi_cls: (i + 1) * multiplier for i, jsi_cls in enumerate(jsi_order)}
@ -87,7 +88,7 @@ def __init__(
self._url = self._sanitize_url(url)
self.preferences: set[JSIPreference] = {
order_to_pref(self._load_pref_from_option(), 10000),
order_to_pref(self._load_jsi_keys_from_option('jsi_preference'), 10000),
order_to_pref(preferred_order, 100),
} | _JSI_PREFERENCES
@ -108,17 +109,20 @@ def _sanitize_url(self, url):
self.report_warning(f'Invalid URL: "{url}", using empty string instead')
return sanitized
def _load_pref_from_option(self):
user_prefs = self._downloader.params.get('jsi_preference', [])
valid_handlers = list(all_handlers())
for invalid_key in [jsi_key for jsi_key in user_prefs if jsi_key not in valid_handlers]:
self.report_warning(f'`{invalid_key}` is not a valid JSI, ignoring preference setting')
user_prefs.remove(invalid_key)
return user_prefs
def _load_jsi_keys_from_option(self, option_key):
jsi_keys = self._downloader.params.get(option_key, [])
valid_handlers = list(get_all_handlers())
for invalid_key in [key for key in jsi_keys if key not in valid_handlers]:
self.report_warning(f'{option_key}: `{invalid_key}` is not a valid JSI', only_once=True)
jsi_keys.remove(invalid_key)
return jsi_keys
def _load_allowed_jsi_cls(self, only_include, exclude):
handler_classes = get_included_jsi(only_include, exclude)
self.write_debug(f'Select JSI: {to_jsi_keys(handler_classes)}, '
self.write_debug(f'Loaded JSI runtimes: {get_all_handlers()}')
handler_classes = filter_dict(
get_included_jsi(only_include, exclude),
lambda _, v: v.supports_extractor(self._ie_key))
self.write_debug(f'Select JSI {"for " + self._ie_key if self._ie_key else ""}: {to_jsi_keys(handler_classes)}, '
f'included: {to_jsi_keys(only_include) or "all"}, excluded: {to_jsi_keys(exclude)}')
return handler_classes
@ -129,13 +133,13 @@ def report_warning(self, message, only_once=False):
return self._downloader.report_warning(f'[JSIDirector] {message}', only_once=only_once)
def _get_handlers(self, method_name: str, *args, **kwargs) -> list[JSI]:
def _supports(jsi: JSI):
def _supports_method_with_params(jsi: JSI):
if not callable(method := getattr(jsi, method_name, None)):
return False
method_params = inspect.signature(method).parameters
return all(key in method_params for key in kwargs)
handlers = [h for h in self._handler_dict.values() if _supports(h)]
handlers = [h for h in self._handler_dict.values() if _supports_method_with_params(h)]
self.write_debug(f'Choosing handlers for method `{method_name}` with kwargs {list(kwargs)}'
f': {to_jsi_keys(handlers)}')
@ -169,6 +173,7 @@ def _dispatch_request(self, method_name: str, *args, **kwargs):
try:
self.write_debug(f'Dispatching `{method_name}` task to {handler.JSI_NAME}')
handler.report_version()
return getattr(handler, method_name)(*args, **kwargs)
except ExtractorError as e:
if self._is_test:
@ -225,6 +230,13 @@ def to_screen(self, msg, *args, **kwargs):
def report_note(self, video_id, note):
self.to_screen(f'{format_field(video_id, None, "%s: ")}{note}')
def report_version(self):
raise NotImplementedError
@classmethod
def supports_extractor(cls, ie_key: str):
return True
@classproperty
def JSI_NAME(cls) -> str:
return cls.__name__[:-3]
@ -250,6 +262,9 @@ def exe(cls):
def is_available(cls):
return bool(cls.exe)
def report_version(self):
self.write_debug(f'{self._EXE_NAME} version {self.exe_version}')
def register_jsi_preference(*handlers: type[JSI]):
assert all(issubclass(handler, JSI) for handler in handlers), f'{handlers} must all be a subclass of JSI'