]>
jfr.im git - yt-dlp.git/blob - test/helper.py
3b550d19271967897c536ffdd27f1ee7370bc676
10 import yt_dlp
.extractor
11 from yt_dlp
import YoutubeDL
12 from yt_dlp
.compat
import compat_os_name
13 from yt_dlp
.utils
import preferredencoding
, try_call
, write_string
, find_available_port
15 if 'pytest' in sys
.modules
:
17 is_download_test
= pytest
.mark
.download
19 def is_download_test(test_class
):
23 def get_params(override
=None):
24 PARAMETERS_FILE
= os
.path
.join(os
.path
.dirname(os
.path
.abspath(__file__
)),
26 LOCAL_PARAMETERS_FILE
= os
.path
.join(os
.path
.dirname(os
.path
.abspath(__file__
)),
27 'local_parameters.json')
28 with open(PARAMETERS_FILE
, encoding
='utf-8') as pf
:
29 parameters
= json
.load(pf
)
30 if os
.path
.exists(LOCAL_PARAMETERS_FILE
):
31 with open(LOCAL_PARAMETERS_FILE
, encoding
='utf-8') as pf
:
32 parameters
.update(json
.load(pf
))
34 parameters
.update(override
)
39 """ Remove a file if it exists """
42 except OSError as ose
:
43 if ose
.errno
!= errno
.ENOENT
:
47 def report_warning(message
, *args
, **kwargs
):
49 Print the message to stderr, it will be prefixed with 'WARNING:'
50 If stderr is a tty file the 'WARNING:' will be colored
52 if sys
.stderr
.isatty() and compat_os_name
!= 'nt':
53 _msg_header
= '\033[0;33mWARNING:\033[0m'
55 _msg_header
= 'WARNING:'
56 output
= f
'{_msg_header} {message}\n'
57 if 'b' in getattr(sys
.stderr
, 'mode', ''):
58 output
= output
.encode(preferredencoding())
59 sys
.stderr
.write(output
)
62 class FakeYDL(YoutubeDL
):
63 def __init__(self
, override
=None):
64 # Different instances of the downloader can't share the same dictionary
65 # some test set the "sublang" parameter, which would break the md5 checks.
66 params
= get_params(override
=override
)
67 super().__init
__(params
, auto_init
=False)
70 def to_screen(self
, s
, *args
, **kwargs
):
73 def trouble(self
, s
, *args
, **kwargs
):
76 def download(self
, x
):
79 def expect_warning(self
, regex
):
80 # Silence an expected warning matching a regex
81 old_report_warning
= self
.report_warning
83 def report_warning(self
, message
, *args
, **kwargs
):
84 if re
.match(regex
, message
):
86 old_report_warning(message
, *args
, **kwargs
)
87 self
.report_warning
= types
.MethodType(report_warning
, self
)
90 def gettestcases(include_onlymatching
=False):
91 for ie
in yt_dlp
.extractor
.gen_extractors():
92 yield from ie
.get_testcases(include_onlymatching
)
95 def getwebpagetestcases():
96 for ie
in yt_dlp
.extractor
.gen_extractors():
97 for tc
in ie
.get_webpage_testcases():
98 tc
.setdefault('add_ie', []).append('Generic')
102 md5
= lambda s
: hashlib
.md5(s
.encode()).hexdigest()
105 def expect_value(self
, got
, expected
, field
):
106 if isinstance(expected
, str) and expected
.startswith('re:'):
107 match_str
= expected
[len('re:'):]
108 match_rex
= re
.compile(match_str
)
111 isinstance(got
, str),
112 f
'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
114 match_rex
.match(got
),
115 f
'field {field} (value: {got!r}) should match {match_str!r}')
116 elif isinstance(expected
, str) and expected
.startswith('startswith:'):
117 start_str
= expected
[len('startswith:'):]
119 isinstance(got
, str),
120 f
'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
122 got
.startswith(start_str
),
123 f
'field {field} (value: {got!r}) should start with {start_str!r}')
124 elif isinstance(expected
, str) and expected
.startswith('contains:'):
125 contains_str
= expected
[len('contains:'):]
127 isinstance(got
, str),
128 f
'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
131 f
'field {field} (value: {got!r}) should contain {contains_str!r}')
132 elif isinstance(expected
, type):
134 isinstance(got
, expected
),
135 f
'Expected type {expected!r} for field {field}, but got value {got!r} of type {type(got)!r}')
136 elif isinstance(expected
, dict) and isinstance(got
, dict):
137 expect_dict(self
, got
, expected
)
138 elif isinstance(expected
, list) and isinstance(got
, list):
140 len(expected
), len(got
),
141 f
'Expect a list of length {len(expected)}, but got a list of length {len(got)} for field {field}')
142 for index
, (item_got
, item_expected
) in enumerate(zip(got
, expected
)):
143 type_got
= type(item_got
)
144 type_expected
= type(item_expected
)
146 type_expected
, type_got
,
147 f
'Type mismatch for list item at index {index} for field {field}, '
148 f
'expected {type_expected!r}, got {type_got!r}')
149 expect_value(self
, item_got
, item_expected
, field
)
151 if isinstance(expected
, str) and expected
.startswith('md5:'):
153 isinstance(got
, str),
154 f
'Expected field {field} to be a unicode object, but got value {got!r} of type {type(got)!r}')
155 got
= 'md5:' + md5(got
)
156 elif isinstance(expected
, str) and re
.match(r
'^(?:min|max)?count:\d+', expected
):
158 isinstance(got
, (list, dict)),
159 f
'Expected field {field} to be a list or a dict, but it is of type {type(got).__name__}')
160 op
, _
, expected_num
= expected
.partition(':')
161 expected_num
= int(expected_num
)
163 assert_func
= assertGreaterEqual
164 msg_tmpl
= 'Expected %d items in field %s, but only got %d'
165 elif op
== 'maxcount':
166 assert_func
= assertLessEqual
167 msg_tmpl
= 'Expected maximum %d items in field %s, but got %d'
169 assert_func
= assertEqual
170 msg_tmpl
= 'Expected exactly %d items in field %s, but got %d'
174 self
, len(got
), expected_num
,
175 msg_tmpl
% (expected_num
, field
, len(got
)))
179 f
'Invalid value for field {field}, expected {expected!r}, got {got!r}')
182 def expect_dict(self
, got_dict
, expected_dict
):
183 for info_field
, expected
in expected_dict
.items():
184 got
= got_dict
.get(info_field
)
185 expect_value(self
, got
, expected
, info_field
)
188 def sanitize_got_info_dict(got_dict
):
190 *YoutubeDL
._format
_fields
,
193 'formats', 'thumbnails', 'subtitles', 'automatic_captions', 'comments', 'entries',
196 'autonumber', 'playlist', 'format_index', 'video_ext', 'audio_ext', 'duration_string', 'epoch', 'n_entries',
197 'fulltitle', 'extractor', 'extractor_key', 'filename', 'filepath', 'infojson_filename', 'original_url',
199 # Only live_status needs to be checked
200 'is_live', 'was_live',
203 IGNORED_PREFIXES
= ('', 'playlist', 'requested', 'webpage')
205 def sanitize(key
, value
):
206 if isinstance(value
, str) and len(value
) > 100 and key
!= 'thumbnail':
207 return f
'md5:{md5(value)}'
208 elif isinstance(value
, list) and len(value
) > 10:
209 return f
'count:{len(value)}'
210 elif key
.endswith('_count') and isinstance(value
, int):
215 key
: sanitize(key
, value
) for key
, value
in got_dict
.items()
216 if value
is not None and key
not in IGNORED_FIELDS
and (
217 not any(key
.startswith(f
'{prefix}_') for prefix
in IGNORED_PREFIXES
)
218 or key
== '_old_archive_ids')
221 # display_id may be generated from id
222 if test_info_dict
.get('display_id') == test_info_dict
.get('id'):
223 test_info_dict
.pop('display_id')
225 # Remove deprecated fields
226 for old
in YoutubeDL
._deprecated
_multivalue
_fields
:
227 test_info_dict
.pop(old
, None)
229 # release_year may be generated from release_date
230 if try_call(lambda: test_info_dict
['release_year'] == int(test_info_dict
['release_date'][:4])):
231 test_info_dict
.pop('release_year')
233 # Check url for flat entries
234 if got_dict
.get('_type', 'video') != 'video' and got_dict
.get('url'):
235 test_info_dict
['url'] = got_dict
['url']
237 return test_info_dict
240 def expect_info_dict(self
, got_dict
, expected_dict
):
241 expect_dict(self
, got_dict
, expected_dict
)
242 # Check for the presence of mandatory fields
243 if got_dict
.get('_type') not in ('playlist', 'multi_video'):
244 mandatory_fields
= ['id', 'title']
245 if expected_dict
.get('ext'):
246 mandatory_fields
.extend(('url', 'ext'))
247 for key
in mandatory_fields
:
248 self
.assertTrue(got_dict
.get(key
), f
'Missing mandatory field {key}')
249 # Check for mandatory fields that are automatically set by YoutubeDL
250 if got_dict
.get('_type', 'video') == 'video':
251 for key
in ['webpage_url', 'extractor', 'extractor_key']:
252 self
.assertTrue(got_dict
.get(key
), f
'Missing field: {key}')
254 test_info_dict
= sanitize_got_info_dict(got_dict
)
256 missing_keys
= set(test_info_dict
.keys()) - set(expected_dict
.keys())
259 if isinstance(v
, str):
260 return "'{}'".format(v
.replace('\\', '\\\\').replace("'", "\\'").replace('\n', '\\n'))
261 elif isinstance(v
, type):
265 info_dict_str
= ''.join(
266 f
' {_repr(k)}: {_repr(v)},\n'
267 for k
, v
in test_info_dict
.items() if k
not in missing_keys
)
269 info_dict_str
+= '\n'
270 info_dict_str
+= ''.join(
271 f
' {_repr(k)}: {_repr(test_info_dict[k])},\n'
272 for k
in missing_keys
)
273 info_dict_str
= '\n\'info_dict\': {\n' + info_dict_str + '},\n'
274 write_string(info_dict_str
.replace('\n', '\n '), out
=sys
.stderr
)
277 'Missing keys in test definition: {}'.format(', '.join(sorted(missing_keys
))))
280 def assertRegexpMatches(self
, text
, regexp
, msg
=None):
281 if hasattr(self
, 'assertRegexp'):
282 return self
.assertRegexp(text
, regexp
, msg
)
284 m
= re
.match(regexp
, text
)
286 note
= f
'Regexp didn\'t match: {regexp!r} not found'
288 note
+= f
' in {text!r}'
292 msg
= note
+ ', ' + msg
293 self
.assertTrue(m
, msg
)
296 def assertGreaterEqual(self
, got
, expected
, msg
=None):
297 if not (got
>= expected
):
299 msg
= f
'{got!r} not greater than or equal to {expected!r}'
300 self
.assertTrue(got
>= expected
, msg
)
303 def assertLessEqual(self
, got
, expected
, msg
=None):
304 if not (got
<= expected
):
306 msg
= f
'{got!r} not less than or equal to {expected!r}'
307 self
.assertTrue(got
<= expected
, msg
)
310 def assertEqual(self
, got
, expected
, msg
=None):
313 msg
= f
'{got!r} not equal to {expected!r}'
314 self
.assertTrue(got
== expected
, msg
)
317 def expect_warnings(ydl
, warnings_re
):
318 real_warning
= ydl
.report_warning
320 def _report_warning(w
, *args
, **kwargs
):
321 if not any(re
.search(w_re
, w
) for w_re
in warnings_re
):
322 real_warning(w
, *args
, **kwargs
)
324 ydl
.report_warning
= _report_warning
327 def http_server_port(httpd
):
328 if os
.name
== 'java' and isinstance(httpd
.socket
, ssl
.SSLSocket
):
329 # In Jython SSLSocket is not a subclass of socket.socket
330 sock
= httpd
.socket
.sock
333 return sock
.getsockname()[1]
336 def verify_address_availability(address
):
337 if find_available_port(address
) is None:
338 pytest
.skip(f
'Unable to bind to source address {address} (address may not exist)')
341 def validate_and_send(rh
, req
):