test/helper.py

   1 import errno
   2 import hashlib
   3 import json
   4 import os.path
   5 import re
   6 import ssl
   7 import sys
   8 import types
   9
  10 import yt_dlp.extractor
  11 from yt_dlp import YoutubeDL
  12 from yt_dlp.utils import preferredencoding, try_call, write_string, find_available_port
  13
  14 if 'pytest' in sys.modules:
  15     import pytest
  16     is_download_test = pytest.mark.download
  17 else:
  18     def is_download_test(test_class):
  19         return test_class
  20
  21
  22 def get_params(override=None):
  23     PARAMETERS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
  24                                    'parameters.json')
  25     LOCAL_PARAMETERS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
  26                                          'local_parameters.json')
  27     with open(PARAMETERS_FILE, encoding='utf-8') as pf:
  28         parameters = json.load(pf)
  29     if os.path.exists(LOCAL_PARAMETERS_FILE):
  30         with open(LOCAL_PARAMETERS_FILE, encoding='utf-8') as pf:
  31             parameters.update(json.load(pf))
  32     if override:
  33         parameters.update(override)
  34     return parameters
  35
  36
  37 def try_rm(filename):
  38     """ Remove a file if it exists """
  39     try:
  40         os.remove(filename)
  41     except OSError as ose:
  42         if ose.errno != errno.ENOENT:
  43             raise
  44
  45
  46 def report_warning(message, *args, **kwargs):
  47     """
  48     Print the message to stderr, it will be prefixed with 'WARNING:'
  49     If stderr is a tty file the 'WARNING:' will be colored
  50     """
  51     if sys.stderr.isatty() and os.name != 'nt':
  52         _msg_header = '\033[0;33mWARNING:\033[0m'
  53     else:
  54         _msg_header = 'WARNING:'
  55     output = f'{_msg_header} {message}\n'
  56     if 'b' in getattr(sys.stderr, 'mode', ''):
  57         output = output.encode(preferredencoding())
  58     sys.stderr.write(output)
  59
  60
  61 class FakeYDL(YoutubeDL):
  62     def __init__(self, override=None):
  63         # Different instances of the downloader can't share the same dictionary
  64         # some test set the "sublang" parameter, which would break the md5 checks.
  65         params = get_params(override=override)
  66         super().__init__(params, auto_init=False)
  67         self.result = []
  68
  69     def to_screen(self, s, *args, **kwargs):
  70         print(s)
  71
  72     def trouble(self, s, *args, **kwargs):
  73         raise Exception(s)
  74
  75     def download(self, x):
  76         self.result.append(x)
  77
  78     def expect_warning(self, regex):
  79         # Silence an expected warning matching a regex
  80         old_report_warning = self.report_warning
  81
  82         def report_warning(self, message, *args, **kwargs):
  83             if re.match(regex, message):
  84                 return
  85             old_report_warning(message, *args, **kwargs)
  86         self.report_warning = types.MethodType(report_warning, self)
  87
  88
  89 def gettestcases(include_onlymatching=False):
  90     for ie in yt_dlp.extractor.gen_extractors():
  91         yield from ie.get_testcases(include_onlymatching)
  92
  93
  94 def getwebpagetestcases():
  95     for ie in yt_dlp.extractor.gen_extractors():
  96         for tc in ie.get_webpage_testcases():
  97             tc.setdefault('add_ie', []).append('Generic')
  98             yield tc
  99
 100
 101 md5 = lambda s: hashlib.md5(s.encode()).hexdigest()
 102
 103
 104 def expect_value(self, got, expected, field):
 105     if isinstance(expected, str) and expected.startswith('re:'):
 106         match_str = expected[len('re:'):]
 107         match_rex = re.compile(match_str)
 108
 109         self.assertTrue(
 110             isinstance(got, str),
 111             f'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
 112         self.assertTrue(
 113             match_rex.match(got),
 114             f'field {field} (value: {got!r}) should match {match_str!r}')
 115     elif isinstance(expected, str) and expected.startswith('startswith:'):
 116         start_str = expected[len('startswith:'):]
 117         self.assertTrue(
 118             isinstance(got, str),
 119             f'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
 120         self.assertTrue(
 121             got.startswith(start_str),
 122             f'field {field} (value: {got!r}) should start with {start_str!r}')
 123     elif isinstance(expected, str) and expected.startswith('contains:'):
 124         contains_str = expected[len('contains:'):]
 125         self.assertTrue(
 126             isinstance(got, str),
 127             f'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
 128         self.assertTrue(
 129             contains_str in got,
 130             f'field {field} (value: {got!r}) should contain {contains_str!r}')
 131     elif isinstance(expected, type):
 132         self.assertTrue(
 133             isinstance(got, expected),
 134             f'Expected type {expected!r} for field {field}, but got value {got!r} of type {type(got)!r}')
 135     elif isinstance(expected, dict) and isinstance(got, dict):
 136         expect_dict(self, got, expected)
 137     elif isinstance(expected, list) and isinstance(got, list):
 138         self.assertEqual(
 139             len(expected), len(got),
 140             f'Expect a list of length {len(expected)}, but got a list of length {len(got)} for field {field}')
 141         for index, (item_got, item_expected) in enumerate(zip(got, expected)):
 142             type_got = type(item_got)
 143             type_expected = type(item_expected)
 144             self.assertEqual(
 145                 type_expected, type_got,
 146                 f'Type mismatch for list item at index {index} for field {field}, '
 147                 f'expected {type_expected!r}, got {type_got!r}')
 148             expect_value(self, item_got, item_expected, field)
 149     else:
 150         if isinstance(expected, str) and expected.startswith('md5:'):
 151             self.assertTrue(
 152                 isinstance(got, str),
 153                 f'Expected field {field} to be a unicode object, but got value {got!r} of type {type(got)!r}')
 154             got = 'md5:' + md5(got)
 155         elif isinstance(expected, str) and re.match(r'^(?:min|max)?count:\d+', expected):
 156             self.assertTrue(
 157                 isinstance(got, (list, dict)),
 158                 f'Expected field {field} to be a list or a dict, but it is of type {type(got).__name__}')
 159             op, _, expected_num = expected.partition(':')
 160             expected_num = int(expected_num)
 161             if op == 'mincount':
 162                 assert_func = assertGreaterEqual
 163                 msg_tmpl = 'Expected %d items in field %s, but only got %d'
 164             elif op == 'maxcount':
 165                 assert_func = assertLessEqual
 166                 msg_tmpl = 'Expected maximum %d items in field %s, but got %d'
 167             elif op == 'count':
 168                 assert_func = assertEqual
 169                 msg_tmpl = 'Expected exactly %d items in field %s, but got %d'
 170             else:
 171                 assert False
 172             assert_func(
 173                 self, len(got), expected_num,
 174                 msg_tmpl % (expected_num, field, len(got)))
 175             return
 176         self.assertEqual(
 177             expected, got,
 178             f'Invalid value for field {field}, expected {expected!r}, got {got!r}')
 179
 180
 181 def expect_dict(self, got_dict, expected_dict):
 182     for info_field, expected in expected_dict.items():
 183         got = got_dict.get(info_field)
 184         expect_value(self, got, expected, info_field)
 185
 186
 187 def sanitize_got_info_dict(got_dict):
 188     IGNORED_FIELDS = (
 189         *YoutubeDL._format_fields,
 190
 191         # Lists
 192         'formats', 'thumbnails', 'subtitles', 'automatic_captions', 'comments', 'entries',
 193
 194         # Auto-generated
 195         'autonumber', 'playlist', 'format_index', 'video_ext', 'audio_ext', 'duration_string', 'epoch', 'n_entries',
 196         'fulltitle', 'extractor', 'extractor_key', 'filename', 'filepath', 'infojson_filename', 'original_url',
 197
 198         # Only live_status needs to be checked
 199         'is_live', 'was_live',
 200     )
 201
 202     IGNORED_PREFIXES = ('', 'playlist', 'requested', 'webpage')
 203
 204     def sanitize(key, value):
 205         if isinstance(value, str) and len(value) > 100 and key != 'thumbnail':
 206             return f'md5:{md5(value)}'
 207         elif isinstance(value, list) and len(value) > 10:
 208             return f'count:{len(value)}'
 209         elif key.endswith('_count') and isinstance(value, int):
 210             return int
 211         return value
 212
 213     test_info_dict = {
 214         key: sanitize(key, value) for key, value in got_dict.items()
 215         if value is not None and key not in IGNORED_FIELDS and (
 216             not any(key.startswith(f'{prefix}_') for prefix in IGNORED_PREFIXES)
 217             or key == '_old_archive_ids')
 218     }
 219
 220     # display_id may be generated from id
 221     if test_info_dict.get('display_id') == test_info_dict.get('id'):
 222         test_info_dict.pop('display_id')
 223
 224     # Remove deprecated fields
 225     for old in YoutubeDL._deprecated_multivalue_fields:
 226         test_info_dict.pop(old, None)
 227
 228     # release_year may be generated from release_date
 229     if try_call(lambda: test_info_dict['release_year'] == int(test_info_dict['release_date'][:4])):
 230         test_info_dict.pop('release_year')
 231
 232     # Check url for flat entries
 233     if got_dict.get('_type', 'video') != 'video' and got_dict.get('url'):
 234         test_info_dict['url'] = got_dict['url']
 235
 236     return test_info_dict
 237
 238
 239 def expect_info_dict(self, got_dict, expected_dict):
 240     expect_dict(self, got_dict, expected_dict)
 241     # Check for the presence of mandatory fields
 242     if got_dict.get('_type') not in ('playlist', 'multi_video'):
 243         mandatory_fields = ['id', 'title']
 244         if expected_dict.get('ext'):
 245             mandatory_fields.extend(('url', 'ext'))
 246         for key in mandatory_fields:
 247             self.assertTrue(got_dict.get(key), f'Missing mandatory field {key}')
 248     # Check for mandatory fields that are automatically set by YoutubeDL
 249     if got_dict.get('_type', 'video') == 'video':
 250         for key in ['webpage_url', 'extractor', 'extractor_key']:
 251             self.assertTrue(got_dict.get(key), f'Missing field: {key}')
 252
 253     test_info_dict = sanitize_got_info_dict(got_dict)
 254
 255     missing_keys = set(test_info_dict.keys()) - set(expected_dict.keys())
 256     if missing_keys:
 257         def _repr(v):
 258             if isinstance(v, str):
 259                 return "'{}'".format(v.replace('\\', '\\\\').replace("'", "\\'").replace('\n', '\\n'))
 260             elif isinstance(v, type):
 261                 return v.__name__
 262             else:
 263                 return repr(v)
 264         info_dict_str = ''.join(
 265             f'    {_repr(k)}: {_repr(v)},\n'
 266             for k, v in test_info_dict.items() if k not in missing_keys)
 267         if info_dict_str:
 268             info_dict_str += '\n'
 269         info_dict_str += ''.join(
 270             f'    {_repr(k)}: {_repr(test_info_dict[k])},\n'
 271             for k in missing_keys)
 272         info_dict_str = '\n\'info_dict\': {\n' + info_dict_str + '},\n'
 273         write_string(info_dict_str.replace('\n', '\n        '), out=sys.stderr)
 274         self.assertFalse(
 275             missing_keys,
 276             'Missing keys in test definition: {}'.format(', '.join(sorted(missing_keys))))
 277
 278
 279 def assertRegexpMatches(self, text, regexp, msg=None):
 280     if hasattr(self, 'assertRegexp'):
 281         return self.assertRegexp(text, regexp, msg)
 282     else:
 283         m = re.match(regexp, text)
 284         if not m:
 285             note = f'Regexp didn\'t match: {regexp!r} not found'
 286             if len(text) < 1000:
 287                 note += f' in {text!r}'
 288             if msg is None:
 289                 msg = note
 290             else:
 291                 msg = note + ', ' + msg
 292             self.assertTrue(m, msg)
 293
 294
 295 def assertGreaterEqual(self, got, expected, msg=None):
 296     if not (got >= expected):
 297         if msg is None:
 298             msg = f'{got!r} not greater than or equal to {expected!r}'
 299         self.assertTrue(got >= expected, msg)
 300
 301
 302 def assertLessEqual(self, got, expected, msg=None):
 303     if not (got <= expected):
 304         if msg is None:
 305             msg = f'{got!r} not less than or equal to {expected!r}'
 306         self.assertTrue(got <= expected, msg)
 307
 308
 309 def assertEqual(self, got, expected, msg=None):
 310     if got != expected:
 311         if msg is None:
 312             msg = f'{got!r} not equal to {expected!r}'
 313         self.assertTrue(got == expected, msg)
 314
 315
 316 def expect_warnings(ydl, warnings_re):
 317     real_warning = ydl.report_warning
 318
 319     def _report_warning(w, *args, **kwargs):
 320         if not any(re.search(w_re, w) for w_re in warnings_re):
 321             real_warning(w, *args, **kwargs)
 322
 323     ydl.report_warning = _report_warning
 324
 325
 326 def http_server_port(httpd):
 327     if os.name == 'java' and isinstance(httpd.socket, ssl.SSLSocket):
 328         # In Jython SSLSocket is not a subclass of socket.socket
 329         sock = httpd.socket.sock
 330     else:
 331         sock = httpd.socket
 332     return sock.getsockname()[1]
 333
 334
 335 def verify_address_availability(address):
 336     if find_available_port(address) is None:
 337         pytest.skip(f'Unable to bind to source address {address} (address may not exist)')
 338
 339
 340 def validate_and_send(rh, req):
 341     rh.validate(req)
 342     return rh.send(req)