X-Git-Url: https://jfr.im/git/yt-dlp.git/blobdiff_plain/60e5016199320333ad3f3584b5b0c0147cfaeba3..168e72dcd3e04e0e19e92c012a04b8a1e4658f50:/test/test_download.py diff --git a/test/test_download.py b/test/test_download.py old mode 100644 new mode 100755 index fd7d2e8be..253079249 --- a/test/test_download.py +++ b/test/test_download.py @@ -1,59 +1,57 @@ -#!/usr/bin/env python - -from __future__ import unicode_literals +#!/usr/bin/env python3 # Allow direct execution import os import sys import unittest + sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import collections +import hashlib +import json + from test.helper import ( assertGreaterEqual, + expect_info_dict, expect_warnings, get_params, gettestcases, - expect_info_dict, - try_rm, + getwebpagetestcases, + is_download_test, report_warning, + try_rm, ) - -import hashlib -import io -import json -import socket - -import youtube_dl.YoutubeDL -from youtube_dl.compat import ( - compat_http_client, - compat_urllib_error, - compat_HTTPError, -) -from youtube_dl.utils import ( +import yt_dlp.YoutubeDL # isort: split +from yt_dlp.extractor import get_info_extractor +from yt_dlp.networking.exceptions import HTTPError, TransportError +from yt_dlp.utils import ( DownloadError, ExtractorError, - format_bytes, UnavailableVideoError, + YoutubeDLError, + format_bytes, + join_nonempty, ) -from youtube_dl.extractor import get_info_extractor RETRIES = 3 -class YoutubeDL(youtube_dl.YoutubeDL): +class YoutubeDL(yt_dlp.YoutubeDL): def __init__(self, *args, **kwargs): self.to_stderr = self.to_screen self.processed_info_dicts = [] - super(YoutubeDL, self).__init__(*args, **kwargs) + super().__init__(*args, **kwargs) - def report_warning(self, message): + def report_warning(self, message, *args, **kwargs): # Don't accept warnings during tests raise ExtractorError(message) def process_info(self, info_dict): - self.processed_info_dicts.append(info_dict) - return super(YoutubeDL, self).process_info(info_dict) + self.processed_info_dicts.append(info_dict.copy()) + return super().process_info(info_dict) def _file_md5(fn): @@ -61,9 +59,12 @@ def _file_md5(fn): return hashlib.md5(f.read()).hexdigest() -defs = gettestcases() +normal_test_cases = gettestcases() +webpage_test_cases = getwebpagetestcases() +tests_counter = collections.defaultdict(collections.Counter) +@is_download_test class TestDownload(unittest.TestCase): # Parallel testing in nosetests. See # http://nose.readthedocs.org/en/latest/doc_tests/test_multiprocess/multiprocess.html @@ -71,56 +72,59 @@ class TestDownload(unittest.TestCase): maxDiff = None + COMPLETED_TESTS = {} + def __str__(self): """Identify each test with the `add_ie` attribute, if available.""" + cls, add_ie = type(self), getattr(self, self._testMethodName).add_ie + return f'{self._testMethodName} ({cls.__module__}.{cls.__name__}){f" [{add_ie}]" if add_ie else ""}:' - def strclass(cls): - """From 2.7's unittest; 2.6 had _strclass so we can't import it.""" - return '%s.%s' % (cls.__module__, cls.__name__) - - add_ie = getattr(self, self._testMethodName).add_ie - return '%s (%s)%s:' % (self._testMethodName, - strclass(self.__class__), - ' [%s]' % add_ie if add_ie else '') - - def setUp(self): - self.defs = defs # Dynamically generate tests - def generator(test_case, tname): - def test_template(self): - ie = youtube_dl.extractor.get_info_extractor(test_case['name']) - other_ies = [get_info_extractor(ie_key) for ie_key in test_case.get('add_ie', [])] + if self.COMPLETED_TESTS.get(tname): + return + self.COMPLETED_TESTS[tname] = True + ie = yt_dlp.extractor.get_info_extractor(test_case['name'])() + other_ies = [get_info_extractor(ie_key)() for ie_key in test_case.get('add_ie', [])] is_playlist = any(k.startswith('playlist') for k in test_case) test_cases = test_case.get( 'playlist', [] if is_playlist else [test_case]) def print_skipping(reason): print('Skipping %s: %s' % (test_case['name'], reason)) + self.skipTest(reason) + if not ie.working(): print_skipping('IE marked as not _WORKING') - return for tc in test_cases: + if tc.get('expected_exception'): + continue info_dict = tc.get('info_dict', {}) - if not (info_dict.get('id') and info_dict.get('ext')): - raise Exception('Test definition incorrect. The output file cannot be known. Are both \'id\' and \'ext\' keys present?') + params = tc.get('params', {}) + if not info_dict.get('id'): + raise Exception(f'Test {tname} definition incorrect - "id" key is not present') + elif not info_dict.get('ext') and info_dict.get('_type', 'video') == 'video': + if params.get('skip_download') and params.get('ignore_no_formats_error'): + continue + raise Exception(f'Test {tname} definition incorrect - "ext" key must be present to define the output file') if 'skip' in test_case: print_skipping(test_case['skip']) - return + for other_ie in other_ies: if not other_ie.working(): print_skipping('test depends on %sIE, marked as not WORKING' % other_ie.ie_key()) - return params = get_params(test_case.get('params', {})) params['outtmpl'] = tname + '_' + params['outtmpl'] if is_playlist and 'playlist' not in test_case: params.setdefault('extract_flat', 'in_playlist') + params.setdefault('playlistend', test_case.get( + 'playlist_mincount', test_case.get('playlist_count', -2) + 1)) params.setdefault('skip_download', True) ydl = YoutubeDL(params, auto_init=False) @@ -134,10 +138,21 @@ def _hook(status): expect_warnings(ydl, test_case.get('expected_warnings', [])) def get_tc_filename(tc): - return ydl.prepare_filename(tc.get('info_dict', {})) + return ydl.prepare_filename(dict(tc.get('info_dict', {}))) res_dict = None + def match_exception(err): + expected_exception = test_case.get('expected_exception') + if not expected_exception: + return False + if err.__class__.__name__ == expected_exception: + return True + for exc in err.exc_info: + if exc.__class__.__name__ == expected_exception: + return True + return False + def try_rm_tcs_files(tcs=None): if tcs is None: tcs = test_cases @@ -159,16 +174,23 @@ def try_rm_tcs_files(tcs=None): force_generic_extractor=params.get('force_generic_extractor', False)) except (DownloadError, ExtractorError) as err: # Check if the exception is not a network related one - if not err.exc_info[0] in (compat_urllib_error.URLError, socket.timeout, UnavailableVideoError, compat_http_client.BadStatusLine) or (err.exc_info[0] == compat_HTTPError and err.exc_info[1].code == 503): + if not isinstance(err.exc_info[1], (TransportError, UnavailableVideoError)) or (isinstance(err.exc_info[1], HTTPError) and err.exc_info[1].status == 503): + if match_exception(err): + return + err.msg = f'{getattr(err, "msg", err)} ({tname})' raise if try_num == RETRIES: report_warning('%s failed due to network errors, skipping...' % tname) return - print('Retrying: {0} failed tries\n\n##########\n\n'.format(try_num)) + print(f'Retrying: {try_num} failed tries\n\n##########\n\n') try_num += 1 + except YoutubeDLError as err: + if match_exception(err): + return + raise else: break @@ -199,9 +221,18 @@ def try_rm_tcs_files(tcs=None): self.assertEqual( test_case['playlist_duration_sum'], got_duration) + # Generalize both playlists and single videos to unified format for + # simplicity + if 'entries' not in res_dict: + res_dict['entries'] = [res_dict] + for tc_num, tc in enumerate(test_cases): - tc_res_dict = res_dict['entries'][tc_num] if is_playlist else res_dict + tc_res_dict = res_dict['entries'][tc_num] + # First, check test cases' data against extracted data alone expect_info_dict(self, tc_res_dict, tc.get('info_dict', {})) + if tc_res_dict.get('_type', 'video') != 'video': + continue + # Now, check downloaded file consistency tc_filename = get_tc_filename(tc) if not test_case.get('params', {}).get('skip_download', False): self.assertTrue(os.path.exists(tc_filename), msg='Missing file ' + tc_filename) @@ -218,14 +249,15 @@ def try_rm_tcs_files(tcs=None): format_bytes(got_fsize))) if 'md5' in tc: md5_for_file = _file_md5(tc_filename) - self.assertEqual(md5_for_file, tc['md5']) + self.assertEqual(tc['md5'], md5_for_file) + # Finally, check test cases' data again but this time against + # extracted data from info JSON file written during processing info_json_fn = os.path.splitext(tc_filename)[0] + '.info.json' self.assertTrue( os.path.exists(info_json_fn), 'Missing info file %s' % info_json_fn) - with io.open(info_json_fn, encoding='utf-8') as infof: + with open(info_json_fn, encoding='utf-8') as infof: info_dict = json.load(infof) - expect_info_dict(self, info_dict, tc.get('info_dict', {})) finally: try_rm_tcs_files() @@ -234,23 +266,48 @@ def try_rm_tcs_files(tcs=None): # extractor returns full results even with extract_flat res_tcs = [{'info_dict': e} for e in res_dict['entries']] try_rm_tcs_files(res_tcs) - + ydl.close() return test_template # And add them to TestDownload -for n, test_case in enumerate(defs): - tname = 'test_' + str(test_case['name']) - i = 1 - while hasattr(TestDownload, tname): - tname = 'test_%s_%d' % (test_case['name'], i) - i += 1 - test_method = generator(test_case, tname) - test_method.__name__ = str(tname) - ie_list = test_case.get('add_ie') - test_method.add_ie = ie_list and ','.join(ie_list) +def inject_tests(test_cases, label=''): + for test_case in test_cases: + name = test_case['name'] + tname = join_nonempty('test', name, label, tests_counter[name][label], delim='_') + tests_counter[name][label] += 1 + + test_method = generator(test_case, tname) + test_method.__name__ = tname + test_method.add_ie = ','.join(test_case.get('add_ie', [])) + setattr(TestDownload, test_method.__name__, test_method) + + +inject_tests(normal_test_cases) + +# TODO: disable redirection to the IE to ensure we are actually testing the webpage extraction +inject_tests(webpage_test_cases, 'webpage') + + +def batch_generator(name): + def test_template(self): + for label, num_tests in tests_counter[name].items(): + for i in range(num_tests): + test_name = join_nonempty('test', name, label, i, delim='_') + try: + getattr(self, test_name)() + except unittest.SkipTest: + print(f'Skipped {test_name}') + + return test_template + + +for name in tests_counter: + test_method = batch_generator(name) + test_method.__name__ = f'test_{name}_all' + test_method.add_ie = '' setattr(TestDownload, test_method.__name__, test_method) - del test_method +del test_method if __name__ == '__main__':