X-Git-Url: https://jfr.im/git/yt-dlp.git/blobdiff_plain/655c410063da3400272b1e08d2f7432f2bb9e182..ac668111128b5f124b4271b3aa4c35f6e71a4749:/test/test_download.py diff --git a/test/test_download.py b/test/test_download.py old mode 100644 new mode 100755 index ebe820dfc..b82f174bb --- a/test/test_download.py +++ b/test/test_download.py @@ -1,59 +1,54 @@ -#!/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 hashlib +import json +import socket +import urllib.error +import http.client + from test.helper import ( assertGreaterEqual, + expect_info_dict, expect_warnings, get_params, gettestcases, - expect_info_dict, - try_rm, + 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.compat import compat_HTTPError +from yt_dlp.extractor import get_info_extractor +from yt_dlp.utils import ( DownloadError, ExtractorError, - format_bytes, UnavailableVideoError, + format_bytes, ) -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): @@ -64,6 +59,7 @@ def _file_md5(fn): defs = gettestcases() +@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,12 +67,14 @@ class TestDownload(unittest.TestCase): maxDiff = None + COMPLETED_TESTS = {} + def __str__(self): """Identify each test with the `add_ie` attribute, if available.""" 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__) + return f'{cls.__module__}.{cls.__name__}' add_ie = getattr(self, self._testMethodName).add_ie return '%s (%s)%s:' % (self._testMethodName, @@ -92,7 +90,10 @@ def setUp(self): def generator(test_case, tname): def test_template(self): - ie = youtube_dl.extractor.get_info_extractor(test_case['name'])() + 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( @@ -100,27 +101,33 @@ def test_template(self): 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: 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('Test definition incorrect. \'id\' key is not present') + elif not info_dict.get('ext'): + if params.get('skip_download') and params.get('ignore_no_formats_error'): + continue + raise Exception('Test definition incorrect. The output file cannot be known. \'ext\' key is not present') 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')) params.setdefault('skip_download', True) ydl = YoutubeDL(params, auto_init=False) @@ -134,7 +141,7 @@ 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 @@ -159,14 +166,14 @@ 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 err.exc_info[0] in (urllib.error.URLError, socket.timeout, UnavailableVideoError, http.client.BadStatusLine) or (err.exc_info[0] == compat_HTTPError and err.exc_info[1].code == 503): 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 else: @@ -232,7 +239,7 @@ def try_rm_tcs_files(tcs=None): 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: @@ -247,12 +254,12 @@ def try_rm_tcs_files(tcs=None): # 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 +tests_counter = {} +for test_case in defs: + name = test_case['name'] + i = tests_counter.get(name, 0) + tests_counter[name] = i + 1 + tname = f'test_{name}_{i}' if i else f'test_{name}' test_method = generator(test_case, tname) test_method.__name__ = str(tname) ie_list = test_case.get('add_ie') @@ -261,5 +268,22 @@ def try_rm_tcs_files(tcs=None): del test_method +def batch_generator(name, num_tests): + + def test_template(self): + for i in range(num_tests): + getattr(self, f'test_{name}_{i}' if i else f'test_{name}')() + + return test_template + + +for name, num_tests in tests_counter.items(): + test_method = batch_generator(name, num_tests) + test_method.__name__ = f'test_{name}_all' + test_method.add_ie = '' + setattr(TestDownload, test_method.__name__, test_method) + del test_method + + if __name__ == '__main__': unittest.main()