3 # Allow direct execution
9 sys
.path
.insert(0, os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))))
26 from email
.message
import Message
27 from http
.cookiejar
import CookieJar
29 from test
.helper
import FakeYDL
, http_server_port
, verify_address_availability
30 from yt_dlp
.cookies
import YoutubeDLCookieJar
31 from yt_dlp
.dependencies
import brotli
, requests
, urllib3
32 from yt_dlp
.networking
import (
40 from yt_dlp
.networking
._urllib
import UrllibRH
41 from yt_dlp
.networking
.exceptions
import (
42 CertificateVerifyError
,
52 from yt_dlp
.utils
._utils
import _YDLLogger
as FakeLogger
53 from yt_dlp
.utils
.networking
import HTTPHeaderDict
55 from test
.conftest
import validate_and_send
57 TEST_DIR
= os
.path
.dirname(os
.path
.abspath(__file__
))
60 def _build_proxy_handler(name
):
61 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
64 def log_message(self
, format
, *args
):
68 self
.send_response(200)
69 self
.send_header('Content-Type', 'text/plain; charset=utf-8')
71 self
.wfile
.write('{self.proxy_name}: {self.path}'.format(self
=self
).encode())
72 return HTTPTestRequestHandler
75 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
76 protocol_version
= 'HTTP/1.1'
78 def log_message(self
, format
, *args
):
82 payload
= str(self
.headers
).encode()
83 self
.send_response(200)
84 self
.send_header('Content-Type', 'application/json')
85 self
.send_header('Content-Length', str(len(payload
)))
87 self
.wfile
.write(payload
)
90 self
.send_response(int(self
.path
[len('/redirect_'):]))
91 self
.send_header('Location', '/method')
92 self
.send_header('Content-Length', '0')
95 def _method(self
, method
, payload
=None):
96 self
.send_response(200)
97 self
.send_header('Content-Length', str(len(payload
or '')))
98 self
.send_header('Method', method
)
101 self
.wfile
.write(payload
)
103 def _status(self
, status
):
104 payload
= f
'<html>{status} NOT FOUND</html>'.encode()
105 self
.send_response(int(status
))
106 self
.send_header('Content-Type', 'text/html; charset=utf-8')
107 self
.send_header('Content-Length', str(len(payload
)))
109 self
.wfile
.write(payload
)
111 def _read_data(self
):
112 if 'Content-Length' in self
.headers
:
113 return self
.rfile
.read(int(self
.headers
['Content-Length']))
116 data
= self
._read
_data
() + str(self
.headers
).encode()
117 if self
.path
.startswith('/redirect_'):
119 elif self
.path
.startswith('/method'):
120 self
._method
('POST', data
)
121 elif self
.path
.startswith('/headers'):
127 if self
.path
.startswith('/redirect_'):
129 elif self
.path
.startswith('/method'):
135 data
= self
._read
_data
() + str(self
.headers
).encode()
136 if self
.path
.startswith('/redirect_'):
138 elif self
.path
.startswith('/method'):
139 self
._method
('PUT', data
)
144 if self
.path
== '/video.html':
145 payload
= b
'<html><video src="/vid.mp4" /></html>'
146 self
.send_response(200)
147 self
.send_header('Content-Type', 'text/html; charset=utf-8')
148 self
.send_header('Content-Length', str(len(payload
)))
150 self
.wfile
.write(payload
)
151 elif self
.path
== '/vid.mp4':
152 payload
= b
'\x00\x00\x00\x00\x20\x66\x74[video]'
153 self
.send_response(200)
154 self
.send_header('Content-Type', 'video/mp4')
155 self
.send_header('Content-Length', str(len(payload
)))
157 self
.wfile
.write(payload
)
158 elif self
.path
== '/%E4%B8%AD%E6%96%87.html':
159 payload
= b
'<html><video src="/vid.mp4" /></html>'
160 self
.send_response(200)
161 self
.send_header('Content-Type', 'text/html; charset=utf-8')
162 self
.send_header('Content-Length', str(len(payload
)))
164 self
.wfile
.write(payload
)
165 elif self
.path
== '/%c7%9f':
166 payload
= b
'<html><video src="/vid.mp4" /></html>'
167 self
.send_response(200)
168 self
.send_header('Content-Type', 'text/html; charset=utf-8')
169 self
.send_header('Content-Length', str(len(payload
)))
171 self
.wfile
.write(payload
)
172 elif self
.path
.startswith('/redirect_loop'):
173 self
.send_response(301)
174 self
.send_header('Location', self
.path
)
175 self
.send_header('Content-Length', '0')
177 elif self
.path
== '/redirect_dotsegments':
178 self
.send_response(301)
179 # redirect to /headers but with dot segments before
180 self
.send_header('Location', '/a/b/./../../headers')
181 self
.send_header('Content-Length', '0')
183 elif self
.path
== '/redirect_dotsegments_absolute':
184 self
.send_response(301)
185 # redirect to /headers but with dot segments before - absolute url
186 self
.send_header('Location', f
'http://127.0.0.1:{http_server_port(self.server)}/a/b/./../../headers')
187 self
.send_header('Content-Length', '0')
189 elif self
.path
.startswith('/redirect_'):
191 elif self
.path
.startswith('/method'):
192 self
._method
('GET', str(self
.headers
).encode())
193 elif self
.path
.startswith('/headers'):
195 elif self
.path
.startswith('/308-to-headers'):
196 self
.send_response(308)
197 self
.send_header('Location', '/headers')
198 self
.send_header('Content-Length', '0')
200 elif self
.path
== '/trailing_garbage':
201 payload
= b
'<html><video src="/vid.mp4" /></html>'
202 self
.send_response(200)
203 self
.send_header('Content-Type', 'text/html; charset=utf-8')
204 self
.send_header('Content-Encoding', 'gzip')
206 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
208 compressed
= buf
.getvalue() + b
'trailing garbage'
209 self
.send_header('Content-Length', str(len(compressed
)))
211 self
.wfile
.write(compressed
)
212 elif self
.path
== '/302-non-ascii-redirect':
213 new_url
= f
'http://127.0.0.1:{http_server_port(self.server)}/中文.html'
214 self
.send_response(301)
215 self
.send_header('Location', new_url
)
216 self
.send_header('Content-Length', '0')
218 elif self
.path
== '/content-encoding':
219 encodings
= self
.headers
.get('ytdl-encoding', '')
220 payload
= b
'<html><video src="/vid.mp4" /></html>'
221 for encoding
in filter(None, (e
.strip() for e
in encodings
.split(','))):
222 if encoding
== 'br' and brotli
:
223 payload
= brotli
.compress(payload
)
224 elif encoding
== 'gzip':
226 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
228 payload
= buf
.getvalue()
229 elif encoding
== 'deflate':
230 payload
= zlib
.compress(payload
)
231 elif encoding
== 'unsupported':
237 self
.send_response(200)
238 self
.send_header('Content-Encoding', encodings
)
239 self
.send_header('Content-Length', str(len(payload
)))
241 self
.wfile
.write(payload
)
242 elif self
.path
.startswith('/gen_'):
243 payload
= b
'<html></html>'
244 self
.send_response(int(self
.path
[len('/gen_'):]))
245 self
.send_header('Content-Type', 'text/html; charset=utf-8')
246 self
.send_header('Content-Length', str(len(payload
)))
248 self
.wfile
.write(payload
)
249 elif self
.path
.startswith('/incompleteread'):
250 payload
= b
'<html></html>'
251 self
.send_response(200)
252 self
.send_header('Content-Type', 'text/html; charset=utf-8')
253 self
.send_header('Content-Length', '234234')
255 self
.wfile
.write(payload
)
257 elif self
.path
.startswith('/timeout_'):
258 time
.sleep(int(self
.path
[len('/timeout_'):]))
260 elif self
.path
== '/source_address':
261 payload
= str(self
.client_address
[0]).encode()
262 self
.send_response(200)
263 self
.send_header('Content-Type', 'text/html; charset=utf-8')
264 self
.send_header('Content-Length', str(len(payload
)))
266 self
.wfile
.write(payload
)
271 def send_header(self
, keyword
, value
):
273 Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
274 This is against what is defined in RFC 3986, however we need to test we support this
275 since some sites incorrectly do this.
277 if keyword
.lower() == 'connection':
278 return super().send_header(keyword
, value
)
280 if not hasattr(self
, '_headers_buffer'):
281 self
._headers
_buffer
= []
283 self
._headers
_buffer
.append(f
'{keyword}: {value}\r\n'.encode())
286 class TestRequestHandlerBase
:
288 def setup_class(cls
):
289 cls
.http_httpd
= http
.server
.ThreadingHTTPServer(
290 ('127.0.0.1', 0), HTTPTestRequestHandler
)
291 cls
.http_port
= http_server_port(cls
.http_httpd
)
292 cls
.http_server_thread
= threading
.Thread(target
=cls
.http_httpd
.serve_forever
)
293 # FIXME: we should probably stop the http server thread after each test
294 # See: https://github.com/yt-dlp/yt-dlp/pull/7094#discussion_r1199746041
295 cls
.http_server_thread
.daemon
= True
296 cls
.http_server_thread
.start()
299 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
300 cls
.https_httpd
= http
.server
.ThreadingHTTPServer(
301 ('127.0.0.1', 0), HTTPTestRequestHandler
)
302 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
303 sslctx
.load_cert_chain(certfn
, None)
304 cls
.https_httpd
.socket
= sslctx
.wrap_socket(cls
.https_httpd
.socket
, server_side
=True)
305 cls
.https_port
= http_server_port(cls
.https_httpd
)
306 cls
.https_server_thread
= threading
.Thread(target
=cls
.https_httpd
.serve_forever
)
307 cls
.https_server_thread
.daemon
= True
308 cls
.https_server_thread
.start()
311 class TestHTTPRequestHandler(TestRequestHandlerBase
):
312 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
313 def test_verify_cert(self
, handler
):
314 with handler() as rh
:
315 with pytest
.raises(CertificateVerifyError
):
316 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
318 with handler(verify
=False) as rh
:
319 r
= validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
320 assert r
.status
== 200
323 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
324 def test_ssl_error(self
, handler
):
325 # HTTPS server with too old TLS version
326 # XXX: is there a better way to test this than to create a new server?
327 https_httpd
= http
.server
.ThreadingHTTPServer(
328 ('127.0.0.1', 0), HTTPTestRequestHandler
)
329 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
330 https_httpd
.socket
= sslctx
.wrap_socket(https_httpd
.socket
, server_side
=True)
331 https_port
= http_server_port(https_httpd
)
332 https_server_thread
= threading
.Thread(target
=https_httpd
.serve_forever
)
333 https_server_thread
.daemon
= True
334 https_server_thread
.start()
336 with handler(verify
=False) as rh
:
337 with pytest
.raises(SSLError
, match
=r
'ssl(?:v3|/tls) alert handshake failure') as exc_info
:
338 validate_and_send(rh
, Request(f
'https://127.0.0.1:{https_port}/headers'))
339 assert not issubclass(exc_info
.type, CertificateVerifyError
)
341 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
342 def test_percent_encode(self
, handler
):
343 with handler() as rh
:
344 # Unicode characters should be encoded with uppercase percent-encoding
345 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/中文.html'))
346 assert res
.status
== 200
348 # don't normalize existing percent encodings
349 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/%c7%9f'))
350 assert res
.status
== 200
353 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
354 @pytest.mark.parametrize('path', [
355 '/a/b/./../../headers',
356 '/redirect_dotsegments',
357 # https://github.com/yt-dlp/yt-dlp/issues/9020
358 '/redirect_dotsegments_absolute',
360 def test_remove_dot_segments(self
, handler
, path
):
361 with handler(verbose
=True) as rh
:
362 # This isn't a comprehensive test,
363 # but it should be enough to check whether the handler is removing dot segments in required scenarios
364 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}{path}'))
365 assert res
.status
== 200
366 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
369 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
370 def test_unicode_path_redirection(self
, handler
):
371 with handler() as rh
:
372 r
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
373 assert r
.url
== f
'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
376 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
377 def test_raise_http_error(self
, handler
):
378 with handler() as rh
:
379 for bad_status
in (400, 500, 599, 302):
380 with pytest
.raises(HTTPError
):
381 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_%d' % (self
.http_port
, bad_status
)))
383 # Should not raise an error
384 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_200' % self
.http_port
)).close()
386 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
387 def test_response_url(self
, handler
):
388 with handler() as rh
:
389 # Response url should be that of the last url in redirect chain
390 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_301'))
391 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/method'
393 res2
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_200'))
394 assert res2
.url
== f
'http://127.0.0.1:{self.http_port}/gen_200'
397 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
398 def test_redirect(self
, handler
):
399 with handler() as rh
:
400 def do_req(redirect_status
, method
, assert_no_content
=False):
401 data
= b
'testdata' if method
in ('POST', 'PUT') else None
402 res
= validate_and_send(
403 rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method
=method
, data
=data
))
408 data_sent
+= res
.read(len(data
))
409 if data_sent
!= data
:
413 headers
+= res
.read()
415 if assert_no_content
or data
is None:
416 assert b
'Content-Type' not in headers
417 assert b
'Content-Length' not in headers
419 assert b
'Content-Type' in headers
420 assert b
'Content-Length' in headers
422 return data_sent
.decode(), res
.headers
.get('method', '')
424 # A 303 must either use GET or HEAD for subsequent request
425 assert do_req(303, 'POST', True) == ('', 'GET')
426 assert do_req(303, 'HEAD') == ('', 'HEAD')
428 assert do_req(303, 'PUT', True) == ('', 'GET')
430 # 301 and 302 turn POST only into a GET
431 assert do_req(301, 'POST', True) == ('', 'GET')
432 assert do_req(301, 'HEAD') == ('', 'HEAD')
433 assert do_req(302, 'POST', True) == ('', 'GET')
434 assert do_req(302, 'HEAD') == ('', 'HEAD')
436 assert do_req(301, 'PUT') == ('testdata', 'PUT')
437 assert do_req(302, 'PUT') == ('testdata', 'PUT')
439 # 307 and 308 should not change method
440 for m
in ('POST', 'PUT'):
441 assert do_req(307, m
) == ('testdata', m
)
442 assert do_req(308, m
) == ('testdata', m
)
444 assert do_req(307, 'HEAD') == ('', 'HEAD')
445 assert do_req(308, 'HEAD') == ('', 'HEAD')
447 # These should not redirect and instead raise an HTTPError
448 for code
in (300, 304, 305, 306):
449 with pytest
.raises(HTTPError
):
452 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
453 def test_request_cookie_header(self
, handler
):
454 # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
455 with handler() as rh
:
456 # Specified Cookie header should be used
457 res
= validate_and_send(
459 f
'http://127.0.0.1:{self.http_port}/headers',
460 headers
={'Cookie': 'test=test'}
)).read().decode()
461 assert 'Cookie: test=test' in res
463 # Specified Cookie header should be removed on any redirect
464 res
= validate_and_send(
466 f
'http://127.0.0.1:{self.http_port}/308-to-headers',
467 headers
={'Cookie': 'test=test'}
)).read().decode()
468 assert 'Cookie: test=test' not in res
470 # Specified Cookie header should override global cookiejar for that request
471 cookiejar
= YoutubeDLCookieJar()
472 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
473 version
=0, name
='test', value
='ytdlp', port
=None, port_specified
=False,
474 domain
='127.0.0.1', domain_specified
=True, domain_initial_dot
=False, path
='/',
475 path_specified
=True, secure
=False, expires
=None, discard
=False, comment
=None,
476 comment_url
=None, rest
={}))
478 with handler(cookiejar
=cookiejar
) as rh
:
479 data
= validate_and_send(
480 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', headers
={'cookie': 'test=test'}
)).read()
481 assert b
'Cookie: test=ytdlp' not in data
482 assert b
'Cookie: test=test' in data
484 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
485 def test_redirect_loop(self
, handler
):
486 with handler() as rh
:
487 with pytest
.raises(HTTPError
, match
='redirect loop'):
488 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_loop'))
490 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
491 def test_incompleteread(self
, handler
):
492 with handler(timeout
=2) as rh
:
493 with pytest
.raises(IncompleteRead
):
494 validate_and_send(rh
, Request('http://127.0.0.1:%d/incompleteread' % self
.http_port
)).read()
496 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
497 def test_cookies(self
, handler
):
498 cookiejar
= YoutubeDLCookieJar()
499 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
500 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
501 False, '/headers', True, False, None, False, None, None, {}))
503 with handler(cookiejar
=cookiejar
) as rh
:
504 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
505 assert b
'Cookie: test=ytdlp' in data
508 with handler() as rh
:
509 data
= validate_and_send(
510 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', extensions
={'cookiejar': cookiejar}
)).read()
511 assert b
'Cookie: test=ytdlp' in data
513 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
514 def test_headers(self
, handler
):
516 with handler(headers
=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'}
)) as rh
:
518 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
519 assert b
'Test1: test' in data
521 # Per request headers, merged with global
522 data
= validate_and_send(rh
, Request(
523 f
'http://127.0.0.1:{self.http_port}/headers', headers
={'test2': 'changed', 'test3': 'test3'}
)).read()
524 assert b
'Test1: test' in data
525 assert b
'Test2: changed' in data
526 assert b
'Test2: test2' not in data
527 assert b
'Test3: test3' in data
529 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
530 def test_timeout(self
, handler
):
531 with handler() as rh
:
532 # Default timeout is 20 seconds, so this should go through
534 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_3'))
536 with handler(timeout
=0.5) as rh
:
537 with pytest
.raises(TransportError
):
539 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1'))
541 # Per request timeout, should override handler timeout
543 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1', extensions
={'timeout': 4}
))
545 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
546 def test_source_address(self
, handler
):
547 source_address
= f
'127.0.0.{random.randint(5, 255)}'
548 # on some systems these loopback addresses we need for testing may not be available
549 # see: https://github.com/yt-dlp/yt-dlp/issues/8890
550 verify_address_availability(source_address
)
551 with handler(source_address
=source_address
) as rh
:
552 data
= validate_and_send(
553 rh
, Request(f
'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
554 assert source_address
== data
556 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
557 def test_gzip_trailing_garbage(self
, handler
):
558 with handler() as rh
:
559 data
= validate_and_send(rh
, Request(f
'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
560 assert data
== '<html><video src="/vid.mp4" /></html>'
562 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
563 @pytest.mark.skipif(not brotli
, reason
='brotli support is not installed')
564 def test_brotli(self
, handler
):
565 with handler() as rh
:
566 res
= validate_and_send(
568 f
'http://127.0.0.1:{self.http_port}/content-encoding',
569 headers
={'ytdl-encoding': 'br'}
))
570 assert res
.headers
.get('Content-Encoding') == 'br'
571 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
573 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
574 def test_deflate(self
, handler
):
575 with handler() as rh
:
576 res
= validate_and_send(
578 f
'http://127.0.0.1:{self.http_port}/content-encoding',
579 headers
={'ytdl-encoding': 'deflate'}
))
580 assert res
.headers
.get('Content-Encoding') == 'deflate'
581 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
583 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
584 def test_gzip(self
, handler
):
585 with handler() as rh
:
586 res
= validate_and_send(
588 f
'http://127.0.0.1:{self.http_port}/content-encoding',
589 headers
={'ytdl-encoding': 'gzip'}
))
590 assert res
.headers
.get('Content-Encoding') == 'gzip'
591 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
593 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
594 def test_multiple_encodings(self
, handler
):
595 with handler() as rh
:
596 for pair
in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
597 res
= validate_and_send(
599 f
'http://127.0.0.1:{self.http_port}/content-encoding',
600 headers
={'ytdl-encoding': pair}
))
601 assert res
.headers
.get('Content-Encoding') == pair
602 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
604 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
605 def test_unsupported_encoding(self
, handler
):
606 with handler() as rh
:
607 res
= validate_and_send(
609 f
'http://127.0.0.1:{self.http_port}/content-encoding',
610 headers
={'ytdl-encoding': 'unsupported'}
))
611 assert res
.headers
.get('Content-Encoding') == 'unsupported'
612 assert res
.read() == b
'raw'
614 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
615 def test_read(self
, handler
):
616 with handler() as rh
:
617 res
= validate_and_send(
618 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers'))
619 assert res
.readable()
620 assert res
.read(1) == b
'H'
621 assert res
.read(3) == b
'ost'
624 class TestHTTPProxy(TestRequestHandlerBase
):
626 def setup_class(cls
):
627 super().setup_class()
629 cls
.proxy
= http
.server
.ThreadingHTTPServer(
630 ('127.0.0.1', 0), _build_proxy_handler('normal'))
631 cls
.proxy_port
= http_server_port(cls
.proxy
)
632 cls
.proxy_thread
= threading
.Thread(target
=cls
.proxy
.serve_forever
)
633 cls
.proxy_thread
.daemon
= True
634 cls
.proxy_thread
.start()
637 cls
.geo_proxy
= http
.server
.ThreadingHTTPServer(
638 ('127.0.0.1', 0), _build_proxy_handler('geo'))
639 cls
.geo_port
= http_server_port(cls
.geo_proxy
)
640 cls
.geo_proxy_thread
= threading
.Thread(target
=cls
.geo_proxy
.serve_forever
)
641 cls
.geo_proxy_thread
.daemon
= True
642 cls
.geo_proxy_thread
.start()
644 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
645 def test_http_proxy(self
, handler
):
646 http_proxy
= f
'http://127.0.0.1:{self.proxy_port}'
647 geo_proxy
= f
'http://127.0.0.1:{self.geo_port}'
649 # Test global http proxy
650 # Test per request http proxy
651 # Test per request http proxy disables proxy
652 url
= 'http://foo.com/bar'
655 with handler(proxies
={'http': http_proxy}
) as rh
:
656 res
= validate_and_send(rh
, Request(url
)).read().decode()
657 assert res
== f
'normal: {url}'
659 # Per request proxy overrides global
660 res
= validate_and_send(rh
, Request(url
, proxies
={'http': geo_proxy}
)).read().decode()
661 assert res
== f
'geo: {url}'
663 # and setting to None disables all proxies for that request
664 real_url
= f
'http://127.0.0.1:{self.http_port}/headers'
665 res
= validate_and_send(
666 rh
, Request(real_url
, proxies
={'http': None}
)).read().decode()
667 assert res
!= f
'normal: {real_url}'
668 assert 'Accept' in res
670 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
671 def test_noproxy(self
, handler
):
672 with handler(proxies
={'proxy': f'http://127.0.0.1:{self.proxy_port}
'}) as rh:
674 for no_proxy in (f'127.0.0.1:{self.http_port}
', '127.0.0.1', 'localhost
'):
675 nop_response = validate_and_send(
676 rh, Request(f'http
://127.0.0.1:{self.http_port}
/headers
', proxies={'no': no_proxy})).read().decode(
678 assert 'Accept
' in nop_response
680 @pytest.mark.parametrize('handler
', ['Urllib
', 'Requests
'], indirect=True)
681 def test_allproxy(self, handler):
682 url = 'http
://foo
.com
/bar
'
683 with handler() as rh:
684 response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
686 assert response
== f
'normal: {url}'
688 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
689 def test_http_proxy_with_idn(self
, handler
):
690 with handler(proxies
={
691 'http': f
'http://127.0.0.1:{self.proxy_port}',
693 url
= 'http://中文.tw/'
694 response
= rh
.send(Request(url
)).read().decode()
695 # b'xn--fiq228c' is '中文'.encode('idna')
696 assert response
== 'normal: http://xn--fiq228c.tw/'
699 class TestClientCertificate
:
702 def setup_class(cls
):
703 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
704 cls
.certdir
= os
.path
.join(TEST_DIR
, 'testdata', 'certificate')
705 cacertfn
= os
.path
.join(cls
.certdir
, 'ca.crt')
706 cls
.httpd
= http
.server
.ThreadingHTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler
)
707 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
708 sslctx
.verify_mode
= ssl
.CERT_REQUIRED
709 sslctx
.load_verify_locations(cafile
=cacertfn
)
710 sslctx
.load_cert_chain(certfn
, None)
711 cls
.httpd
.socket
= sslctx
.wrap_socket(cls
.httpd
.socket
, server_side
=True)
712 cls
.port
= http_server_port(cls
.httpd
)
713 cls
.server_thread
= threading
.Thread(target
=cls
.httpd
.serve_forever
)
714 cls
.server_thread
.daemon
= True
715 cls
.server_thread
.start()
717 def _run_test(self
, handler
, **handler_kwargs
):
719 # Disable client-side validation of unacceptable self-signed testcert.pem
720 # The test is of a check on the server side, so unaffected
724 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.port}/video.html')).read().decode()
726 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
727 def test_certificate_combined_nopass(self
, handler
):
728 self
._run
_test
(handler
, client_cert
={
729 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithkey.crt'),
732 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
733 def test_certificate_nocombined_nopass(self
, handler
):
734 self
._run
_test
(handler
, client_cert
={
735 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
736 'client_certificate_key': os
.path
.join(self
.certdir
, 'client.key'),
739 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
740 def test_certificate_combined_pass(self
, handler
):
741 self
._run
_test
(handler
, client_cert
={
742 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithencryptedkey.crt'),
743 'client_certificate_password': 'foobar',
746 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
747 def test_certificate_nocombined_pass(self
, handler
):
748 self
._run
_test
(handler
, client_cert
={
749 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
750 'client_certificate_key': os
.path
.join(self
.certdir
, 'clientencrypted.key'),
751 'client_certificate_password': 'foobar',
755 class TestUrllibRequestHandler(TestRequestHandlerBase
):
756 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
757 def test_file_urls(self
, handler
):
758 # See https://github.com/ytdl-org/youtube-dl/issues/8227
759 tf
= tempfile
.NamedTemporaryFile(delete
=False)
762 req
= Request(pathlib
.Path(tf
.name
).as_uri())
763 with handler() as rh
:
764 with pytest
.raises(UnsupportedRequest
):
767 # Test that urllib never loaded FileHandler
768 with pytest
.raises(TransportError
):
771 with handler(enable_file_urls
=True) as rh
:
772 res
= validate_and_send(rh
, req
)
773 assert res
.read() == b
'foobar'
778 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
779 def test_http_error_returns_content(self
, handler
):
780 # urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
782 with handler() as rh
:
785 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_404'))
786 except HTTPError
as e
:
789 assert get_response().read() == b
'<html></html>'
791 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
792 def test_verify_cert_error_text(self
, handler
):
793 # Check the output of the error message
794 with handler() as rh
:
796 CertificateVerifyError
,
797 match
=r
'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate'
799 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
801 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
802 @pytest.mark.parametrize('req,match,version_check', [
803 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
804 # bpo-39603: Check implemented in 3.7.9+, 3.8.5+
806 Request('http://127.0.0.1', method
='GET\n'),
807 'method can\'t contain control characters',
808 lambda v
: v
< (3, 7, 9) or (3, 8, 0) <= v
< (3, 8, 5)
810 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
811 # bpo-38576: Check implemented in 3.7.8+, 3.8.3+
813 Request('http://127.0.0. 1', method
='GET'),
814 'URL can\'t contain control characters',
815 lambda v
: v
< (3, 7, 8) or (3, 8, 0) <= v
< (3, 8, 3)
817 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
818 (Request('http://127.0.0.1', headers
={'foo\n': 'bar'}
), 'Invalid header name', None),
820 def test_httplib_validation_errors(self
, handler
, req
, match
, version_check
):
821 if version_check
and version_check(sys
.version_info
):
822 pytest
.skip(f
'Python {sys.version} version does not have the required validation for this test.')
824 with handler() as rh
:
825 with pytest
.raises(RequestError
, match
=match
) as exc_info
:
826 validate_and_send(rh
, req
)
827 assert not isinstance(exc_info
.value
, TransportError
)
830 class TestRequestsRequestHandler(TestRequestHandlerBase
):
831 @pytest.mark.parametrize('raised,expected', [
832 (lambda: requests
.exceptions
.ConnectTimeout(), TransportError
),
833 (lambda: requests
.exceptions
.ReadTimeout(), TransportError
),
834 (lambda: requests
.exceptions
.Timeout(), TransportError
),
835 (lambda: requests
.exceptions
.ConnectionError(), TransportError
),
836 (lambda: requests
.exceptions
.ProxyError(), ProxyError
),
837 (lambda: requests
.exceptions
.SSLError('12[CERTIFICATE_VERIFY_FAILED]34'), CertificateVerifyError
),
838 (lambda: requests
.exceptions
.SSLError(), SSLError
),
839 (lambda: requests
.exceptions
.InvalidURL(), RequestError
),
840 (lambda: requests
.exceptions
.InvalidHeader(), RequestError
),
841 # catch-all: https://github.com/psf/requests/blob/main/src/requests/adapters.py#L535
842 (lambda: urllib3
.exceptions
.HTTPError(), TransportError
),
843 (lambda: requests
.exceptions
.RequestException(), RequestError
)
844 # (lambda: requests.exceptions.TooManyRedirects(), HTTPError) - Needs a response object
846 @pytest.mark.parametrize('handler', ['Requests'], indirect
=True)
847 def test_request_error_mapping(self
, handler
, monkeypatch
, raised
, expected
):
848 with handler() as rh
:
849 def mock_get_instance(*args
, **kwargs
):
851 def request(self
, *args
, **kwargs
):
855 monkeypatch
.setattr(rh
, '_get_instance', mock_get_instance
)
857 with pytest
.raises(expected
) as exc_info
:
858 rh
.send(Request('http://fake'))
860 assert exc_info
.type is expected
862 @pytest.mark.parametrize('raised,expected,match', [
863 (lambda: urllib3
.exceptions
.SSLError(), SSLError
, None),
864 (lambda: urllib3
.exceptions
.TimeoutError(), TransportError
, None),
865 (lambda: urllib3
.exceptions
.ReadTimeoutError(None, None, None), TransportError
, None),
866 (lambda: urllib3
.exceptions
.ProtocolError(), TransportError
, None),
867 (lambda: urllib3
.exceptions
.DecodeError(), TransportError
, None),
868 (lambda: urllib3
.exceptions
.HTTPError(), TransportError
, None), # catch-all
870 lambda: urllib3
.exceptions
.ProtocolError('error', http
.client
.IncompleteRead(partial
=b
'abc', expected
=4)),
872 '3 bytes read, 4 more expected'
875 lambda: urllib3
.exceptions
.ProtocolError('error', urllib3
.exceptions
.IncompleteRead(partial
=3, expected
=5)),
877 '3 bytes read, 5 more expected'
880 @pytest.mark.parametrize('handler', ['Requests'], indirect
=True)
881 def test_response_error_mapping(self
, handler
, monkeypatch
, raised
, expected
, match
):
882 from requests
.models
import Response
as RequestsResponse
883 from urllib3
.response
import HTTPResponse
as Urllib3Response
885 from yt_dlp
.networking
._requests
import RequestsResponseAdapter
886 requests_res
= RequestsResponse()
887 requests_res
.raw
= Urllib3Response(body
=b
'', status
=200)
888 res
= RequestsResponseAdapter(requests_res
)
890 def mock_read(*args
, **kwargs
):
892 monkeypatch
.setattr(res
.fp
, 'read', mock_read
)
894 with pytest
.raises(expected
, match
=match
) as exc_info
:
897 assert exc_info
.type is expected
900 def run_validation(handler
, error
, req
, **handler_kwargs
):
901 with handler(**handler_kwargs
) as rh
:
903 with pytest
.raises(error
):
909 class TestRequestHandlerValidation
:
911 class ValidationRH(RequestHandler
):
912 def _send(self
, request
):
913 raise RequestError('test')
915 class NoCheckRH(ValidationRH
):
916 _SUPPORTED_FEATURES
= None
917 _SUPPORTED_PROXY_SCHEMES
= None
918 _SUPPORTED_URL_SCHEMES
= None
920 def _check_extensions(self
, extensions
):
923 class HTTPSupportedRH(ValidationRH
):
924 _SUPPORTED_URL_SCHEMES
= ('http',)
927 # scheme, expected to fail, handler kwargs
930 ('https', False, {}),
933 ('file', UnsupportedRequest
, {}),
934 ('file', False, {'enable_file_urls': True}
),
938 ('https', False, {}),
944 (NoCheckRH
, [('http', False, {})]),
945 (ValidationRH
, [('http', UnsupportedRequest
, {})])
948 PROXY_SCHEME_TESTS
= [
949 # scheme, expected to fail
952 ('https', UnsupportedRequest
),
957 ('socks', UnsupportedRequest
),
959 ('Requests', 'http', [
967 (NoCheckRH
, 'http', [('http', False)]),
968 (HTTPSupportedRH
, 'http', [('http', UnsupportedRequest
)]),
969 ('Websockets', 'ws', [('http', UnsupportedRequest
)]),
970 (NoCheckRH
, 'http', [('http', False)]),
971 (HTTPSupportedRH
, 'http', [('http', UnsupportedRequest
)]),
975 # key, expected to fail
978 ('unrelated', False),
982 ('unrelated', False),
984 (NoCheckRH
, [('all', False)]),
985 (HTTPSupportedRH
, [('all', UnsupportedRequest
)]),
986 (HTTPSupportedRH
, [('no', UnsupportedRequest
)]),
991 ({'cookiejar': 'notacookiejar'}
, AssertionError),
992 ({'cookiejar': YoutubeDLCookieJar()}
, False),
993 ({'cookiejar': CookieJar()}
, AssertionError),
994 ({'timeout': 1}
, False),
995 ({'timeout': 'notatimeout'}
, AssertionError),
996 ({'unsupported': 'value'}
, UnsupportedRequest
),
998 ('Requests', 'http', [
999 ({'cookiejar': 'notacookiejar'}
, AssertionError),
1000 ({'cookiejar': YoutubeDLCookieJar()}
, False),
1001 ({'timeout': 1}
, False),
1002 ({'timeout': 'notatimeout'}
, AssertionError),
1003 ({'unsupported': 'value'}
, UnsupportedRequest
),
1005 (NoCheckRH
, 'http', [
1006 ({'cookiejar': 'notacookiejar'}
, False),
1007 ({'somerandom': 'test'}
, False), # but any extension is allowed through
1009 ('Websockets', 'ws', [
1010 ({'cookiejar': YoutubeDLCookieJar()}
, False),
1011 ({'timeout': 2}
, False),
1015 @pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
1016 (handler_tests
[0], scheme
, fail
, handler_kwargs
)
1017 for handler_tests
in URL_SCHEME_TESTS
1018 for scheme
, fail
, handler_kwargs
in handler_tests
[1]
1020 ], indirect
=['handler'])
1021 def test_url_scheme(self
, handler
, scheme
, fail
, handler_kwargs
):
1022 run_validation(handler
, fail
, Request(f
'{scheme}://'), **(handler_kwargs
or {}))
1024 @pytest.mark.parametrize('handler,fail', [('Urllib', False), ('Requests', False)], indirect
=['handler'])
1025 def test_no_proxy(self
, handler
, fail
):
1026 run_validation(handler
, fail
, Request('http://', proxies
={'no': '127.0.0.1,github.com'}
))
1027 run_validation(handler
, fail
, Request('http://'), proxies
={'no': '127.0.0.1,github.com'}
)
1029 @pytest.mark.parametrize('handler,proxy_key,fail', [
1030 (handler_tests
[0], proxy_key
, fail
)
1031 for handler_tests
in PROXY_KEY_TESTS
1032 for proxy_key
, fail
in handler_tests
[1]
1033 ], indirect
=['handler'])
1034 def test_proxy_key(self
, handler
, proxy_key
, fail
):
1035 run_validation(handler
, fail
, Request('http://', proxies
={proxy_key: 'http://example.com'}
))
1036 run_validation(handler
, fail
, Request('http://'), proxies
={proxy_key: 'http://example.com'}
)
1038 @pytest.mark.parametrize('handler,req_scheme,scheme,fail', [
1039 (handler_tests
[0], handler_tests
[1], scheme
, fail
)
1040 for handler_tests
in PROXY_SCHEME_TESTS
1041 for scheme
, fail
in handler_tests
[2]
1042 ], indirect
=['handler'])
1043 def test_proxy_scheme(self
, handler
, req_scheme
, scheme
, fail
):
1044 run_validation(handler
, fail
, Request(f
'{req_scheme}://', proxies
={req_scheme: f'{scheme}
://example
.com
'}))
1045 run_validation(handler, fail, Request(f'{req_scheme}
://'), proxies={req_scheme: f'{scheme}://example.com'})
1047 @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH
, 'Requests'], indirect
=True)
1048 def test_empty_proxy(self
, handler
):
1049 run_validation(handler
, False, Request('http://', proxies
={'http': None}
))
1050 run_validation(handler
, False, Request('http://'), proxies
={'http': None}
)
1052 @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
1053 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
1054 def test_invalid_proxy_url(self
, handler
, proxy_url
):
1055 run_validation(handler
, UnsupportedRequest
, Request('http://', proxies
={'http': proxy_url}
))
1057 @pytest.mark.parametrize('handler,scheme,extensions,fail', [
1058 (handler_tests
[0], handler_tests
[1], extensions
, fail
)
1059 for handler_tests
in EXTENSION_TESTS
1060 for extensions
, fail
in handler_tests
[2]
1061 ], indirect
=['handler'])
1062 def test_extension(self
, handler
, scheme
, extensions
, fail
):
1064 handler
, fail
, Request(f
'{scheme}://', extensions
=extensions
))
1066 def test_invalid_request_type(self
):
1067 rh
= self
.ValidationRH(logger
=FakeLogger())
1068 for method
in (rh
.validate
, rh
.send
):
1069 with pytest
.raises(TypeError, match
='Expected an instance of Request'):
1070 method('not a request')
1073 class FakeResponse(Response
):
1074 def __init__(self
, request
):
1075 # XXX: we could make request part of standard response interface
1076 self
.request
= request
1077 super().__init
__(fp
=io
.BytesIO(b
''), headers
={}, url
=request
.url
)
1080 class FakeRH(RequestHandler
):
1082 def _validate(self
, request
):
1085 def _send(self
, request
: Request
):
1086 if request
.url
.startswith('ssl://'):
1087 raise SSLError(request
.url
[len('ssl://'):])
1088 return FakeResponse(request
)
1091 class FakeRHYDL(FakeYDL
):
1092 def __init__(self
, *args
, **kwargs
):
1093 super().__init
__(*args
, **kwargs
)
1094 self
._request
_director
= self
.build_request_director([FakeRH
])
1097 class AllUnsupportedRHYDL(FakeYDL
):
1099 def __init__(self
, *args
, **kwargs
):
1101 class UnsupportedRH(RequestHandler
):
1102 def _send(self
, request
: Request
):
1105 _SUPPORTED_FEATURES
= ()
1106 _SUPPORTED_PROXY_SCHEMES
= ()
1107 _SUPPORTED_URL_SCHEMES
= ()
1109 super().__init
__(*args
, **kwargs
)
1110 self
._request
_director
= self
.build_request_director([UnsupportedRH
])
1113 class TestRequestDirector
:
1115 def test_handler_operations(self
):
1116 director
= RequestDirector(logger
=FakeLogger())
1117 handler
= FakeRH(logger
=FakeLogger())
1118 director
.add_handler(handler
)
1119 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler
1121 # Handler should overwrite
1122 handler2
= FakeRH(logger
=FakeLogger())
1123 director
.add_handler(handler2
)
1124 assert director
.handlers
.get(FakeRH
.RH_KEY
) is not handler
1125 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler2
1126 assert len(director
.handlers
) == 1
1128 class AnotherFakeRH(FakeRH
):
1130 director
.add_handler(AnotherFakeRH(logger
=FakeLogger()))
1131 assert len(director
.handlers
) == 2
1132 assert director
.handlers
.get(AnotherFakeRH
.RH_KEY
).RH_KEY
== AnotherFakeRH
.RH_KEY
1134 director
.handlers
.pop(FakeRH
.RH_KEY
, None)
1135 assert director
.handlers
.get(FakeRH
.RH_KEY
) is None
1136 assert len(director
.handlers
) == 1
1138 # RequestErrors should passthrough
1139 with pytest
.raises(SSLError
):
1140 director
.send(Request('ssl://something'))
1142 def test_send(self
):
1143 director
= RequestDirector(logger
=FakeLogger())
1144 with pytest
.raises(RequestError
):
1145 director
.send(Request('any://'))
1146 director
.add_handler(FakeRH(logger
=FakeLogger()))
1147 assert isinstance(director
.send(Request('http://')), FakeResponse
)
1149 def test_unsupported_handlers(self
):
1150 class SupportedRH(RequestHandler
):
1151 _SUPPORTED_URL_SCHEMES
= ['http']
1153 def _send(self
, request
: Request
):
1154 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
1156 director
= RequestDirector(logger
=FakeLogger())
1157 director
.add_handler(SupportedRH(logger
=FakeLogger()))
1158 director
.add_handler(FakeRH(logger
=FakeLogger()))
1160 # First should take preference
1161 assert director
.send(Request('http://')).read() == b
'supported'
1162 assert director
.send(Request('any://')).read() == b
''
1164 director
.handlers
.pop(FakeRH
.RH_KEY
)
1165 with pytest
.raises(NoSupportingHandlers
):
1166 director
.send(Request('any://'))
1168 def test_unexpected_error(self
):
1169 director
= RequestDirector(logger
=FakeLogger())
1171 class UnexpectedRH(FakeRH
):
1172 def _send(self
, request
: Request
):
1173 raise TypeError('something')
1175 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1176 with pytest
.raises(NoSupportingHandlers
, match
=r
'1 unexpected error'):
1177 director
.send(Request('any://'))
1179 director
.handlers
.clear()
1180 assert len(director
.handlers
) == 0
1182 # Should not be fatal
1183 director
.add_handler(FakeRH(logger
=FakeLogger()))
1184 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1185 assert director
.send(Request('any://'))
1187 def test_preference(self
):
1188 director
= RequestDirector(logger
=FakeLogger())
1189 director
.add_handler(FakeRH(logger
=FakeLogger()))
1191 class SomeRH(RequestHandler
):
1192 _SUPPORTED_URL_SCHEMES
= ['http']
1194 def _send(self
, request
: Request
):
1195 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
1197 def some_preference(rh
, request
):
1198 return (0 if not isinstance(rh
, SomeRH
)
1199 else 100 if 'prefer' in request
.headers
1202 director
.add_handler(SomeRH(logger
=FakeLogger()))
1203 director
.preferences
.add(some_preference
)
1205 assert director
.send(Request('http://')).read() == b
''
1206 assert director
.send(Request('http://', headers
={'prefer': '1'}
)).read() == b
'supported'
1209 # XXX: do we want to move this to test_YoutubeDL.py?
1210 class TestYoutubeDLNetworking
:
1213 def build_handler(ydl
, handler
: RequestHandler
= FakeRH
):
1214 return ydl
.build_request_director([handler
]).handlers
.get(handler
.RH_KEY
)
1216 def test_compat_opener(self
):
1217 with FakeYDL() as ydl
:
1218 with warnings
.catch_warnings():
1219 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1220 assert isinstance(ydl
._opener
, urllib
.request
.OpenerDirector
)
1222 @pytest.mark.parametrize('proxy,expected', [
1223 ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}
),
1224 ('', {'all': '__noproxy__'}
),
1225 (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}
) # env, set https
1227 def test_proxy(self
, proxy
, expected
):
1228 old_http_proxy
= os
.environ
.get('HTTP_PROXY')
1230 os
.environ
['HTTP_PROXY'] = 'http://127.0.0.1:8081' # ensure that provided proxies override env
1231 with FakeYDL({'proxy': proxy}
) as ydl
:
1232 assert ydl
.proxies
== expected
1235 os
.environ
['HTTP_PROXY'] = old_http_proxy
1237 def test_compat_request(self
):
1238 with FakeRHYDL() as ydl
:
1239 assert ydl
.urlopen('test://')
1240 urllib_req
= urllib
.request
.Request('http://foo.bar', data
=b
'test', method
='PUT', headers
={'X-Test': '1'}
)
1241 urllib_req
.add_unredirected_header('Cookie', 'bob=bob')
1242 urllib_req
.timeout
= 2
1243 with warnings
.catch_warnings():
1244 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1245 req
= ydl
.urlopen(urllib_req
).request
1246 assert req
.url
== urllib_req
.get_full_url()
1247 assert req
.data
== urllib_req
.data
1248 assert req
.method
== urllib_req
.get_method()
1249 assert 'X-Test' in req
.headers
1250 assert 'Cookie' in req
.headers
1251 assert req
.extensions
.get('timeout') == 2
1253 with pytest
.raises(AssertionError):
1256 def test_extract_basic_auth(self
):
1257 with FakeRHYDL() as ydl
:
1258 res
= ydl
.urlopen(Request('http://user:pass@foo.bar'))
1259 assert res
.request
.headers
['Authorization'] == 'Basic dXNlcjpwYXNz'
1261 def test_sanitize_url(self
):
1262 with FakeRHYDL() as ydl
:
1263 res
= ydl
.urlopen(Request('httpss://foo.bar'))
1264 assert res
.request
.url
== 'https://foo.bar'
1266 def test_file_urls_error(self
):
1267 # use urllib handler
1268 with FakeYDL() as ydl
:
1269 with pytest
.raises(RequestError
, match
=r
'file:// URLs are disabled by default'):
1270 ydl
.urlopen('file://')
1272 @pytest.mark.parametrize('scheme', (['ws', 'wss']))
1273 def test_websocket_unavailable_error(self
, scheme
):
1274 with AllUnsupportedRHYDL() as ydl
:
1275 with pytest
.raises(RequestError
, match
=r
'This request requires WebSocket support'):
1276 ydl
.urlopen(f
'{scheme}://')
1278 def test_legacy_server_connect_error(self
):
1279 with FakeRHYDL() as ydl
:
1280 for error
in ('UNSAFE_LEGACY_RENEGOTIATION_DISABLED', 'SSLV3_ALERT_HANDSHAKE_FAILURE'):
1281 with pytest
.raises(RequestError
, match
=r
'Try using --legacy-server-connect'):
1282 ydl
.urlopen(f
'ssl://{error}')
1284 with pytest
.raises(SSLError
, match
='testerror'):
1285 ydl
.urlopen('ssl://testerror')
1287 @pytest.mark.parametrize('proxy_key,proxy_url,expected', [
1288 ('http', '__noproxy__', None),
1289 ('no', '127.0.0.1,foo.bar', '127.0.0.1,foo.bar'),
1290 ('https', 'example.com', 'http://example.com'),
1291 ('https', '//example.com', 'http://example.com'),
1292 ('https', 'socks5://example.com', 'socks5h://example.com'),
1293 ('http', 'socks://example.com', 'socks4://example.com'),
1294 ('http', 'socks4://example.com', 'socks4://example.com'),
1295 ('unrelated', '/bad/proxy', '/bad/proxy'), # clean_proxies should ignore bad proxies
1297 def test_clean_proxy(self
, proxy_key
, proxy_url
, expected
):
1298 # proxies should be cleaned in urlopen()
1299 with FakeRHYDL() as ydl
:
1300 req
= ydl
.urlopen(Request('test://', proxies
={proxy_key: proxy_url}
)).request
1301 assert req
.proxies
[proxy_key
] == expected
1303 # and should also be cleaned when building the handler
1304 env_key
= f
'{proxy_key.upper()}_PROXY'
1305 old_env_proxy
= os
.environ
.get(env_key
)
1307 os
.environ
[env_key
] = proxy_url
# ensure that provided proxies override env
1308 with FakeYDL() as ydl
:
1309 rh
= self
.build_handler(ydl
)
1310 assert rh
.proxies
[proxy_key
] == expected
1313 os
.environ
[env_key
] = old_env_proxy
1315 def test_clean_proxy_header(self
):
1316 with FakeRHYDL() as ydl
:
1317 req
= ydl
.urlopen(Request('test://', headers
={'ytdl-request-proxy': '//foo.bar'}
)).request
1318 assert 'ytdl-request-proxy' not in req
.headers
1319 assert req
.proxies
== {'all': 'http://foo.bar'}
1321 with FakeYDL({'http_headers': {'ytdl-request-proxy': '//foo.bar'}
}) as ydl
:
1322 rh
= self
.build_handler(ydl
)
1323 assert 'ytdl-request-proxy' not in rh
.headers
1324 assert rh
.proxies
== {'all': 'http://foo.bar'}
1326 def test_clean_header(self
):
1327 with FakeRHYDL() as ydl
:
1328 res
= ydl
.urlopen(Request('test://', headers
={'Youtubedl-no-compression': True}
))
1329 assert 'Youtubedl-no-compression' not in res
.request
.headers
1330 assert res
.request
.headers
.get('Accept-Encoding') == 'identity'
1332 with FakeYDL({'http_headers': {'Youtubedl-no-compression': True}
}) as ydl
:
1333 rh
= self
.build_handler(ydl
)
1334 assert 'Youtubedl-no-compression' not in rh
.headers
1335 assert rh
.headers
.get('Accept-Encoding') == 'identity'
1337 with FakeYDL({'http_headers': {'Ytdl-socks-proxy': 'socks://localhost:1080'}
}) as ydl
:
1338 rh
= self
.build_handler(ydl
)
1339 assert 'Ytdl-socks-proxy' not in rh
.headers
1341 def test_build_handler_params(self
):
1343 'http_headers': {'test': 'testtest'}
,
1344 'socket_timeout': 2,
1345 'proxy': 'http://127.0.0.1:8080',
1346 'source_address': '127.0.0.45',
1347 'debug_printtraffic': True,
1348 'compat_opts': ['no-certifi'],
1349 'nocheckcertificate': True,
1350 'legacyserverconnect': True,
1352 rh
= self
.build_handler(ydl
)
1353 assert rh
.headers
.get('test') == 'testtest'
1354 assert 'Accept' in rh
.headers
# ensure std_headers are still there
1355 assert rh
.timeout
== 2
1356 assert rh
.proxies
.get('all') == 'http://127.0.0.1:8080'
1357 assert rh
.source_address
== '127.0.0.45'
1358 assert rh
.verbose
is True
1359 assert rh
.prefer_system_certs
is True
1360 assert rh
.verify
is False
1361 assert rh
.legacy_ssl_support
is True
1363 @pytest.mark.parametrize('ydl_params', [
1364 {'client_certificate': 'fakecert.crt'}
,
1365 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key'}
,
1366 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1367 {'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1369 def test_client_certificate(self
, ydl_params
):
1370 with FakeYDL(ydl_params
) as ydl
:
1371 rh
= self
.build_handler(ydl
)
1372 assert rh
._client
_cert
== ydl_params
# XXX: Too bound to implementation
1374 def test_urllib_file_urls(self
):
1375 with FakeYDL({'enable_file_urls': False}
) as ydl
:
1376 rh
= self
.build_handler(ydl
, UrllibRH
)
1377 assert rh
.enable_file_urls
is False
1379 with FakeYDL({'enable_file_urls': True}
) as ydl
:
1380 rh
= self
.build_handler(ydl
, UrllibRH
)
1381 assert rh
.enable_file_urls
is True
1383 def test_compat_opt_prefer_urllib(self
):
1384 # This assumes urllib only has a preference when this compat opt is given
1385 with FakeYDL({'compat_opts': ['prefer-legacy-http-handler']}
) as ydl
:
1386 director
= ydl
.build_request_director([UrllibRH
])
1387 assert len(director
.preferences
) == 1
1388 assert director
.preferences
.pop()(UrllibRH
, None)
1393 def test_query(self
):
1394 req
= Request('http://example.com?q=something', query
={'v': 'xyz'}
)
1395 assert req
.url
== 'http://example.com?q=something&v=xyz'
1397 req
.update(query
={'v': '123'}
)
1398 assert req
.url
== 'http://example.com?q=something&v=123'
1399 req
.update(url
='http://example.com', query
={'v': 'xyz'}
)
1400 assert req
.url
== 'http://example.com?v=xyz'
1402 def test_method(self
):
1403 req
= Request('http://example.com')
1404 assert req
.method
== 'GET'
1406 assert req
.method
== 'POST'
1408 assert req
.method
== 'GET'
1411 assert req
.method
== 'PUT'
1413 assert req
.method
== 'PUT'
1414 with pytest
.raises(TypeError):
1417 def test_request_helpers(self
):
1418 assert HEADRequest('http://example.com').method
== 'HEAD'
1419 assert PUTRequest('http://example.com').method
== 'PUT'
1421 def test_headers(self
):
1422 req
= Request('http://example.com', headers
={'tesT': 'test'}
)
1423 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1424 req
.update(headers
={'teSt2': 'test2'}
)
1425 assert req
.headers
== HTTPHeaderDict({'test': 'test', 'test2': 'test2'}
)
1427 req
.headers
= new_headers
= HTTPHeaderDict({'test': 'test'}
)
1428 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1429 assert req
.headers
is new_headers
1431 # test converts dict to case insensitive dict
1432 req
.headers
= new_headers
= {'test2': 'test2'}
1433 assert isinstance(req
.headers
, HTTPHeaderDict
)
1434 assert req
.headers
is not new_headers
1436 with pytest
.raises(TypeError):
1439 def test_data_type(self
):
1440 req
= Request('http://example.com')
1441 assert req
.data
is None
1442 # test bytes is allowed
1444 assert req
.data
== b
'test'
1445 # test iterable of bytes is allowed
1446 i
= [b
'test', b
'test2']
1448 assert req
.data
== i
1450 # test file-like object is allowed
1451 f
= io
.BytesIO(b
'test')
1453 assert req
.data
== f
1455 # common mistake: test str not allowed
1456 with pytest
.raises(TypeError):
1458 assert req
.data
!= 'test'
1460 # common mistake: test dict is not allowed
1461 with pytest
.raises(TypeError):
1462 req
.data
= {'test': 'test'}
1463 assert req
.data
!= {'test': 'test'}
1465 def test_content_length_header(self
):
1466 req
= Request('http://example.com', headers
={'Content-Length': '0'}
, data
=b
'')
1467 assert req
.headers
.get('Content-Length') == '0'
1470 assert 'Content-Length' not in req
.headers
1472 req
= Request('http://example.com', headers
={'Content-Length': '10'}
)
1473 assert 'Content-Length' not in req
.headers
1475 def test_content_type_header(self
):
1476 req
= Request('http://example.com', headers
={'Content-Type': 'test'}
, data
=b
'test')
1477 assert req
.headers
.get('Content-Type') == 'test'
1479 assert req
.headers
.get('Content-Type') == 'test'
1481 assert 'Content-Type' not in req
.headers
1483 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1485 def test_update_req(self
):
1486 req
= Request('http://example.com')
1487 assert req
.data
is None
1488 assert req
.method
== 'GET'
1489 assert 'Content-Type' not in req
.headers
1490 # Test that zero-byte payloads will be sent
1491 req
.update(data
=b
'')
1492 assert req
.data
== b
''
1493 assert req
.method
== 'POST'
1494 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1496 def test_proxies(self
):
1497 req
= Request(url
='http://example.com', proxies
={'http': 'http://127.0.0.1:8080'}
)
1498 assert req
.proxies
== {'http': 'http://127.0.0.1:8080'}
1500 def test_extensions(self
):
1501 req
= Request(url
='http://example.com', extensions
={'timeout': 2}
)
1502 assert req
.extensions
== {'timeout': 2}
1504 def test_copy(self
):
1506 url
='http://example.com',
1507 extensions
={'cookiejar': CookieJar()}
,
1508 headers
={'Accept-Encoding': 'br'}
,
1509 proxies
={'http': 'http://127.0.0.1'}
,
1512 req_copy
= req
.copy()
1513 assert req_copy
is not req
1514 assert req_copy
.url
== req
.url
1515 assert req_copy
.headers
== req
.headers
1516 assert req_copy
.headers
is not req
.headers
1517 assert req_copy
.proxies
== req
.proxies
1518 assert req_copy
.proxies
is not req
.proxies
1520 # Data is not able to be copied
1521 assert req_copy
.data
== req
.data
1522 assert req_copy
.data
is req
.data
1524 # Shallow copy extensions
1525 assert req_copy
.extensions
is not req
.extensions
1526 assert req_copy
.extensions
['cookiejar'] == req
.extensions
['cookiejar']
1528 # Subclasses are copied by default
1529 class AnotherRequest(Request
):
1532 req
= AnotherRequest(url
='http://127.0.0.1')
1533 assert isinstance(req
.copy(), AnotherRequest
)
1536 req
= Request(url
='https://фtest.example.com/ some spaceв?ä=c',)
1537 assert req
.url
== 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c'
1539 assert Request(url
='//example.com').url
== 'http://example.com'
1541 with pytest
.raises(TypeError):
1542 Request(url
='https://').url
= None
1547 @pytest.mark.parametrize('reason,status,expected', [
1548 ('custom', 200, 'custom'),
1549 (None, 404, 'Not Found'), # fallback status
1550 ('', 403, 'Forbidden'),
1553 def test_reason(self
, reason
, status
, expected
):
1554 res
= Response(io
.BytesIO(b
''), url
='test://', headers
={}, status
=status
, reason
=reason
)
1555 assert res
.reason
== expected
1557 def test_headers(self
):
1559 headers
.add_header('Test', 'test')
1560 headers
.add_header('Test', 'test2')
1561 headers
.add_header('content-encoding', 'br')
1562 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1563 assert res
.headers
.get_all('test') == ['test', 'test2']
1564 assert 'Content-Encoding' in res
.headers
1566 def test_get_header(self
):
1568 headers
.add_header('Set-Cookie', 'cookie1')
1569 headers
.add_header('Set-cookie', 'cookie2')
1570 headers
.add_header('Test', 'test')
1571 headers
.add_header('Test', 'test2')
1572 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1573 assert res
.get_header('test') == 'test, test2'
1574 assert res
.get_header('set-Cookie') == 'cookie1'
1575 assert res
.get_header('notexist', 'default') == 'default'
1577 def test_compat(self
):
1578 res
= Response(io
.BytesIO(b
''), url
='test://', status
=404, headers
={'test': 'test'}
)
1579 with warnings
.catch_warnings():
1580 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1581 assert res
.code
== res
.getcode() == res
.status
1582 assert res
.geturl() == res
.url
1583 assert res
.info() is res
.headers
1584 assert res
.getheader('test') == res
.get_header('test')