- fd.download([test_case['url']])
- self.assertTrue(os.path.exists(test_case['file']))
- if 'md5' in test_case:
- md5_for_file = _file_md5(test_case['file'])
- self.assertEqual(md5_for_file, test_case['md5'])
+ finished_hook_called = set()
+ def _hook(status):
+ if status['status'] == 'finished':
+ finished_hook_called.add(status['filename'])
+ fd.add_progress_hook(_hook)
+
+ test_cases = test_case.get('playlist', [test_case])
+ for tc in test_cases:
+ _try_rm(tc['file'])
+ _try_rm(tc['file'] + '.part')
+ _try_rm(tc['file'] + '.info.json')
+ try:
+ fd.download([test_case['url']])
+
+ for tc in test_cases:
+ if not test_case.get('params', {}).get('skip_download', False):
+ self.assertTrue(os.path.exists(tc['file']), msg='Missing file ' + tc['file'])
+ self.assertTrue(tc['file'] in finished_hook_called)
+ self.assertTrue(os.path.exists(tc['file'] + '.info.json'))
+ if 'md5' in tc:
+ md5_for_file = _file_md5(tc['file'])
+ self.assertEqual(md5_for_file, tc['md5'])
+ with io.open(tc['file'] + '.info.json', encoding='utf-8') as infof:
+ info_dict = json.load(infof)
+ for (info_field, value) in tc.get('info_dict', {}).items():
+ self.assertEqual(value, info_dict.get(info_field))
+ finally:
+ for tc in test_cases:
+ _try_rm(tc['file'])
+ _try_rm(tc['file'] + '.part')
+ _try_rm(tc['file'] + '.info.json')