13 # Allow direct execution
14 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
16 import youtube_dl.FileDownloader
17 import youtube_dl.InfoExtractors
18 from youtube_dl.utils import *
20 DEF_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'tests.json')
21 PARAMETERS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "parameters.json")
23 # General configuration (from __init__, not very elegant...)
24 jar = compat_cookiejar.CookieJar()
25 cookie_processor = compat_urllib_request.HTTPCookieProcessor(jar)
26 proxy_handler = compat_urllib_request.ProxyHandler()
27 opener = compat_urllib_request.build_opener(proxy_handler, cookie_processor, YoutubeDLHandler())
28 compat_urllib_request.install_opener(opener)
29 socket.setdefaulttimeout(10)
31 def _try_rm(filename):
32 """ Remove a file if it exists """
35 except OSError as ose:
36 if ose.errno != errno.ENOENT:
39 class FileDownloader(youtube_dl.FileDownloader):
40 def __init__(self, *args, **kwargs):
41 self.to_stderr = self.to_screen
42 self.processed_info_dicts = []
43 return youtube_dl.FileDownloader.__init__(self, *args, **kwargs)
44 def process_info(self, info_dict):
45 self.processed_info_dicts.append(info_dict)
46 return youtube_dl.FileDownloader.process_info(self, info_dict)
49 with open(fn, 'rb') as f:
50 return hashlib.md5(f.read()).hexdigest()
52 with io.open(DEF_FILE, encoding='utf-8') as deff:
53 defs = json.load(deff)
54 with io.open(PARAMETERS_FILE, encoding='utf-8') as pf:
55 parameters = json.load(pf)
58 class TestDownload(unittest.TestCase):
60 self.parameters = parameters
63 ### Dynamically generate tests
64 def generator(test_case):
66 def test_template(self):
67 ie = getattr(youtube_dl.InfoExtractors, test_case['name'] + 'IE')
69 print('Skipping: IE marked as not _WORKING')
71 if 'playlist' not in test_case and not test_case['file']:
72 print('Skipping: No output file specified')
74 if 'skip' in test_case:
75 print('Skipping: {0}'.format(test_case['skip']))
78 params = self.parameters.copy()
79 params.update(test_case.get('params', {}))
81 fd = FileDownloader(params)
82 fd.add_info_extractor(ie())
83 for ien in test_case.get('add_ie', []):
84 fd.add_info_extractor(getattr(youtube_dl.InfoExtractors, ien + 'IE')())
85 finished_hook_called = set()
87 if status['status'] == 'finished':
88 finished_hook_called.add(status['filename'])
89 fd.add_progress_hook(_hook)
91 test_cases = test_case.get('playlist', [test_case])
94 _try_rm(tc['file'] + '.part')
95 _try_rm(tc['file'] + '.info.json')
97 fd.download([test_case['url']])
100 if not test_case.get('params', {}).get('skip_download', False):
101 self.assertTrue(os.path.exists(tc['file']))
102 self.assertTrue(tc['file'] in finished_hook_called)
103 self.assertTrue(os.path.exists(tc['file'] + '.info.json'))
105 md5_for_file = _file_md5(tc['file'])
106 self.assertEqual(md5_for_file, tc['md5'])
107 with io.open(tc['file'] + '.info.json', encoding='utf-8') as infof:
108 info_dict = json.load(infof)
109 for (info_field, value) in tc.get('info_dict', {}).items():
110 self.assertEqual(value, info_dict.get(info_field))
112 for tc in test_cases:
114 _try_rm(tc['file'] + '.part')
115 _try_rm(tc['file'] + '.info.json')
119 ### And add them to TestDownload
120 for test_case in defs:
121 test_method = generator(test_case)
122 test_method.__name__ = "test_{0}".format(test_case["name"])
123 setattr(TestDownload, test_method.__name__, test_method)
127 if __name__ == '__main__':