]>
jfr.im git - yt-dlp.git/blob - test/test_downloader_http.py
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__':