logo

youtube-dl

[mirror] Download/Watch videos from video hostersgit clone https://hacktivis.me/git/mirror/youtube-dl.git

test_downloader_external.py (8635B)

    

  1. #!/usr/bin/env python
  2. # coding: utf-8
  3. from __future__ import unicode_literals
  5. # Allow direct execution
  6. import os
  7. import re
  8. import sys
  9. import subprocess
  10. import unittest
  11. sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  13. from test.helper import (
  14.     FakeLogger,
  15.     FakeYDL,
  16.     http_server_port,
  17.     try_rm,
  18. )
  19. from youtube_dl import YoutubeDL
  20. from youtube_dl.compat import (
  21.     compat_http_cookiejar_Cookie,
  22.     compat_http_server,
  23.     compat_kwargs,
  24. )
  25. from youtube_dl.utils import (
  26.     encodeFilename,
  27.     join_nonempty,
  28. )
  29. from youtube_dl.downloader.external import (
  30.     Aria2cFD,
  31.     Aria2pFD,
  32.     AxelFD,
  33.     CurlFD,
  34.     FFmpegFD,
  35.     HttpieFD,
  36.     WgetFD,
  37. )
  38. import threading
  40. TEST_SIZE = 10 * 1024
  42. TEST_COOKIE = {
  43.     'version': 0,
  44.     'name': 'test',
  45.     'value': 'ytdlp',
  46.     'port': None,
  47.     'port_specified': False,
  48.     'domain': '.example.com',
  49.     'domain_specified': True,
  50.     'domain_initial_dot': False,
  51.     'path': '/',
  52.     'path_specified': True,
  53.     'secure': False,
  54.     'expires': None,
  55.     'discard': False,
  56.     'comment': None,
  57.     'comment_url': None,
  58.     'rest': {},
  59. }
  61. TEST_COOKIE_VALUE = join_nonempty('name', 'value', delim='=', from_dict=TEST_COOKIE)
  63. TEST_INFO = {'url': 'http://www.example.com/'}
  66. def cookiejar_Cookie(**cookie_args):
  67.     return compat_http_cookiejar_Cookie(**compat_kwargs(cookie_args))
  70. def ifExternalFDAvailable(externalFD):
  71.     return unittest.skipUnless(externalFD.available(),
  72.                                externalFD.get_basename() + ' not found')
  75. class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
  76.     def log_message(self, format, *args):
  77.         pass
  79.     def send_content_range(self, total=None):
  80.         range_header = self.headers.get('Range')
  81.         start = end = None
  82.         if range_header:
  83.             mobj = re.match(r'bytes=(\d+)-(\d+)', range_header)
  84.             if mobj:
  85.                 start, end = (int(mobj.group(i)) for i in (1, 2))
  86.         valid_range = start is not None and end is not None
  87.         if valid_range:
  88.             content_range = 'bytes %d-%d' % (start, end)
  89.             if total:
  90.                 content_range += '/%d' % total
  91.             self.send_header('Content-Range', content_range)
  92.         return (end - start + 1) if valid_range else total
  94.     def serve(self, range=True, content_length=True):
  95.         self.send_response(200)
  96.         self.send_header('Content-Type', 'video/mp4')
  97.         size = TEST_SIZE
  98.         if range:
  99.             size = self.send_content_range(TEST_SIZE)
  100.         if content_length:
  101.             self.send_header('Content-Length', size)
  102.         self.end_headers()
  103.         self.wfile.write(b'#' * size)
  105.     def do_GET(self):
  106.         if self.path == '/regular':
  107.             self.serve()
  108.         elif self.path == '/no-content-length':
  109.             self.serve(content_length=False)
  110.         elif self.path == '/no-range':
  111.             self.serve(range=False)
  112.         elif self.path == '/no-range-no-content-length':
  113.             self.serve(range=False, content_length=False)
  114.         else:
  115.             assert False, 'unrecognised server path'
  118. @ifExternalFDAvailable(Aria2pFD)
  119. class TestAria2pFD(unittest.TestCase):
  120.     def setUp(self):
  121.         self.httpd = compat_http_server.HTTPServer(
  122.             ('127.0.0.1', 0), HTTPTestRequestHandler)
  123.         self.port = http_server_port(self.httpd)
  124.         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
  125.         self.server_thread.daemon = True
  126.         self.server_thread.start()
  128.     def download(self, params, ep):
  129.         with subprocess.Popen(
  130.             ['aria2c', '--enable-rpc'],
  131.             stdout=subprocess.DEVNULL,
  132.             stderr=subprocess.DEVNULL
  133.         ) as process:
  134.             if not process.poll():
  135.                 filename = 'testfile.mp4'
  136.                 params['logger'] = FakeLogger()
  137.                 params['outtmpl'] = filename
  138.                 ydl = YoutubeDL(params)
  139.                 try_rm(encodeFilename(filename))
  140.                 self.assertEqual(ydl.download(['http://127.0.0.1:%d/%s' % (self.port, ep)]), 0)
  141.                 self.assertEqual(os.path.getsize(encodeFilename(filename)), TEST_SIZE)
  142.                 try_rm(encodeFilename(filename))
  143.             process.kill()
  145.     def download_all(self, params):
  146.         for ep in ('regular', 'no-content-length', 'no-range', 'no-range-no-content-length'):
  147.             self.download(params, ep)
  149.     def test_regular(self):
  150.         self.download_all({'external_downloader': 'aria2p'})
  152.     def test_chunked(self):
  153.         self.download_all({
  154.             'external_downloader': 'aria2p',
  155.             'http_chunk_size': 1000,
  156.         })
  159. @ifExternalFDAvailable(HttpieFD)
  160. class TestHttpieFD(unittest.TestCase):
  161.     def test_make_cmd(self):
  162.         with FakeYDL() as ydl:
  163.             downloader = HttpieFD(ydl, {})
  164.             self.assertEqual(
  165.                 downloader._make_cmd('test', TEST_INFO),
  166.                 ['http', '--download', '--output', 'test', 'http://www.example.com/'])
  168.             # Test cookie header is added
  169.             ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
  170.             self.assertEqual(
  171.                 downloader._make_cmd('test', TEST_INFO),
  172.                 ['http', '--download', '--output', 'test',
  173.                  'http://www.example.com/', 'Cookie:' + TEST_COOKIE_VALUE])
  176. @ifExternalFDAvailable(AxelFD)
  177. class TestAxelFD(unittest.TestCase):
  178.     def test_make_cmd(self):
  179.         with FakeYDL() as ydl:
  180.             downloader = AxelFD(ydl, {})
  181.             self.assertEqual(
  182.                 downloader._make_cmd('test', TEST_INFO),
  183.                 ['axel', '-o', 'test', '--', 'http://www.example.com/'])
  185.             # Test cookie header is added
  186.             ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
  187.             self.assertEqual(
  188.                 downloader._make_cmd('test', TEST_INFO),
  189.                 ['axel', '-o', 'test', '-H', 'Cookie: ' + TEST_COOKIE_VALUE,
  190.                  '--max-redirect=0', '--', 'http://www.example.com/'])
  193. @ifExternalFDAvailable(WgetFD)
  194. class TestWgetFD(unittest.TestCase):
  195.     def test_make_cmd(self):
  196.         with FakeYDL() as ydl:
  197.             downloader = WgetFD(ydl, {})
  198.             self.assertNotIn('--load-cookies', downloader._make_cmd('test', TEST_INFO))
  199.             # Test cookiejar tempfile arg is added
  200.             ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
  201.             self.assertIn('--load-cookies', downloader._make_cmd('test', TEST_INFO))
  204. @ifExternalFDAvailable(CurlFD)
  205. class TestCurlFD(unittest.TestCase):
  206.     def test_make_cmd(self):
  207.         with FakeYDL() as ydl:
  208.             downloader = CurlFD(ydl, {})
  209.             self.assertNotIn('--cookie', downloader._make_cmd('test', TEST_INFO))
  210.             # Test cookie header is added
  211.             ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
  212.             self.assertIn('--cookie', downloader._make_cmd('test', TEST_INFO))
  213.             self.assertIn(TEST_COOKIE_VALUE, downloader._make_cmd('test', TEST_INFO))
  216. @ifExternalFDAvailable(Aria2cFD)
  217. class TestAria2cFD(unittest.TestCase):
  218.     def test_make_cmd(self):
  219.         with FakeYDL() as ydl:
  220.             downloader = Aria2cFD(ydl, {})
  221.             downloader._make_cmd('test', TEST_INFO)
  222.             self.assertFalse(hasattr(downloader, '_cookies_tempfile'))
  224.             # Test cookiejar tempfile arg is added
  225.             ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
  226.             cmd = downloader._make_cmd('test', TEST_INFO)
  227.             self.assertIn('--load-cookies=%s' % downloader._cookies_tempfile, cmd)
  230. @ifExternalFDAvailable(FFmpegFD)
  231. class TestFFmpegFD(unittest.TestCase):
  232.     _args = []
  234.     def _test_cmd(self, args):
  235.         self._args = args
  237.     def test_make_cmd(self):
  238.         with FakeYDL() as ydl:
  239.             downloader = FFmpegFD(ydl, {})
  240.             downloader._debug_cmd = self._test_cmd
  241.             info_dict = TEST_INFO.copy()
  242.             info_dict['ext'] = 'mp4'
  244.             downloader._call_downloader('test', info_dict)
  245.             self.assertEqual(self._args, [
  246.                 'ffmpeg', '-y', '-i', 'http://www.example.com/',
  247.                 '-c', 'copy', '-f', 'mp4', 'file:test'])
  249.             # Test cookies arg is added
  250.             ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
  251.             downloader._call_downloader('test', info_dict)
  252.             self.assertEqual(self._args, [
  253.                 'ffmpeg', '-y', '-cookies', TEST_COOKIE_VALUE + '; path=/; domain=.example.com;\r\n',
  254.                 '-i', 'http://www.example.com/', '-c', 'copy', '-f', 'mp4', 'file:test'])
  257. if __name__ == '__main__':
  258.     unittest.main()