]>
jfr.im git - yt-dlp.git/blob - test/test_download.py
3 # Allow direct execution
7 sys
.path
.insert(0, os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))))
9 from test
.helper
import (
24 import youtube_dl
.YoutubeDL
25 from youtube_dl
.utils
import (
32 UnavailableVideoError
,
34 from youtube_dl
.extractor
import get_info_extractor
38 class YoutubeDL(youtube_dl
.YoutubeDL
):
39 def __init__(self
, *args
, **kwargs
):
40 self
.to_stderr
= self
.to_screen
41 self
.processed_info_dicts
= []
42 super(YoutubeDL
, self
).__init
__(*args
, **kwargs
)
43 def report_warning(self
, message
):
44 # Don't accept warnings during tests
45 raise ExtractorError(message
)
46 def process_info(self
, info_dict
):
47 self
.processed_info_dicts
.append(info_dict
)
48 return super(YoutubeDL
, self
).process_info(info_dict
)
51 with open(fn
, 'rb') as f
:
52 return hashlib
.md5(f
.read()).hexdigest()
57 class TestDownload(unittest
.TestCase
):
62 ### Dynamically generate tests
63 def generator(test_case
):
65 def test_template(self
):
66 ie
= youtube_dl
.extractor
.get_info_extractor(test_case
['name'])
67 other_ies
= [get_info_extractor(ie_key
) for ie_key
in test_case
.get('add_ie', [])]
68 def print_skipping(reason
):
69 print('Skipping %s: %s' % (test_case
['name'], reason
))
71 print_skipping('IE marked as not _WORKING')
73 if 'playlist' not in test_case
:
74 info_dict
= test_case
.get('info_dict', {})
75 if not test_case
.get('file') and not (info_dict
.get('id') and info_dict
.get('ext')):
76 raise Exception('Test definition incorrect. The output file cannot be known. Are both \'id\' and \'ext\' keys present?')
77 if 'skip' in test_case
:
78 print_skipping(test_case
['skip'])
80 for other_ie
in other_ies
:
81 if not other_ie
.working():
82 print_skipping(u
'test depends on %sIE, marked as not WORKING' % other_ie
.ie_key())
85 params
= get_params(test_case
.get('params', {}))
87 ydl
= YoutubeDL(params
)
88 ydl
.add_default_info_extractors()
89 finished_hook_called
= set()
91 if status
['status'] == 'finished':
92 finished_hook_called
.add(status
['filename'])
93 ydl
.add_progress_hook(_hook
)
95 def get_tc_filename(tc
):
96 return tc
.get('file') or ydl
.prepare_filename(tc
.get('info_dict', {}))
98 test_cases
= test_case
.get('playlist', [test_case
])
99 def try_rm_tcs_files():
100 for tc
in test_cases
:
101 tc_filename
= get_tc_filename(tc
)
103 try_rm(tc_filename
+ '.part')
104 try_rm(os
.path
.splitext(tc_filename
)[0] + '.info.json')
110 ydl
.download([test_case
['url']])
111 except (DownloadError
, ExtractorError
) as err
:
112 # Check if the exception is not a network related one
113 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):
116 if try_num
== RETRIES
:
117 report_warning(u
'Failed due to network errors, skipping...')
120 print('Retrying: {0} failed tries\n\n##########\n\n'.format(try_num
))
126 for tc
in test_cases
:
127 tc_filename
= get_tc_filename(tc
)
128 if not test_case
.get('params', {}).get('skip_download', False):
129 self
.assertTrue(os
.path
.exists(tc_filename
), msg
='Missing file ' + tc_filename
)
130 self
.assertTrue(tc_filename
in finished_hook_called
)
131 info_json_fn
= os
.path
.splitext(tc_filename
)[0] + '.info.json'
132 self
.assertTrue(os
.path
.exists(info_json_fn
))
134 md5_for_file
= _file_md5(tc_filename
)
135 self
.assertEqual(md5_for_file
, tc
['md5'])
136 with io
.open(info_json_fn
, 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('re:'):
140 got
= info_dict
.get(info_field
)
141 match_str
= expected
[len('re:'):]
142 match_rex
= re
.compile(match_str
)
145 isinstance(got
, compat_str
) and match_rex
.match(got
),
146 u
'field %s (value: %r) should match %r' % (info_field
, got
, match_str
))
147 elif isinstance(expected
, type):
148 got
= info_dict
.get(info_field
)
149 self
.assertTrue(isinstance(got
, expected
),
150 u
'Expected type %r, but got value %r of type %r' % (expected
, got
, type(got
)))
152 if isinstance(expected
, compat_str
) and expected
.startswith('md5:'):
153 got
= 'md5:' + md5(info_dict
.get(info_field
))
155 got
= info_dict
.get(info_field
)
156 self
.assertEqual(expected
, got
,
157 u
'invalid value for field %s, expected %r, got %r' % (info_field
, expected
, got
))
159 # Check for the presence of mandatory fields
160 for key
in ('id', 'url', 'title', 'ext'):
161 self
.assertTrue(key
in info_dict
.keys() and info_dict
[key
])
162 # Check for mandatory fields that are automatically set by YoutubeDL
163 for key
in ['webpage_url', 'extractor', 'extractor_key']:
164 self
.assertTrue(info_dict
.get(key
), u
'Missing field: %s' % key
)
166 # Are checkable fields missing from the test case definition?
167 test_info_dict
= dict((key
, value
if not isinstance(value
, compat_str
) or len(value
) < 250 else 'md5:' + md5(value
))
168 for key
, value
in info_dict
.items()
169 if value
and key
in ('title', 'description', 'uploader', 'upload_date', 'timestamp', 'uploader_id', 'location'))
170 missing_keys
= set(test_info_dict
.keys()) - set(tc
.get('info_dict', {}).keys())
172 sys
.stderr
.write(u
'\n"info_dict": ' + json
.dumps(test_info_dict
, ensure_ascii
=False, indent
=4) + u
'\n')
175 'Missing keys in test definition: %s' % (
176 ','.join(sorted(missing_keys
))))
182 ### And add them to TestDownload
183 for n
, test_case
in enumerate(defs
):
184 test_method
= generator(test_case
)
185 tname
= 'test_' + str(test_case
['name'])
187 while hasattr(TestDownload
, tname
):
188 tname
= 'test_' + str(test_case
['name']) + '_' + str(i
)
190 test_method
.__name
__ = tname
191 setattr(TestDownload
, test_method
.__name
__, test_method
)
195 if __name__
== '__main__':