]>
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 (
25 import youtube_dl
. YoutubeDL
26 from youtube_dl
. utils
import (
32 UnavailableVideoError
,
37 class YoutubeDL ( youtube_dl
. YoutubeDL
):
38 def __init__ ( self
, * args
, ** kwargs
):
39 self
. to_stderr
= self
. to_screen
40 self
. processed_info_dicts
= []
41 super ( YoutubeDL
, self
) .__ init
__ (* args
, ** kwargs
)
42 def report_warning ( self
, message
):
43 # Don't accept warnings during tests
44 raise ExtractorError ( message
)
45 def process_info ( self
, info_dict
):
46 self
. processed_info_dicts
. append ( info_dict
)
47 return super ( YoutubeDL
, self
). process_info ( info_dict
)
50 with open ( fn
, 'rb' ) as f
:
51 return hashlib
. md5 ( f
. read ()). hexdigest ()
53 defs
= get_testcases ()
56 class TestDownload ( unittest
. TestCase
):
61 ### Dynamically generate tests
62 def generator ( test_case
):
64 def test_template ( self
):
65 ie
= youtube_dl
. extractor
. get_info_extractor ( test_case
[ 'name' ])
66 def print_skipping ( reason
):
67 print ( 'Skipping %s : %s ' % ( test_case
[ 'name' ], reason
))
69 print_skipping ( 'IE marked as not _WORKING' )
71 if 'playlist' not in test_case
:
72 info_dict
= test_case
. get ( 'info_dict' , {})
73 if not test_case
. get ( 'file' ) and not ( info_dict
. get ( 'id' ) and info_dict
. get ( 'ext' )):
74 print_skipping ( 'The output file cannot be know, the "file" '
75 'key is missing or the info_dict is incomplete' )
77 if 'skip' in test_case
:
78 print_skipping ( test_case
[ 'skip' ])
81 params
= get_params ( test_case
. get ( 'params' , {}))
83 ydl
= YoutubeDL ( params
)
84 ydl
. add_default_info_extractors ()
85 finished_hook_called
= set ()
87 if status
[ 'status' ] == 'finished' :
88 finished_hook_called
. add ( status
[ 'filename' ])
89 ydl
. fd
. add_progress_hook ( _hook
)
91 def get_tc_filename ( tc
):
92 return tc
. get ( 'file' ) or ydl
. prepare_filename ( tc
. get ( 'info_dict' , {}))
94 test_cases
= test_case
. get ( 'playlist' , [ test_case
])
95 def try_rm_tcs_files ():
97 tc_filename
= get_tc_filename ( tc
)
99 try_rm ( tc_filename
+ '.part' )
100 try_rm ( tc_filename
+ '.info.json' )
106 ydl
. download ([ test_case
[ 'url' ]])
107 except ( DownloadError
, ExtractorError
) as err
:
108 # Check if the exception is not a network related one
109 if not err
. exc_info
[ 0 ] in ( compat_urllib_error
. URLError
, socket
. timeout
, UnavailableVideoError
) or ( err
. exc_info
[ 0 ] == compat_HTTPError
and err
. exc_info
[ 1 ]. code
== 503 ):
112 if try_num
== RETRIES
:
113 report_warning ( u
'Failed due to network errors, skipping...' )
116 print ( 'Retrying: {0} failed tries \n\n ########## \n\n ' . format ( try_num
))
122 for tc
in test_cases
:
123 tc_filename
= get_tc_filename ( tc
)
124 if not test_case
. get ( 'params' , {}). get ( 'skip_download' , False ):
125 self
. assertTrue ( os
. path
. exists ( tc_filename
), msg
= 'Missing file ' + tc_filename
)
126 self
. assertTrue ( tc_filename
in finished_hook_called
)
127 self
. assertTrue ( os
. path
. exists ( tc_filename
+ '.info.json' ))
129 md5_for_file
= _file_md5 ( tc_filename
)
130 self
. assertEqual ( md5_for_file
, tc
[ 'md5' ])
131 with io
. open ( tc_filename
+ '.info.json' , encoding
= 'utf-8' ) as infof
:
132 info_dict
= json
. load ( infof
)
133 for ( info_field
, expected
) in tc
. get ( 'info_dict' , {}). items ():
134 if isinstance ( expected
, compat_str
) and expected
. startswith ( 'md5:' ):
135 got
= 'md5:' + md5 ( info_dict
. get ( info_field
))
137 got
= info_dict
. get ( info_field
)
138 self
. assertEqual ( expected
, got
,
139 u
'invalid value for field %s , expected %r , got %r ' % ( info_field
, expected
, got
))
141 # If checkable fields are missing from the test case, print the info_dict
142 test_info_dict
= dict (( key
, value
if not isinstance ( value
, compat_str
) or len ( value
) < 250 else 'md5:' + md5 ( value
))
143 for key
, value
in info_dict
. items ()
144 if value
and key
in ( 'title' , 'description' , 'uploader' , 'upload_date' , 'uploader_id' , 'location' ))
145 if not all ( key
in tc
. get ( 'info_dict' , {}). keys () for key
in test_info_dict
. keys ()):
146 sys
. stderr
. write ( u
' \n "info_dict": ' + json
. dumps ( test_info_dict
, ensure_ascii
= False , indent
= 2 ) + u
' \n ' )
148 # Check for the presence of mandatory fields
149 for key
in ( 'id' , 'url' , 'title' , 'ext' ):
150 self
. assertTrue ( key
in info_dict
. keys () and info_dict
[ key
])
151 # Check for mandatory fields that are automatically set by YoutubeDL
152 for key
in [ 'webpage_url' , 'extractor' , 'extractor_key' ]:
153 self
. assertTrue ( info_dict
. get ( key
), u
'Missing field: %s ' % key
)
159 ### And add them to TestDownload
160 for n
, test_case
in enumerate ( defs
):
161 test_method
= generator ( test_case
)
162 tname
= 'test_' + str ( test_case
[ 'name' ])
164 while hasattr ( TestDownload
, tname
):
165 tname
= 'test_' + str ( test_case
[ 'name' ]) + '_' + str ( i
)
167 test_method
.__ name
__ = tname
168 setattr ( TestDownload
, test_method
.__ name
__ , test_method
)
172 if __name__
== '__main__' :