]>
jfr.im git - yt-dlp.git/blob - test/test_download.py
13 # Allow direct execution
14 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))))
16 import youtube_dl
.YoutubeDL
17 import youtube_dl
.extractor
18 from youtube_dl
.utils
import *
20 DEF_FILE
= os
.path
.join(os
.path
.dirname(os
.path
.abspath(__file__
)), 'tests.json')
21 PARAMETERS_FILE
= os
.path
.join(os
.path
.dirname(os
.path
.abspath(__file__
)), "parameters.json")
25 # General configuration (from __init__, not very elegant...)
26 jar
= compat_cookiejar
.CookieJar()
27 cookie_processor
= compat_urllib_request
.HTTPCookieProcessor(jar
)
28 proxy_handler
= compat_urllib_request
.ProxyHandler()
29 opener
= compat_urllib_request
.build_opener(proxy_handler
, cookie_processor
, YoutubeDLHandler())
30 compat_urllib_request
.install_opener(opener
)
31 socket
.setdefaulttimeout(10)
33 def _try_rm(filename
):
34 """ Remove a file if it exists """
37 except OSError as ose
:
38 if ose
.errno
!= errno
.ENOENT
:
41 md5
= lambda s
: hashlib
.md5(s
.encode('utf-8')).hexdigest()
43 class YoutubeDL(youtube_dl
.YoutubeDL
):
44 def __init__(self
, *args
, **kwargs
):
45 self
.to_stderr
= self
.to_screen
46 self
.processed_info_dicts
= []
47 super(YoutubeDL
, self
).__init
__(*args
, **kwargs
)
48 def report_warning(self
, message
):
49 # Don't accept warnings during tests
50 raise ExtractorError(message
)
51 def process_info(self
, info_dict
):
52 self
.processed_info_dicts
.append(info_dict
)
53 return super(YoutubeDL
, self
).process_info(info_dict
)
56 with open(fn
, 'rb') as f
:
57 return hashlib
.md5(f
.read()).hexdigest()
59 with io
.open(DEF_FILE
, encoding
='utf-8') as deff
:
60 defs
= json
.load(deff
)
61 for ie
in youtube_dl
.extractor
.gen_extractors():
62 t
= getattr(ie
, '_TEST', None)
64 t
['name'] = type(ie
).__name
__[:-len('IE')]
66 for t
in getattr(ie
, '_TESTS', []):
67 t
['name'] = type(ie
).__name
__[:-len('IE')]
71 with io
.open(PARAMETERS_FILE
, encoding
='utf-8') as pf
:
72 parameters
= json
.load(pf
)
75 class TestDownload(unittest
.TestCase
):
78 self
.parameters
= parameters
81 ### Dynamically generate tests
82 def generator(test_case
):
84 def test_template(self
):
85 ie
= youtube_dl
.extractor
.get_info_extractor(test_case
['name'])
87 print('Skipping: IE marked as not _WORKING')
89 if 'playlist' not in test_case
and not test_case
['file']:
90 print('Skipping: No output file specified')
92 if 'skip' in test_case
:
93 print('Skipping: {0}'.format(test_case
['skip']))
96 params
= self
.parameters
.copy()
97 params
.update(test_case
.get('params', {}))
99 ydl
= YoutubeDL(params
)
100 for ie
in youtube_dl
.extractor
.gen_extractors():
101 ydl
.add_info_extractor(ie
)
102 finished_hook_called
= set()
104 if status
['status'] == 'finished':
105 finished_hook_called
.add(status
['filename'])
106 ydl
.fd
.add_progress_hook(_hook
)
108 test_cases
= test_case
.get('playlist', [test_case
])
109 for tc
in test_cases
:
111 _try_rm(tc
['file'] + '.part')
112 _try_rm(tc
['file'] + '.info.json')
114 for retry
in range(1, RETRIES
+ 1):
116 ydl
.download([test_case
['url']])
117 except (DownloadError
, ExtractorError
) as err
:
118 if retry
== RETRIES
: raise
120 # Check if the exception is not a network related one
121 if not err
.exc_info
[0] in (compat_urllib_error
.URLError
, socket
.timeout
, UnavailableVideoError
):
124 print('Retrying: {0} failed tries\n\n##########\n\n'.format(retry
))
128 for tc
in test_cases
:
129 if not test_case
.get('params', {}).get('skip_download', False):
130 self
.assertTrue(os
.path
.exists(tc
['file']), msg
='Missing file ' + tc
['file'])
131 self
.assertTrue(tc
['file'] in finished_hook_called
)
132 self
.assertTrue(os
.path
.exists(tc
['file'] + '.info.json'))
134 md5_for_file
= _file_md5(tc
['file'])
135 self
.assertEqual(md5_for_file
, tc
['md5'])
136 with io
.open(tc
['file'] + '.info.json', encoding
='utf-8') as infof
:
137 info_dict
= json
.load(infof
)
138 for (info_field
, expected
) in tc
.get('info_dict', {}).items():
139 if isinstance(expected
, compat_str
) and expected
.startswith('md5:'):
140 self
.assertEqual(expected
, 'md5:' + md5(info_dict
.get(info_field
)))
142 got
= info_dict
.get(info_field
)
145 u
'invalid value for field %s, expected %r, got %r' % (info_field
, expected
, got
))
147 # If checkable fields are missing from the test case, print the info_dict
148 test_info_dict
= dict((key
, value
if not isinstance(value
, compat_str
) or len(value
) < 250 else 'md5:' + md5(value
))
149 for key
, value
in info_dict
.items()
150 if value
and key
in ('title', 'description', 'uploader', 'upload_date', 'uploader_id', 'location'))
151 if not all(key
in tc
.get('info_dict', {}).keys() for key
in test_info_dict
.keys()):
152 sys
.stderr
.write(u
'\n"info_dict": ' + json
.dumps(test_info_dict
, ensure_ascii
=False, indent
=2) + u
'\n')
154 # Check for the presence of mandatory fields
155 for key
in ('id', 'url', 'title', 'ext'):
156 self
.assertTrue(key
in info_dict
.keys() and info_dict
[key
])
158 for tc
in test_cases
:
160 _try_rm(tc
['file'] + '.part')
161 _try_rm(tc
['file'] + '.info.json')
165 ### And add them to TestDownload
166 for n
, test_case
in enumerate(defs
):
167 test_method
= generator(test_case
)
168 tname
= 'test_' + str(test_case
['name'])
170 while hasattr(TestDownload
, tname
):
171 tname
= 'test_' + test_case
['name'] + '_' + str(i
)
173 test_method
.__name
__ = tname
174 setattr(TestDownload
, test_method
.__name
__, test_method
)
178 if __name__
== '__main__':