mirror of
https://github.com/ytdl-org/youtube-dl.git
synced 2024-11-18 00:05:46 +00:00
258 lines
8.4 KiB
Python
258 lines
8.4 KiB
Python
#!/usr/bin/env python
|
|
# coding: utf-8
|
|
from __future__ import unicode_literals
|
|
|
|
# Allow direct execution
|
|
import os
|
|
import re
|
|
import sys
|
|
import subprocess
|
|
import unittest
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from test.helper import (
|
|
FakeLogger,
|
|
FakeYDL,
|
|
http_server_port,
|
|
try_rm,
|
|
)
|
|
from youtube_dl import YoutubeDL
|
|
from youtube_dl.compat import (
|
|
compat_http_cookiejar_Cookie,
|
|
compat_http_server,
|
|
compat_kwargs,
|
|
)
|
|
from youtube_dl.utils import (
|
|
encodeFilename,
|
|
join_nonempty,
|
|
)
|
|
from youtube_dl.downloader.external import (
|
|
Aria2cFD,
|
|
Aria2pFD,
|
|
AxelFD,
|
|
CurlFD,
|
|
FFmpegFD,
|
|
HttpieFD,
|
|
WgetFD,
|
|
)
|
|
import threading
|
|
|
|
TEST_SIZE = 10 * 1024
|
|
|
|
TEST_COOKIE = {
|
|
'version': 0,
|
|
'name': 'test',
|
|
'value': 'ytdlp',
|
|
'port': None,
|
|
'port_specified': False,
|
|
'domain': '.example.com',
|
|
'domain_specified': True,
|
|
'domain_initial_dot': False,
|
|
'path': '/',
|
|
'path_specified': True,
|
|
'secure': False,
|
|
'expires': None,
|
|
'discard': False,
|
|
'comment': None,
|
|
'comment_url': None,
|
|
'rest': {},
|
|
}
|
|
|
|
TEST_COOKIE_VALUE = join_nonempty('name', 'value', delim='=', from_dict=TEST_COOKIE)
|
|
|
|
TEST_INFO = {'url': 'http://www.example.com/'}
|
|
|
|
|
|
def cookiejar_Cookie(**cookie_args):
|
|
return compat_http_cookiejar_Cookie(**compat_kwargs(cookie_args))
|
|
|
|
|
|
def ifExternalFDAvailable(externalFD):
|
|
return unittest.skipUnless(externalFD.available(),
|
|
externalFD.get_basename() + ' not found')
|
|
|
|
|
|
class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
def send_content_range(self, total=None):
|
|
range_header = self.headers.get('Range')
|
|
start = end = None
|
|
if range_header:
|
|
mobj = re.match(r'bytes=(\d+)-(\d+)', range_header)
|
|
if mobj:
|
|
start, end = (int(mobj.group(i)) for i in (1, 2))
|
|
valid_range = start is not None and end is not None
|
|
if valid_range:
|
|
content_range = 'bytes %d-%d' % (start, end)
|
|
if total:
|
|
content_range += '/%d' % total
|
|
self.send_header('Content-Range', content_range)
|
|
return (end - start + 1) if valid_range else total
|
|
|
|
def serve(self, range=True, content_length=True):
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'video/mp4')
|
|
size = TEST_SIZE
|
|
if range:
|
|
size = self.send_content_range(TEST_SIZE)
|
|
if content_length:
|
|
self.send_header('Content-Length', size)
|
|
self.end_headers()
|
|
self.wfile.write(b'#' * size)
|
|
|
|
def do_GET(self):
|
|
if self.path == '/regular':
|
|
self.serve()
|
|
elif self.path == '/no-content-length':
|
|
self.serve(content_length=False)
|
|
elif self.path == '/no-range':
|
|
self.serve(range=False)
|
|
elif self.path == '/no-range-no-content-length':
|
|
self.serve(range=False, content_length=False)
|
|
else:
|
|
assert False, 'unrecognised server path'
|
|
|
|
|
|
@ifExternalFDAvailable(Aria2pFD)
|
|
class TestAria2pFD(unittest.TestCase):
|
|
def setUp(self):
|
|
self.httpd = compat_http_server.HTTPServer(
|
|
('127.0.0.1', 0), HTTPTestRequestHandler)
|
|
self.port = http_server_port(self.httpd)
|
|
self.server_thread = threading.Thread(target=self.httpd.serve_forever)
|
|
self.server_thread.daemon = True
|
|
self.server_thread.start()
|
|
|
|
def download(self, params, ep):
|
|
with subprocess.Popen(
|
|
['aria2c', '--enable-rpc'],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
) as process:
|
|
if not process.poll():
|
|
filename = 'testfile.mp4'
|
|
params['logger'] = FakeLogger()
|
|
params['outtmpl'] = filename
|
|
ydl = YoutubeDL(params)
|
|
try_rm(encodeFilename(filename))
|
|
self.assertEqual(ydl.download(['http://127.0.0.1:%d/%s' % (self.port, ep)]), 0)
|
|
self.assertEqual(os.path.getsize(encodeFilename(filename)), TEST_SIZE)
|
|
try_rm(encodeFilename(filename))
|
|
process.kill()
|
|
|
|
def download_all(self, params):
|
|
for ep in ('regular', 'no-content-length', 'no-range', 'no-range-no-content-length'):
|
|
self.download(params, ep)
|
|
|
|
def test_regular(self):
|
|
self.download_all({'external_downloader': 'aria2p'})
|
|
|
|
def test_chunked(self):
|
|
self.download_all({
|
|
'external_downloader': 'aria2p',
|
|
'http_chunk_size': 1000,
|
|
})
|
|
|
|
|
|
@ifExternalFDAvailable(HttpieFD)
|
|
class TestHttpieFD(unittest.TestCase):
|
|
def test_make_cmd(self):
|
|
with FakeYDL() as ydl:
|
|
downloader = HttpieFD(ydl, {})
|
|
self.assertEqual(
|
|
downloader._make_cmd('test', TEST_INFO),
|
|
['http', '--download', '--output', 'test', 'http://www.example.com/'])
|
|
|
|
# Test cookie header is added
|
|
ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
|
|
self.assertEqual(
|
|
downloader._make_cmd('test', TEST_INFO),
|
|
['http', '--download', '--output', 'test',
|
|
'http://www.example.com/', 'Cookie:' + TEST_COOKIE_VALUE])
|
|
|
|
|
|
@ifExternalFDAvailable(AxelFD)
|
|
class TestAxelFD(unittest.TestCase):
|
|
def test_make_cmd(self):
|
|
with FakeYDL() as ydl:
|
|
downloader = AxelFD(ydl, {})
|
|
self.assertEqual(
|
|
downloader._make_cmd('test', TEST_INFO),
|
|
['axel', '-o', 'test', '--', 'http://www.example.com/'])
|
|
|
|
# Test cookie header is added
|
|
ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
|
|
self.assertEqual(
|
|
downloader._make_cmd('test', TEST_INFO),
|
|
['axel', '-o', 'test', '-H', 'Cookie: ' + TEST_COOKIE_VALUE,
|
|
'--max-redirect=0', '--', 'http://www.example.com/'])
|
|
|
|
|
|
@ifExternalFDAvailable(WgetFD)
|
|
class TestWgetFD(unittest.TestCase):
|
|
def test_make_cmd(self):
|
|
with FakeYDL() as ydl:
|
|
downloader = WgetFD(ydl, {})
|
|
self.assertNotIn('--load-cookies', downloader._make_cmd('test', TEST_INFO))
|
|
# Test cookiejar tempfile arg is added
|
|
ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
|
|
self.assertIn('--load-cookies', downloader._make_cmd('test', TEST_INFO))
|
|
|
|
|
|
@ifExternalFDAvailable(CurlFD)
|
|
class TestCurlFD(unittest.TestCase):
|
|
def test_make_cmd(self):
|
|
with FakeYDL() as ydl:
|
|
downloader = CurlFD(ydl, {})
|
|
self.assertNotIn('--cookie', downloader._make_cmd('test', TEST_INFO))
|
|
# Test cookie header is added
|
|
ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
|
|
self.assertIn('--cookie', downloader._make_cmd('test', TEST_INFO))
|
|
self.assertIn(TEST_COOKIE_VALUE, downloader._make_cmd('test', TEST_INFO))
|
|
|
|
|
|
@ifExternalFDAvailable(Aria2cFD)
|
|
class TestAria2cFD(unittest.TestCase):
|
|
def test_make_cmd(self):
|
|
with FakeYDL() as ydl:
|
|
downloader = Aria2cFD(ydl, {})
|
|
downloader._make_cmd('test', TEST_INFO)
|
|
self.assertFalse(hasattr(downloader, '_cookies_tempfile'))
|
|
|
|
# Test cookiejar tempfile arg is added
|
|
ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
|
|
cmd = downloader._make_cmd('test', TEST_INFO)
|
|
self.assertIn('--load-cookies=%s' % downloader._cookies_tempfile, cmd)
|
|
|
|
|
|
@ifExternalFDAvailable(FFmpegFD)
|
|
class TestFFmpegFD(unittest.TestCase):
|
|
_args = []
|
|
|
|
def _test_cmd(self, args):
|
|
self._args = args
|
|
|
|
def test_make_cmd(self):
|
|
with FakeYDL() as ydl:
|
|
downloader = FFmpegFD(ydl, {})
|
|
downloader._debug_cmd = self._test_cmd
|
|
info_dict = TEST_INFO.copy()
|
|
info_dict['ext'] = 'mp4'
|
|
|
|
downloader._call_downloader('test', info_dict)
|
|
self.assertEqual(self._args, [
|
|
'ffmpeg', '-y', '-i', 'http://www.example.com/',
|
|
'-c', 'copy', '-f', 'mp4', 'file:test'])
|
|
|
|
# Test cookies arg is added
|
|
ydl.cookiejar.set_cookie(cookiejar_Cookie(**TEST_COOKIE))
|
|
downloader._call_downloader('test', info_dict)
|
|
self.assertEqual(self._args, [
|
|
'ffmpeg', '-y', '-cookies', TEST_COOKIE_VALUE + '; path=/; domain=.example.com;\r\n',
|
|
'-i', 'http://www.example.com/', '-c', 'copy', '-f', 'mp4', 'file:test'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|