]> jfr.im git - yt-dlp.git/blobdiff - test/helper.py
[ie/brightcove] Upgrade requests to HTTPS (#10202)
[yt-dlp.git] / test / helper.py
index d940e327cf5a29ee96e8169a4749fd002d444788..3b550d19271967897c536ffdd27f1ee7370bc676 100644 (file)
@@ -3,28 +3,21 @@
 import json
 import os.path
 import re
-import types
 import ssl
 import sys
+import types
 
 import yt_dlp.extractor
 from yt_dlp import YoutubeDL
-from yt_dlp.compat import (
-    compat_os_name,
-    compat_str,
-)
-from yt_dlp.utils import (
-    preferredencoding,
-    write_string,
-)
-
+from yt_dlp.compat import compat_os_name
+from yt_dlp.utils import preferredencoding, try_call, write_string, find_available_port
 
 if 'pytest' in sys.modules:
     import pytest
     is_download_test = pytest.mark.download
 else:
-    def is_download_test(testClass):
-        return testClass
+    def is_download_test(test_class):
+        return test_class
 
 
 def get_params(override=None):
@@ -51,11 +44,11 @@ def try_rm(filename):
             raise
 
 
-def report_warning(message):
-    '''
+def report_warning(message, *args, **kwargs):
+    """
     Print the message to stderr, it will be prefixed with 'WARNING:'
     If stderr is a tty file the 'WARNING:' will be colored
-    '''
+    """
     if sys.stderr.isatty() and compat_os_name != 'nt':
         _msg_header = '\033[0;33mWARNING:\033[0m'
     else:
@@ -74,10 +67,10 @@ def __init__(self, override=None):
         super().__init__(params, auto_init=False)
         self.result = []
 
-    def to_screen(self, s, skip_eol=None):
+    def to_screen(self, s, *args, **kwargs):
         print(s)
 
-    def trouble(self, s, tb=None):
+    def trouble(self, s, *args, **kwargs):
         raise Exception(s)
 
     def download(self, x):
@@ -87,10 +80,10 @@ def expect_warning(self, regex):
         # Silence an expected warning matching a regex
         old_report_warning = self.report_warning
 
-        def report_warning(self, message):
+        def report_warning(self, message, *args, **kwargs):
             if re.match(regex, message):
                 return
-            old_report_warning(message)
+            old_report_warning(message, *args, **kwargs)
         self.report_warning = types.MethodType(report_warning, self)
 
 
@@ -99,33 +92,40 @@ def gettestcases(include_onlymatching=False):
         yield from ie.get_testcases(include_onlymatching)
 
 
-md5 = lambda s: hashlib.md5(s.encode('utf-8')).hexdigest()
+def getwebpagetestcases():
+    for ie in yt_dlp.extractor.gen_extractors():
+        for tc in ie.get_webpage_testcases():
+            tc.setdefault('add_ie', []).append('Generic')
+            yield tc
+
+
+md5 = lambda s: hashlib.md5(s.encode()).hexdigest()
 
 
 def expect_value(self, got, expected, field):
-    if isinstance(expected, compat_str) and expected.startswith('re:'):
+    if isinstance(expected, str) and expected.startswith('re:'):
         match_str = expected[len('re:'):]
         match_rex = re.compile(match_str)
 
         self.assertTrue(
-            isinstance(got, compat_str),
-            f'Expected a {compat_str.__name__} object, but got {type(got).__name__} for field {field}')
+            isinstance(got, str),
+            f'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
         self.assertTrue(
             match_rex.match(got),
             f'field {field} (value: {got!r}) should match {match_str!r}')
-    elif isinstance(expected, compat_str) and expected.startswith('startswith:'):
+    elif isinstance(expected, str) and expected.startswith('startswith:'):
         start_str = expected[len('startswith:'):]
         self.assertTrue(
-            isinstance(got, compat_str),
-            f'Expected a {compat_str.__name__} object, but got {type(got).__name__} for field {field}')
+            isinstance(got, str),
+            f'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
         self.assertTrue(
             got.startswith(start_str),
             f'field {field} (value: {got!r}) should start with {start_str!r}')
-    elif isinstance(expected, compat_str) and expected.startswith('contains:'):
+    elif isinstance(expected, str) and expected.startswith('contains:'):
         contains_str = expected[len('contains:'):]
         self.assertTrue(
-            isinstance(got, compat_str),
-            f'Expected a {compat_str.__name__} object, but got {type(got).__name__} for field {field}')
+            isinstance(got, str),
+            f'Expected a {str.__name__} object, but got {type(got).__name__} for field {field}')
         self.assertTrue(
             contains_str in got,
             f'field {field} (value: {got!r}) should contain {contains_str!r}')
@@ -138,23 +138,22 @@ def expect_value(self, got, expected, field):
     elif isinstance(expected, list) and isinstance(got, list):
         self.assertEqual(
             len(expected), len(got),
-            'Expect a list of length %d, but got a list of length %d for field %s' % (
-                len(expected), len(got), field))
+            f'Expect a list of length {len(expected)}, but got a list of length {len(got)} for field {field}')
         for index, (item_got, item_expected) in enumerate(zip(got, expected)):
             type_got = type(item_got)
             type_expected = type(item_expected)
             self.assertEqual(
                 type_expected, type_got,
-                'Type mismatch for list item at index %d for field %s, expected %r, got %r' % (
-                    index, field, type_expected, type_got))
+                f'Type mismatch for list item at index {index} for field {field}, '
+                f'expected {type_expected!r}, got {type_got!r}')
             expect_value(self, item_got, item_expected, field)
     else:
-        if isinstance(expected, compat_str) and expected.startswith('md5:'):
+        if isinstance(expected, str) and expected.startswith('md5:'):
             self.assertTrue(
-                isinstance(got, compat_str),
+                isinstance(got, str),
                 f'Expected field {field} to be a unicode object, but got value {got!r} of type {type(got)!r}')
             got = 'md5:' + md5(got)
-        elif isinstance(expected, compat_str) and re.match(r'^(?:min|max)?count:\d+', expected):
+        elif isinstance(expected, str) and re.match(r'^(?:min|max)?count:\d+', expected):
             self.assertTrue(
                 isinstance(got, (list, dict)),
                 f'Expected field {field} to be a list or a dict, but it is of type {type(got).__name__}')
@@ -194,8 +193,8 @@ def sanitize_got_info_dict(got_dict):
         'formats', 'thumbnails', 'subtitles', 'automatic_captions', 'comments', 'entries',
 
         # Auto-generated
-        'autonumber', 'playlist', 'format_index', 'video_ext', 'audio_ext', 'duration_string', 'epoch',
-        'fulltitle', 'extractor', 'extractor_key', 'filepath', 'infojson_filename', 'original_url', 'n_entries',
+        'autonumber', 'playlist', 'format_index', 'video_ext', 'audio_ext', 'duration_string', 'epoch', 'n_entries',
+        'fulltitle', 'extractor', 'extractor_key', 'filename', 'filepath', 'infojson_filename', 'original_url',
 
         # Only live_status needs to be checked
         'is_live', 'was_live',
@@ -214,14 +213,27 @@ def sanitize(key, value):
 
     test_info_dict = {
         key: sanitize(key, value) for key, value in got_dict.items()
-        if value is not None and key not in IGNORED_FIELDS and not any(
-            key.startswith(f'{prefix}_') for prefix in IGNORED_PREFIXES)
+        if value is not None and key not in IGNORED_FIELDS and (
+            not any(key.startswith(f'{prefix}_') for prefix in IGNORED_PREFIXES)
+            or key == '_old_archive_ids')
     }
 
     # display_id may be generated from id
     if test_info_dict.get('display_id') == test_info_dict.get('id'):
         test_info_dict.pop('display_id')
 
+    # Remove deprecated fields
+    for old in YoutubeDL._deprecated_multivalue_fields:
+        test_info_dict.pop(old, None)
+
+    # release_year may be generated from release_date
+    if try_call(lambda: test_info_dict['release_year'] == int(test_info_dict['release_date'][:4])):
+        test_info_dict.pop('release_year')
+
+    # Check url for flat entries
+    if got_dict.get('_type', 'video') != 'video' and got_dict.get('url'):
+        test_info_dict['url'] = got_dict['url']
+
     return test_info_dict
 
 
@@ -233,39 +245,36 @@ def expect_info_dict(self, got_dict, expected_dict):
         if expected_dict.get('ext'):
             mandatory_fields.extend(('url', 'ext'))
         for key in mandatory_fields:
-            self.assertTrue(got_dict.get(key), 'Missing mandatory field %s' % key)
+            self.assertTrue(got_dict.get(key), f'Missing mandatory field {key}')
     # Check for mandatory fields that are automatically set by YoutubeDL
-    for key in ['webpage_url', 'extractor', 'extractor_key']:
-        self.assertTrue(got_dict.get(key), 'Missing field: %s' % key)
+    if got_dict.get('_type', 'video') == 'video':
+        for key in ['webpage_url', 'extractor', 'extractor_key']:
+            self.assertTrue(got_dict.get(key), f'Missing field: {key}')
 
     test_info_dict = sanitize_got_info_dict(got_dict)
 
     missing_keys = set(test_info_dict.keys()) - set(expected_dict.keys())
     if missing_keys:
         def _repr(v):
-            if isinstance(v, compat_str):
-                return "'%s'" % v.replace('\\', '\\\\').replace("'", "\\'").replace('\n', '\\n')
+            if isinstance(v, str):
+                return "'{}'".format(v.replace('\\', '\\\\').replace("'", "\\'").replace('\n', '\\n'))
             elif isinstance(v, type):
                 return v.__name__
             else:
                 return repr(v)
-        info_dict_str = ''
-        if len(missing_keys) != len(expected_dict):
-            info_dict_str += ''.join(
-                f'    {_repr(k)}: {_repr(v)},\n'
-                for k, v in test_info_dict.items() if k not in missing_keys)
-
-            if info_dict_str:
-                info_dict_str += '\n'
+        info_dict_str = ''.join(
+            f'    {_repr(k)}: {_repr(v)},\n'
+            for k, v in test_info_dict.items() if k not in missing_keys)
+        if info_dict_str:
+            info_dict_str += '\n'
         info_dict_str += ''.join(
             f'    {_repr(k)}: {_repr(test_info_dict[k])},\n'
             for k in missing_keys)
-        write_string(
-            '\n\'info_dict\': {\n' + info_dict_str + '},\n', out=sys.stderr)
+        info_dict_str = '\n\'info_dict\': {\n' + info_dict_str + '},\n'
+        write_string(info_dict_str.replace('\n', '\n        '), out=sys.stderr)
         self.assertFalse(
             missing_keys,
-            'Missing keys in test definition: %s' % (
-                ', '.join(sorted(missing_keys))))
+            'Missing keys in test definition: {}'.format(', '.join(sorted(missing_keys))))
 
 
 def assertRegexpMatches(self, text, regexp, msg=None):
@@ -274,9 +283,9 @@ def assertRegexpMatches(self, text, regexp, msg=None):
     else:
         m = re.match(regexp, text)
         if not m:
-            note = 'Regexp didn\'t match: %r not found' % (regexp)
+            note = f'Regexp didn\'t match: {regexp!r} not found'
             if len(text) < 1000:
-                note += ' in %r' % text
+                note += f' in {text!r}'
             if msg is None:
                 msg = note
             else:
@@ -299,7 +308,7 @@ def assertLessEqual(self, got, expected, msg=None):
 
 
 def assertEqual(self, got, expected, msg=None):
-    if not (got == expected):
+    if got != expected:
         if msg is None:
             msg = f'{got!r} not equal to {expected!r}'
         self.assertTrue(got == expected, msg)
@@ -308,9 +317,9 @@ def assertEqual(self, got, expected, msg=None):
 def expect_warnings(ydl, warnings_re):
     real_warning = ydl.report_warning
 
-    def _report_warning(w):
+    def _report_warning(w, *args, **kwargs):
         if not any(re.search(w_re, w) for w_re in warnings_re):
-            real_warning(w)
+            real_warning(w, *args, **kwargs)
 
     ydl.report_warning = _report_warning
 
@@ -322,3 +331,13 @@ def http_server_port(httpd):
     else:
         sock = httpd.socket
     return sock.getsockname()[1]
+
+
+def verify_address_availability(address):
+    if find_available_port(address) is None:
+        pytest.skip(f'Unable to bind to source address {address} (address may not exist)')
+
+
+def validate_and_send(rh, req):
+    rh.validate(req)
+    return rh.send(req)