You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

142 lines
4.9 KiB

12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
  1. #!/usr/bin/env python
  2. import errno
  3. import hashlib
  4. import io
  5. import os
  6. import json
  7. import unittest
  8. import sys
  9. import hashlib
  10. import socket
  11. # Allow direct execution
  12. sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  13. import youtube_dl.FileDownloader
  14. import youtube_dl.InfoExtractors
  15. from youtube_dl.utils import *
  16. DEF_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'tests.json')
  17. PARAMETERS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "parameters.json")
  18. RETRIES = 3
  19. # General configuration (from __init__, not very elegant...)
  20. jar = compat_cookiejar.CookieJar()
  21. cookie_processor = compat_urllib_request.HTTPCookieProcessor(jar)
  22. proxy_handler = compat_urllib_request.ProxyHandler()
  23. opener = compat_urllib_request.build_opener(proxy_handler, cookie_processor, YoutubeDLHandler())
  24. compat_urllib_request.install_opener(opener)
  25. socket.setdefaulttimeout(10)
  26. def _try_rm(filename):
  27. """ Remove a file if it exists """
  28. try:
  29. os.remove(filename)
  30. except OSError as ose:
  31. if ose.errno != errno.ENOENT:
  32. raise
  33. class FileDownloader(youtube_dl.FileDownloader):
  34. def __init__(self, *args, **kwargs):
  35. self.to_stderr = self.to_screen
  36. self.processed_info_dicts = []
  37. return youtube_dl.FileDownloader.__init__(self, *args, **kwargs)
  38. def process_info(self, info_dict):
  39. self.processed_info_dicts.append(info_dict)
  40. return youtube_dl.FileDownloader.process_info(self, info_dict)
  41. def _file_md5(fn):
  42. with open(fn, 'rb') as f:
  43. return hashlib.md5(f.read()).hexdigest()
  44. with io.open(DEF_FILE, encoding='utf-8') as deff:
  45. defs = json.load(deff)
  46. with io.open(PARAMETERS_FILE, encoding='utf-8') as pf:
  47. parameters = json.load(pf)
  48. class TestDownload(unittest.TestCase):
  49. maxDiff = None
  50. def setUp(self):
  51. self.parameters = parameters
  52. self.defs = defs
  53. ### Dynamically generate tests
  54. def generator(test_case):
  55. def test_template(self):
  56. ie = youtube_dl.InfoExtractors.get_info_extractor(test_case['name'])
  57. if not ie._WORKING:
  58. print('Skipping: IE marked as not _WORKING')
  59. return
  60. if 'playlist' not in test_case and not test_case['file']:
  61. print('Skipping: No output file specified')
  62. return
  63. if 'skip' in test_case:
  64. print('Skipping: {0}'.format(test_case['skip']))
  65. return
  66. params = self.parameters.copy()
  67. params.update(test_case.get('params', {}))
  68. fd = FileDownloader(params)
  69. for ie in youtube_dl.InfoExtractors.gen_extractors():
  70. fd.add_info_extractor(ie)
  71. finished_hook_called = set()
  72. def _hook(status):
  73. if status['status'] == 'finished':
  74. finished_hook_called.add(status['filename'])
  75. fd.add_progress_hook(_hook)
  76. test_cases = test_case.get('playlist', [test_case])
  77. for tc in test_cases:
  78. _try_rm(tc['file'])
  79. _try_rm(tc['file'] + '.part')
  80. _try_rm(tc['file'] + '.info.json')
  81. try:
  82. for retry in range(1, RETRIES + 1):
  83. try:
  84. fd.download([test_case['url']])
  85. except (DownloadError, ExtractorError) as err:
  86. if retry == RETRIES: raise
  87. # Check if the exception is not a network related one
  88. if not err.exc_info[0] in (compat_urllib_error.URLError, socket.timeout, UnavailableVideoError):
  89. raise
  90. print('Retrying: {0} failed tries\n\n##########\n\n'.format(retry))
  91. else:
  92. break
  93. for tc in test_cases:
  94. if not test_case.get('params', {}).get('skip_download', False):
  95. self.assertTrue(os.path.exists(tc['file']), msg='Missing file ' + tc['file'])
  96. self.assertTrue(tc['file'] in finished_hook_called)
  97. self.assertTrue(os.path.exists(tc['file'] + '.info.json'))
  98. if 'md5' in tc:
  99. md5_for_file = _file_md5(tc['file'])
  100. self.assertEqual(md5_for_file, tc['md5'])
  101. with io.open(tc['file'] + '.info.json', encoding='utf-8') as infof:
  102. info_dict = json.load(infof)
  103. for (info_field, value) in tc.get('info_dict', {}).items():
  104. self.assertEqual(value, info_dict.get(info_field))
  105. finally:
  106. for tc in test_cases:
  107. _try_rm(tc['file'])
  108. _try_rm(tc['file'] + '.part')
  109. _try_rm(tc['file'] + '.info.json')
  110. return test_template
  111. ### And add them to TestDownload
  112. for test_case in defs:
  113. test_method = generator(test_case)
  114. test_method.__name__ = "test_{0}".format(test_case["name"])
  115. setattr(TestDownload, test_method.__name__, test_method)
  116. del test_method
  117. if __name__ == '__main__':
  118. unittest.main()