3 from __future__ import unicode_literals
5 # Allow direct execution
10 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
12 from youtube_dl import YoutubeDL
13 from youtube_dl.compat import compat_http_server
14 from youtube_dl.downloader.http import HttpFD
15 from youtube_dl.utils import encodeFilename
19 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
22 def http_server_port(httpd):
23 if os.name == 'java' and isinstance(httpd.socket, ssl.SSLSocket):
24 # In Jython SSLSocket is not a subclass of socket.socket
25 sock = httpd.socket.sock
28 return sock.getsockname()[1]
34 class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
35 def log_message(self, format, *args):
38 def send_content_range(self, total=None):
39 range_header = self.headers.get('Range')
42 mobj = re.search(r'^bytes=(\d+)-(\d+)', range_header)
44 start = int(mobj.group(1))
45 end = int(mobj.group(2))
46 valid_range = start is not None and end is not None
48 content_range = 'bytes %d-%d' % (start, end)
50 content_range += '/%d' % total
51 self.send_header('Content-Range', content_range)
52 return (end - start + 1) if valid_range else total
54 def serve(self, range=True, content_length=True):
55 self.send_response(200)
56 self.send_header('Content-Type', 'video/mp4')
59 size = self.send_content_range(TEST_SIZE)
61 self.send_header('Content-Length', size)
63 self.wfile.write(b'#' * size)
66 if self.path == '/regular':
68 elif self.path == '/no-content-length':
69 self.serve(content_length=False)
70 elif self.path == '/no-range':
71 self.serve(range=False)
72 elif self.path == '/no-range-no-content-length':
73 self.serve(range=False, content_length=False)
78 class FakeLogger(object):
82 def warning(self, msg):
89 class TestHttpFD(unittest.TestCase):
91 self.httpd = compat_http_server.HTTPServer(
92 ('127.0.0.1', 0), HTTPTestRequestHandler)
93 self.port = http_server_port(self.httpd)
94 self.server_thread = threading.Thread(target=self.httpd.serve_forever)
95 self.server_thread.daemon = True
96 self.server_thread.start()
98 def download(self, params, ep):
99 params['logger'] = FakeLogger()
100 ydl = YoutubeDL(params)
101 downloader = HttpFD(ydl, params)
102 filename = 'testfile.mp4'
103 self.assertTrue(downloader.real_download(filename, {
104 'url': 'http://127.0.0.1:%d/%s' % (self.port, ep),
106 self.assertEqual(os.path.getsize(encodeFilename(filename)), TEST_SIZE)
107 os.remove(encodeFilename(filename))
109 def download_all(self, params):
110 for ep in ('regular', 'no-content-length', 'no-range', 'no-range-no-content-length'):
111 self.download(params, ep)
113 def test_regular(self):
114 self.download_all({})
116 def test_chunked(self):
118 'http_chunk_size': 1000,
122 if __name__ == '__main__':