3 from __future__
import unicode_literals
5 # Allow direct execution
9 sys
.path
.insert(0, os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))))
11 from youtube_dl
import YoutubeDL
12 from youtube_dl
.compat
import compat_http_server
, compat_urllib_request
16 TEST_DIR
= os
.path
.dirname(os
.path
.abspath(__file__
))
19 class HTTPTestRequestHandler(compat_http_server
.BaseHTTPRequestHandler
):
20 def log_message(self
, format
, *args
):
24 if self
.path
== '/video.html':
25 self
.send_response(200)
26 self
.send_header('Content-Type', 'text/html; charset=utf-8')
28 self
.wfile
.write(b
'<html><video src="/vid.mp4" /></html>')
29 elif self
.path
== '/vid.mp4':
30 self
.send_response(200)
31 self
.send_header('Content-Type', 'video/mp4')
33 self
.wfile
.write(b
'\x00\x00\x00\x00\x20\x66\x74[video]')
38 class FakeLogger(object):
42 def warning(self
, msg
):
49 class TestHTTP(unittest
.TestCase
):
51 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
52 self
.httpd
= compat_http_server
.HTTPServer(
53 ('localhost', 0), HTTPTestRequestHandler
)
54 self
.httpd
.socket
= ssl
.wrap_socket(
55 self
.httpd
.socket
, certfile
=certfn
, server_side
=True)
57 # In Jython SSLSocket is not a subclass of socket.socket
58 sock
= self
.httpd
.socket
.sock
60 sock
= self
.httpd
.socket
61 self
.port
= sock
.getsockname()[1]
62 self
.server_thread
= threading
.Thread(target
=self
.httpd
.serve_forever
)
63 self
.server_thread
.daemon
= True
64 self
.server_thread
.start()
66 def test_nocheckcertificate(self
):
67 if sys
.version_info
>= (2, 7, 9): # No certificate checking anyways
68 ydl
= YoutubeDL({'logger': FakeLogger()}
)
71 ydl
.extract_info
, 'https://localhost:%d/video.html' % self
.port
)
73 ydl
= YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True}
)
74 r
= ydl
.extract_info('https://localhost:%d/video.html' % self
.port
)
75 self
.assertEqual(r
['url'], 'https://localhost:%d/vid.mp4' % self
.port
)
78 def _build_proxy_handler(name
):
79 class HTTPTestRequestHandler(compat_http_server
.BaseHTTPRequestHandler
):
82 def log_message(self
, format
, *args
):
86 self
.send_response(200)
87 self
.send_header('Content-Type', 'text/plain; charset=utf-8')
89 self
.wfile
.write('{self.proxy_name}: {self.path}'.format(self
=self
).encode('utf-8'))
90 return HTTPTestRequestHandler
93 class TestProxy(unittest
.TestCase
):
95 self
.proxy
= compat_http_server
.HTTPServer(
96 ('localhost', 0), _build_proxy_handler('normal'))
97 self
.port
= self
.proxy
.socket
.getsockname()[1]
98 self
.proxy_thread
= threading
.Thread(target
=self
.proxy
.serve_forever
)
99 self
.proxy_thread
.daemon
= True
100 self
.proxy_thread
.start()
102 self
.cn_proxy
= compat_http_server
.HTTPServer(
103 ('localhost', 0), _build_proxy_handler('cn'))
104 self
.cn_port
= self
.cn_proxy
.socket
.getsockname()[1]
105 self
.cn_proxy_thread
= threading
.Thread(target
=self
.cn_proxy
.serve_forever
)
106 self
.cn_proxy_thread
.daemon
= True
107 self
.cn_proxy_thread
.start()
109 def test_proxy(self
):
110 cn_proxy
= 'localhost:{0}'.format(self
.cn_port
)
112 'proxy': 'localhost:{0}'.format(self
.port
),
113 'cn_verification_proxy': cn_proxy
,
115 url
= 'http://foo.com/bar'
116 response
= ydl
.urlopen(url
).read().decode('utf-8')
117 self
.assertEqual(response
, 'normal: {0}'.format(url
))
119 req
= compat_urllib_request
.Request(url
)
120 req
.add_header('Ytdl-request-proxy', cn_proxy
)
121 response
= ydl
.urlopen(req
).read().decode('utf-8')
122 self
.assertEqual(response
, 'cn: {0}'.format(url
))
124 def test_proxy_with_idn(self
):
126 'proxy': 'localhost:{0}'.format(self
.port
),
128 url
= 'http://中文.tw/'
129 response
= ydl
.urlopen(url
).read().decode('utf-8')
130 # b'xn--fiq228c' is '中文'.encode('idna')
131 self
.assertEqual(response
, 'normal: http://xn--fiq228c.tw/')
133 if __name__
== '__main__':