mirror/yt-dlp
1
0
Fork 0
mirror of https://github.com/yt-dlp/yt-dlp.git synced 2025-03-09 12:50:23 -05:00
Code Activity

test behavior

Browse source
This commit is contained in:
c-basalt 2024-09-19 19:24:50 -04:00
parent 062a9785f1
commit 302ee5d0d4
1 changed files with 34 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
yt_dlp/jsinterp/common.py
View file

@ -2,7 +2,6 @@
import abc import abc
import typing import typing
# import dataclasses
from ..utils import classproperty from ..utils import classproperty
@ -90,6 +89,10 @@ def add_handler(self, handler: JSI):
def write_debug(self): def write_debug(self):
return self._downloader.write_debug return self._downloader.write_debug
@property
def report_warning(self):
return self._downloader.report_warning
def _get_handlers(self, method: str, *args, **kwargs) -> list[JSI]: def _get_handlers(self, method: str, *args, **kwargs) -> list[JSI]:
handlers = [h for h in self._handler_dict.values() if getattr(h, method, None)] handlers = [h for h in self._handler_dict.values() if getattr(h, method, None)]
self.write_debug(f'JSIDirector has handlers for `{method}`: {handlers}') self.write_debug(f'JSIDirector has handlers for `{method}`: {handlers}')
@ -125,22 +128,48 @@ def _get_handler_method(method_name: str):
def handler(self: JSIDirector, *args, **kwargs): def handler(self: JSIDirector, *args, **kwargs):
unavailable: list[JSI] = [] unavailable: list[JSI] = []
exceptions: list[tuple[JSI, Exception]] = [] exceptions: list[tuple[JSI, Exception]] = []
is_test = self._downloader.params.get('test', False)
results: list[tuple[JSI, typing.Any]] = []
for handler in self._get_handlers(method_name, *args, **kwargs): for handler in self._get_handlers(method_name, *args, **kwargs):
if not handler.is_available: if not handler.is_available:
if is_test:
raise Exception(f'{handler.JSI_NAME} is not available for testing, '
f'add "{handler.JSI_KEY}" in `exclude` if it should not be used')
self.write_debug(f'{handler.JSI_NAME} is not available') self.write_debug(f'{handler.JSI_NAME} is not available')
unavailable.append(handler) unavailable.append(handler)
continue continue
try: try:
self.write_debug(f'Dispatching `{method_name}` task to {handler.JSI_NAME}') self.write_debug(f'Dispatching `{method_name}` task to {handler.JSI_NAME}')
return getattr(handler, method_name)(*args, **kwargs) result = getattr(handler, method_name)(*args, **kwargs)
if is_test:
results.append((handler, result))
else:
return result
except Exception as e: except Exception as e:
if handler.JSI_KEY not in self._fallback_jsi: if handler.JSI_KEY not in self._fallback_jsi:
raise raise
else: else:
exceptions.append((handler, e)) exceptions.append((handler, e))
if not exceptions: self.write_debug(f'{handler.JSI_NAME} encountered error, fallback to next handler: {e}')
raise Exception(f'No available JSI installed, please install one of: {join_jsi_name(unavailable)}')
raise Exception(f'Failed to perform {method_name}, total {len(exceptions)} errors. Following JSI have been skipped and you can try installing one of them: {join_jsi_name(unavailable)}') if not is_test or not results:
if not exceptions:
msg = f'No available JSI installed, please install one of: {join_jsi_name(unavailable)}'
else:
msg = f'Failed to perform {method_name}, total {len(exceptions)} errors'
if unavailable:
msg = f'{msg}. You can try installing one of unavailable JSI: {join_jsi_name(unavailable)}'
raise Exception(msg)
if is_test:
ref_handler, ref_result = results[0]
for handler, result in results[1:]:
if result != ref_result:
self.report_warning(
f'Different JSI results produced from {ref_handler.JSI_NAME} and {handler.JSI_NAME}')
return ref_result
return handler return handler
execute = _get_handler_method('execute') execute = _get_handler_method('execute')