]>
jfr.im git - yt-dlp.git/blob - test/helper.py
10 import yt_dlp
.extractor
11 from yt_dlp
import YoutubeDL
12 from yt_dlp
.compat
import compat_os_name
, compat_str
13 from yt_dlp
.utils
import preferredencoding
, write_string
15 if 'pytest' in sys
.modules
:
17 is_download_test
= pytest
.mark
.download
19 def is_download_test(testClass
):
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
):
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
, skip_eol
=None):
73 def trouble(self
, s
, tb
=None):
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
):
84 if re
.match(regex
, message
):
86 old_report_warning(message
)
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 md5
= lambda s
: hashlib
.md5(s
.encode()).hexdigest()
98 def expect_value(self
, got
, expected
, field
):
99 if isinstance(expected
, compat_str
) and expected
.startswith('re:'):
100 match_str
= expected
[len('re:'):]
101 match_rex
= re
.compile(match_str
)
104 isinstance(got
, compat_str
),
105 f
'Expected a {compat_str.__name__} object, but got {type(got).__name__} for field {field}')
107 match_rex
.match(got
),
108 f
'field {field} (value: {got!r}) should match {match_str!r}')
109 elif isinstance(expected
, compat_str
) and expected
.startswith('startswith:'):
110 start_str
= expected
[len('startswith:'):]
112 isinstance(got
, compat_str
),
113 f
'Expected a {compat_str.__name__} object, but got {type(got).__name__} for field {field}')
115 got
.startswith(start_str
),
116 f
'field {field} (value: {got!r}) should start with {start_str!r}')
117 elif isinstance(expected
, compat_str
) and expected
.startswith('contains:'):
118 contains_str
= expected
[len('contains:'):]
120 isinstance(got
, compat_str
),
121 f
'Expected a {compat_str.__name__} object, but got {type(got).__name__} for field {field}')
124 f
'field {field} (value: {got!r}) should contain {contains_str!r}')
125 elif isinstance(expected
, type):
127 isinstance(got
, expected
),
128 f
'Expected type {expected!r} for field {field}, but got value {got!r} of type {type(got)!r}')
129 elif isinstance(expected
, dict) and isinstance(got
, dict):
130 expect_dict(self
, got
, expected
)
131 elif isinstance(expected
, list) and isinstance(got
, list):
133 len(expected
), len(got
),
134 'Expect a list of length %d, but got a list of length %d for field %s' % (
135 len(expected
), len(got
), field
))
136 for index
, (item_got
, item_expected
) in enumerate(zip(got
, expected
)):
137 type_got
= type(item_got
)
138 type_expected
= type(item_expected
)
140 type_expected
, type_got
,
141 'Type mismatch for list item at index %d for field %s, expected %r, got %r' % (
142 index
, field
, type_expected
, type_got
))
143 expect_value(self
, item_got
, item_expected
, field
)
145 if isinstance(expected
, compat_str
) and expected
.startswith('md5:'):
147 isinstance(got
, compat_str
),
148 f
'Expected field {field} to be a unicode object, but got value {got!r} of type {type(got)!r}')
149 got
= 'md5:' + md5(got
)
150 elif isinstance(expected
, compat_str
) and re
.match(r
'^(?:min|max)?count:\d+', expected
):
152 isinstance(got
, (list, dict)),
153 f
'Expected field {field} to be a list or a dict, but it is of type {type(got).__name__}')
154 op
, _
, expected_num
= expected
.partition(':')
155 expected_num
= int(expected_num
)
157 assert_func
= assertGreaterEqual
158 msg_tmpl
= 'Expected %d items in field %s, but only got %d'
159 elif op
== 'maxcount':
160 assert_func
= assertLessEqual
161 msg_tmpl
= 'Expected maximum %d items in field %s, but got %d'
163 assert_func
= assertEqual
164 msg_tmpl
= 'Expected exactly %d items in field %s, but got %d'
168 self
, len(got
), expected_num
,
169 msg_tmpl
% (expected_num
, field
, len(got
)))
173 f
'Invalid value for field {field}, expected {expected!r}, got {got!r}')
176 def expect_dict(self
, got_dict
, expected_dict
):
177 for info_field
, expected
in expected_dict
.items():
178 got
= got_dict
.get(info_field
)
179 expect_value(self
, got
, expected
, info_field
)
182 def sanitize_got_info_dict(got_dict
):
184 *YoutubeDL
._format
_fields
,
187 'formats', 'thumbnails', 'subtitles', 'automatic_captions', 'comments', 'entries',
190 'autonumber', 'playlist', 'format_index', 'video_ext', 'audio_ext', 'duration_string', 'epoch',
191 'fulltitle', 'extractor', 'extractor_key', 'filepath', 'infojson_filename', 'original_url', 'n_entries',
193 # Only live_status needs to be checked
194 'is_live', 'was_live',
197 IGNORED_PREFIXES
= ('', 'playlist', 'requested', 'webpage')
199 def sanitize(key
, value
):
200 if isinstance(value
, str) and len(value
) > 100 and key
!= 'thumbnail':
201 return f
'md5:{md5(value)}'
202 elif isinstance(value
, list) and len(value
) > 10:
203 return f
'count:{len(value)}'
204 elif key
.endswith('_count') and isinstance(value
, int):
209 key
: sanitize(key
, value
) for key
, value
in got_dict
.items()
210 if value
is not None and key
not in IGNORED_FIELDS
and not any(
211 key
.startswith(f
'{prefix}_') for prefix
in IGNORED_PREFIXES
)
214 # display_id may be generated from id
215 if test_info_dict
.get('display_id') == test_info_dict
.get('id'):
216 test_info_dict
.pop('display_id')
218 return test_info_dict
221 def expect_info_dict(self
, got_dict
, expected_dict
):
222 expect_dict(self
, got_dict
, expected_dict
)
223 # Check for the presence of mandatory fields
224 if got_dict
.get('_type') not in ('playlist', 'multi_video'):
225 mandatory_fields
= ['id', 'title']
226 if expected_dict
.get('ext'):
227 mandatory_fields
.extend(('url', 'ext'))
228 for key
in mandatory_fields
:
229 self
.assertTrue(got_dict
.get(key
), 'Missing mandatory field %s' % key
)
230 # Check for mandatory fields that are automatically set by YoutubeDL
231 for key
in ['webpage_url', 'extractor', 'extractor_key']:
232 self
.assertTrue(got_dict
.get(key
), 'Missing field: %s' % key
)
234 test_info_dict
= sanitize_got_info_dict(got_dict
)
236 missing_keys
= set(test_info_dict
.keys()) - set(expected_dict
.keys())
239 if isinstance(v
, compat_str
):
240 return "'%s'" % v
.replace('\\', '\\\\').replace("'", "\\'").replace('\n', '\\n')
241 elif isinstance(v
, type):
246 if len(missing_keys
) != len(expected_dict
):
247 info_dict_str
+= ''.join(
248 f
' {_repr(k)}: {_repr(v)},\n'
249 for k
, v
in test_info_dict
.items() if k
not in missing_keys
)
252 info_dict_str
+= '\n'
253 info_dict_str
+= ''.join(
254 f
' {_repr(k)}: {_repr(test_info_dict[k])},\n'
255 for k
in missing_keys
)
257 '\n\'info_dict\': {\n' + info_dict_str + '},\n', out
=sys
.stderr
)
260 'Missing keys in test definition: %s' % (
261 ', '.join(sorted(missing_keys
))))
264 def assertRegexpMatches(self
, text
, regexp
, msg
=None):
265 if hasattr(self
, 'assertRegexp'):
266 return self
.assertRegexp(text
, regexp
, msg
)
268 m
= re
.match(regexp
, text
)
270 note
= 'Regexp didn\'t match: %r not found' % (regexp
)
272 note
+= ' in %r' % text
276 msg
= note
+ ', ' + msg
277 self
.assertTrue(m
, msg
)
280 def assertGreaterEqual(self
, got
, expected
, msg
=None):
281 if not (got
>= expected
):
283 msg
= f
'{got!r} not greater than or equal to {expected!r}'
284 self
.assertTrue(got
>= expected
, msg
)
287 def assertLessEqual(self
, got
, expected
, msg
=None):
288 if not (got
<= expected
):
290 msg
= f
'{got!r} not less than or equal to {expected!r}'
291 self
.assertTrue(got
<= expected
, msg
)
294 def assertEqual(self
, got
, expected
, msg
=None):
295 if not (got
== expected
):
297 msg
= f
'{got!r} not equal to {expected!r}'
298 self
.assertTrue(got
== expected
, msg
)
301 def expect_warnings(ydl
, warnings_re
):
302 real_warning
= ydl
.report_warning
304 def _report_warning(w
):
305 if not any(re
.search(w_re
, w
) for w_re
in warnings_re
):
308 ydl
.report_warning
= _report_warning
311 def http_server_port(httpd
):
312 if os
.name
== 'java' and isinstance(httpd
.socket
, ssl
.SSLSocket
):
313 # In Jython SSLSocket is not a subclass of socket.socket
314 sock
= httpd
.socket
.sock
317 return sock
.getsockname()[1]