]>
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 get_params
, get_testcases
, global_setup
, try_rm
, md5
18 import youtube_dl
.YoutubeDL
19 from youtube_dl
.utils
import (
24 UnavailableVideoError
,
29 class YoutubeDL(youtube_dl
.YoutubeDL
):
30 def __init__(self
, *args
, **kwargs
):
31 self
.to_stderr
= self
.to_screen
32 self
.processed_info_dicts
= []
33 super(YoutubeDL
, self
).__init
__(*args
, **kwargs
)
34 def report_warning(self
, message
):
35 # Don't accept warnings during tests
36 raise ExtractorError(message
)
37 def process_info(self
, info_dict
):
38 self
.processed_info_dicts
.append(info_dict
)
39 return super(YoutubeDL
, self
).process_info(info_dict
)
42 with open(fn
, 'rb') as f
:
43 return hashlib
.md5(f
.read()).hexdigest()
45 defs
= get_testcases()
48 class TestDownload(unittest
.TestCase
):
53 ### Dynamically generate tests
54 def generator(test_case
):
56 def test_template(self
):
57 ie
= youtube_dl
.extractor
.get_info_extractor(test_case
['name'])
58 def print_skipping(reason
):
59 print('Skipping %s: %s' % (test_case
['name'], reason
))
61 print_skipping('IE marked as not _WORKING')
63 if 'playlist' not in test_case
:
64 info_dict
= test_case
.get('info_dict', {})
65 if not test_case
.get('file') and not (info_dict
.get('id') and info_dict
.get('ext')):
66 print_skipping('The output file cannot be know, the "file" '
67 'key is missing or the info_dict is incomplete')
69 if 'skip' in test_case
:
70 print_skipping(test_case
['skip'])
73 params
= get_params(test_case
.get('params', {}))
75 ydl
= YoutubeDL(params
)
76 ydl
.add_default_info_extractors()
77 finished_hook_called
= set()
79 if status
['status'] == 'finished':
80 finished_hook_called
.add(status
['filename'])
81 ydl
.fd
.add_progress_hook(_hook
)
83 def get_tc_filename(tc
):
84 return tc
.get('file') or ydl
.prepare_filename(tc
.get('info_dict', {}))
86 test_cases
= test_case
.get('playlist', [test_case
])
87 def try_rm_tcs_files():
89 tc_filename
= get_tc_filename(tc
)
91 try_rm(tc_filename
+ '.part')
92 try_rm(tc_filename
+ '.info.json')
95 for retry
in range(1, RETRIES
+ 1):
97 ydl
.download([test_case
['url']])
98 except (DownloadError
, ExtractorError
) as err
:
99 if retry
== RETRIES
: raise
101 # Check if the exception is not a network related one
102 if not err
.exc_info
[0] in (compat_urllib_error
.URLError
, socket
.timeout
, UnavailableVideoError
):
105 print('Retrying: {0} failed tries\n\n##########\n\n'.format(retry
))
109 for tc
in test_cases
:
110 tc_filename
= get_tc_filename(tc
)
111 if not test_case
.get('params', {}).get('skip_download', False):
112 self
.assertTrue(os
.path
.exists(tc_filename
), msg
='Missing file ' + tc_filename
)
113 self
.assertTrue(tc_filename
in finished_hook_called
)
114 self
.assertTrue(os
.path
.exists(tc_filename
+ '.info.json'))
116 md5_for_file
= _file_md5(tc_filename
)
117 self
.assertEqual(md5_for_file
, tc
['md5'])
118 with io
.open(tc_filename
+ '.info.json', encoding
='utf-8') as infof
:
119 info_dict
= json
.load(infof
)
120 for (info_field
, expected
) in tc
.get('info_dict', {}).items():
121 if isinstance(expected
, compat_str
) and expected
.startswith('md5:'):
122 got
= 'md5:' + md5(info_dict
.get(info_field
))
124 got
= info_dict
.get(info_field
)
125 self
.assertEqual(expected
, got
,
126 u
'invalid value for field %s, expected %r, got %r' % (info_field
, expected
, got
))
128 # If checkable fields are missing from the test case, print the info_dict
129 test_info_dict
= dict((key
, value
if not isinstance(value
, compat_str
) or len(value
) < 250 else 'md5:' + md5(value
))
130 for key
, value
in info_dict
.items()
131 if value
and key
in ('title', 'description', 'uploader', 'upload_date', 'uploader_id', 'location'))
132 if not all(key
in tc
.get('info_dict', {}).keys() for key
in test_info_dict
.keys()):
133 sys
.stderr
.write(u
'\n"info_dict": ' + json
.dumps(test_info_dict
, ensure_ascii
=False, indent
=2) + u
'\n')
135 # Check for the presence of mandatory fields
136 for key
in ('id', 'url', 'title', 'ext'):
137 self
.assertTrue(key
in info_dict
.keys() and info_dict
[key
])
143 ### And add them to TestDownload
144 for n
, test_case
in enumerate(defs
):
145 test_method
= generator(test_case
)
146 tname
= 'test_' + str(test_case
['name'])
148 while hasattr(TestDownload
, tname
):
149 tname
= 'test_' + str(test_case
['name']) + '_' + str(i
)
151 test_method
.__name
__ = tname
152 setattr(TestDownload
, test_method
.__name
__, test_method
)
156 if __name__
== '__main__':