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.

128 lines
4.4 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. # General configuration (from __init__, not very elegant...)
  19. jar = compat_cookiejar.CookieJar()
  20. cookie_processor = compat_urllib_request.HTTPCookieProcessor(jar)
  21. proxy_handler = compat_urllib_request.ProxyHandler()
  22. opener = compat_urllib_request.build_opener(proxy_handler, cookie_processor, YoutubeDLHandler())
  23. compat_urllib_request.install_opener(opener)
  24. socket.setdefaulttimeout(10)
  25. def _try_rm(filename):
  26. """ Remove a file if it exists """
  27. try:
  28. os.remove(filename)
  29. except OSError as ose:
  30. if ose.errno != errno.ENOENT:
  31. raise
  32. class FileDownloader(youtube_dl.FileDownloader):
  33. def __init__(self, *args, **kwargs):
  34. self.to_stderr = self.to_screen
  35. self.processed_info_dicts = []
  36. return youtube_dl.FileDownloader.__init__(self, *args, **kwargs)
  37. def process_info(self, info_dict):
  38. self.processed_info_dicts.append(info_dict)
  39. return youtube_dl.FileDownloader.process_info(self, info_dict)
  40. def _file_md5(fn):
  41. with open(fn, 'rb') as f:
  42. return hashlib.md5(f.read()).hexdigest()
  43. with io.open(DEF_FILE, encoding='utf-8') as deff:
  44. defs = json.load(deff)
  45. with io.open(PARAMETERS_FILE, encoding='utf-8') as pf:
  46. parameters = json.load(pf)
  47. class TestDownload(unittest.TestCase):
  48. def setUp(self):
  49. self.parameters = parameters
  50. self.defs = defs
  51. ### Dynamically generate tests
  52. def generator(test_case):
  53. def test_template(self):
  54. ie = getattr(youtube_dl.InfoExtractors, test_case['name'] + 'IE')
  55. if not ie._WORKING:
  56. print('Skipping: IE marked as not _WORKING')
  57. return
  58. if 'playlist' not in test_case and not test_case['file']:
  59. print('Skipping: No output file specified')
  60. return
  61. if 'skip' in test_case:
  62. print('Skipping: {0}'.format(test_case['skip']))
  63. return
  64. params = self.parameters.copy()
  65. params.update(test_case.get('params', {}))
  66. fd = FileDownloader(params)
  67. fd.add_info_extractor(ie())
  68. for ien in test_case.get('add_ie', []):
  69. fd.add_info_extractor(getattr(youtube_dl.InfoExtractors, ien + 'IE')())
  70. finished_hook_called = set()
  71. def _hook(status):
  72. if status['status'] == 'finished':
  73. finished_hook_called.add(status['filename'])
  74. fd.add_progress_hook(_hook)
  75. test_cases = test_case.get('playlist', [test_case])
  76. for tc in test_cases:
  77. _try_rm(tc['file'])
  78. _try_rm(tc['file'] + '.part')
  79. _try_rm(tc['file'] + '.info.json')
  80. try:
  81. fd.download([test_case['url']])
  82. for tc in test_cases:
  83. if not test_case.get('params', {}).get('skip_download', False):
  84. self.assertTrue(os.path.exists(tc['file']), msg='Missing file ' + tc['file'])
  85. self.assertTrue(tc['file'] in finished_hook_called)
  86. self.assertTrue(os.path.exists(tc['file'] + '.info.json'))
  87. if 'md5' in tc:
  88. md5_for_file = _file_md5(tc['file'])
  89. self.assertEqual(md5_for_file, tc['md5'])
  90. with io.open(tc['file'] + '.info.json', encoding='utf-8') as infof:
  91. info_dict = json.load(infof)
  92. for (info_field, value) in tc.get('info_dict', {}).items():
  93. self.assertEqual(value, info_dict.get(info_field))
  94. finally:
  95. for tc in test_cases:
  96. _try_rm(tc['file'])
  97. _try_rm(tc['file'] + '.part')
  98. _try_rm(tc['file'] + '.info.json')
  99. return test_template
  100. ### And add them to TestDownload
  101. for test_case in defs:
  102. test_method = generator(test_case)
  103. test_method.__name__ = "test_{0}".format(test_case["name"])
  104. setattr(TestDownload, test_method.__name__, test_method)
  105. del test_method
  106. if __name__ == '__main__':
  107. unittest.main()