test/test_download.py

   1 #!/usr/bin/env python
   2
   3 import hashlib
   4 import io
   5 import os
   6 import json
   7 import unittest
   8 import sys
   9 import socket
  10
  11 # Allow direct execution
  12 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  13
  14 import youtube_dl.FileDownloader
  15 import youtube_dl.InfoExtractors
  16 from youtube_dl.utils import *
  17
  18 DEF_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'tests.json')
  19 PARAMETERS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "parameters.json")
  20
  21 # General configuration (from __init__, not very elegant...)
  22 jar = compat_cookiejar.CookieJar()
  23 cookie_processor = compat_urllib_request.HTTPCookieProcessor(jar)
  24 proxy_handler = compat_urllib_request.ProxyHandler()
  25 opener = compat_urllib_request.build_opener(proxy_handler, cookie_processor, YoutubeDLHandler())
  26 compat_urllib_request.install_opener(opener)
  27 socket.setdefaulttimeout(300) # 5 minutes should be enough (famous last words)
  28
  29 class FileDownloader(youtube_dl.FileDownloader):
  30     def __init__(self, *args, **kwargs):
  31         youtube_dl.FileDownloader.__init__(self, *args, **kwargs)
  32         self.to_stderr = self.to_screen
  33
  34 def _file_md5(fn):
  35     with open(fn, 'rb') as f:
  36         return hashlib.md5(f.read()).hexdigest()
  37
  38 with io.open(DEF_FILE, encoding='utf-8') as deff:
  39     defs = json.load(deff)
  40 with io.open(PARAMETERS_FILE, encoding='utf-8') as pf:
  41     parameters = json.load(pf)
  42
  43 class TestDownload(unittest.TestCase):
  44     def setUp(self):
  45         self.parameters = parameters
  46         self.defs = defs
  47
  48         # Clear old files
  49         self.tearDown()
  50
  51     def tearDown(self):
  52         for fn in [ test.get('file', False) for test in self.defs ]:
  53             if fn and os.path.exists(fn):
  54                 os.remove(fn)
  55
  56
  57 ### Dinamically generate tests
  58 def generator(test_case):
  59
  60     def test_template(self):
  61         ie = getattr(youtube_dl.InfoExtractors, test_case['name'] + 'IE')
  62         if not ie._WORKING:
  63             print('Skipping: IE marked as not _WORKING')
  64             return
  65         if not test_case['file']:
  66             print('Skipping: No output file specified')
  67             return
  68         if 'skip' in test_case:
  69             print('Skipping: {0}'.format(test_case['skip']))
  70             return
  71         params = dict(self.parameters) # Duplicate it locally
  72         for p in test_case.get('params', {}):
  73             params[p] = test_case['params'][p]
  74         fd = FileDownloader(params)
  75         fd.add_info_extractor(ie())
  76         for ien in test_case.get('add_ie', []):
  77             fd.add_info_extractor(getattr(youtube_dl.InfoExtractors, ien + 'IE')())
  78         fd.download([test_case['url']])
  79         self.assertTrue(os.path.exists(test_case['file']))
  80         if 'md5' in test_case:
  81             md5_for_file = _file_md5(test_case['file'])
  82             self.assertEqual(md5_for_file, test_case['md5'])
  83
  84     return test_template
  85
  86 ### And add them to TestDownload
  87 for test_case in defs:
  88     test_method = generator(test_case)
  89     test_method.__name__ = "test_{0}".format(test_case["name"])
  90     setattr(TestDownload, test_method.__name__, test_method)
  91     del test_method
  92
  93
  94 if __name__ == '__main__':
  95     unittest.main()