3 # Allow direct execution
8 sys
.path
.insert(0, os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))))
16 from test
.helper
import http_server_port
17 from yt_dlp
import YoutubeDL
19 TEST_DIR
= os
.path
.dirname(os
.path
.abspath(__file__
))
22 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
23 def log_message(self
, format
, *args
):
27 if self
.path
== '/video.html':
28 self
.send_response(200)
29 self
.send_header('Content-Type', 'text/html; charset=utf-8')
31 self
.wfile
.write(b
'<html><video src="/vid.mp4" /></html>')
32 elif self
.path
== '/vid.mp4':
33 self
.send_response(200)
34 self
.send_header('Content-Type', 'video/mp4')
36 self
.wfile
.write(b
'\x00\x00\x00\x00\x20\x66\x74[video]')
37 elif self
.path
== '/%E4%B8%AD%E6%96%87.html':
38 self
.send_response(200)
39 self
.send_header('Content-Type', 'text/html; charset=utf-8')
41 self
.wfile
.write(b
'<html><video src="/vid.mp4" /></html>')
50 def warning(self
, msg
):
57 class TestHTTP(unittest
.TestCase
):
59 self
.httpd
= http
.server
.HTTPServer(
60 ('127.0.0.1', 0), HTTPTestRequestHandler
)
61 self
.port
= http_server_port(self
.httpd
)
62 self
.server_thread
= threading
.Thread(target
=self
.httpd
.serve_forever
)
63 self
.server_thread
.daemon
= True
64 self
.server_thread
.start()
67 class TestHTTPS(unittest
.TestCase
):
69 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
70 self
.httpd
= http
.server
.HTTPServer(
71 ('127.0.0.1', 0), HTTPTestRequestHandler
)
72 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
73 sslctx
.load_cert_chain(certfn
, None)
74 self
.httpd
.socket
= sslctx
.wrap_socket(self
.httpd
.socket
, server_side
=True)
75 self
.port
= http_server_port(self
.httpd
)
76 self
.server_thread
= threading
.Thread(target
=self
.httpd
.serve_forever
)
77 self
.server_thread
.daemon
= True
78 self
.server_thread
.start()
80 def test_nocheckcertificate(self
):
81 ydl
= YoutubeDL({'logger': FakeLogger()}
)
84 ydl
.extract_info
, 'https://127.0.0.1:%d/video.html' % self
.port
)
86 ydl
= YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True}
)
87 r
= ydl
.extract_info('https://127.0.0.1:%d/video.html' % self
.port
)
88 self
.assertEqual(r
['url'], 'https://127.0.0.1:%d/vid.mp4' % self
.port
)
91 class TestClientCert(unittest
.TestCase
):
93 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
94 self
.certdir
= os
.path
.join(TEST_DIR
, 'testdata', 'certificate')
95 cacertfn
= os
.path
.join(self
.certdir
, 'ca.crt')
96 self
.httpd
= http
.server
.HTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler
)
97 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
98 sslctx
.verify_mode
= ssl
.CERT_REQUIRED
99 sslctx
.load_verify_locations(cafile
=cacertfn
)
100 sslctx
.load_cert_chain(certfn
, None)
101 self
.httpd
.socket
= sslctx
.wrap_socket(self
.httpd
.socket
, server_side
=True)
102 self
.port
= http_server_port(self
.httpd
)
103 self
.server_thread
= threading
.Thread(target
=self
.httpd
.serve_forever
)
104 self
.server_thread
.daemon
= True
105 self
.server_thread
.start()
107 def _run_test(self
, **params
):
109 'logger': FakeLogger(),
110 # Disable client-side validation of unacceptable self-signed testcert.pem
111 # The test is of a check on the server side, so unaffected
112 'nocheckcertificate': True,
115 r
= ydl
.extract_info('https://127.0.0.1:%d/video.html' % self
.port
)
116 self
.assertEqual(r
['url'], 'https://127.0.0.1:%d/vid.mp4' % self
.port
)
118 def test_certificate_combined_nopass(self
):
119 self
._run
_test
(client_certificate
=os
.path
.join(self
.certdir
, 'clientwithkey.crt'))
121 def test_certificate_nocombined_nopass(self
):
122 self
._run
_test
(client_certificate
=os
.path
.join(self
.certdir
, 'client.crt'),
123 client_certificate_key
=os
.path
.join(self
.certdir
, 'client.key'))
125 def test_certificate_combined_pass(self
):
126 self
._run
_test
(client_certificate
=os
.path
.join(self
.certdir
, 'clientwithencryptedkey.crt'),
127 client_certificate_password
='foobar')
129 def test_certificate_nocombined_pass(self
):
130 self
._run
_test
(client_certificate
=os
.path
.join(self
.certdir
, 'client.crt'),
131 client_certificate_key
=os
.path
.join(self
.certdir
, 'clientencrypted.key'),
132 client_certificate_password
='foobar')
135 def _build_proxy_handler(name
):
136 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
139 def log_message(self
, format
, *args
):
143 self
.send_response(200)
144 self
.send_header('Content-Type', 'text/plain; charset=utf-8')
146 self
.wfile
.write(f
'{self.proxy_name}: {self.path}'.encode())
147 return HTTPTestRequestHandler
150 class TestProxy(unittest
.TestCase
):
152 self
.proxy
= http
.server
.HTTPServer(
153 ('127.0.0.1', 0), _build_proxy_handler('normal'))
154 self
.port
= http_server_port(self
.proxy
)
155 self
.proxy_thread
= threading
.Thread(target
=self
.proxy
.serve_forever
)
156 self
.proxy_thread
.daemon
= True
157 self
.proxy_thread
.start()
159 self
.geo_proxy
= http
.server
.HTTPServer(
160 ('127.0.0.1', 0), _build_proxy_handler('geo'))
161 self
.geo_port
= http_server_port(self
.geo_proxy
)
162 self
.geo_proxy_thread
= threading
.Thread(target
=self
.geo_proxy
.serve_forever
)
163 self
.geo_proxy_thread
.daemon
= True
164 self
.geo_proxy_thread
.start()
166 def test_proxy(self
):
167 geo_proxy
= f
'127.0.0.1:{self.geo_port}'
169 'proxy': f
'127.0.0.1:{self.port}',
170 'geo_verification_proxy': geo_proxy
,
172 url
= 'http://foo.com/bar'
173 response
= ydl
.urlopen(url
).read().decode()
174 self
.assertEqual(response
, f
'normal: {url}')
176 req
= urllib
.request
.Request(url
)
177 req
.add_header('Ytdl-request-proxy', geo_proxy
)
178 response
= ydl
.urlopen(req
).read().decode()
179 self
.assertEqual(response
, f
'geo: {url}')
181 def test_proxy_with_idn(self
):
183 'proxy': f
'127.0.0.1:{self.port}',
185 url
= 'http://中文.tw/'
186 response
= ydl
.urlopen(url
).read().decode()
187 # b'xn--fiq228c' is '中文'.encode('idna')
188 self
.assertEqual(response
, 'normal: http://xn--fiq228c.tw/')
191 if __name__
== '__main__':