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
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 TEST_DIR
= os
.path
.dirname(os
.path
.abspath(__file__
))
58 def _build_proxy_handler(name
):
59 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
62 def log_message(self
, format
, *args
):
66 self
.send_response(200)
67 self
.send_header('Content-Type', 'text/plain; charset=utf-8')
69 self
.wfile
.write('{self.proxy_name}: {self.path}'.format(self
=self
).encode())
70 return HTTPTestRequestHandler
73 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
74 protocol_version
= 'HTTP/1.1'
76 def log_message(self
, format
, *args
):
80 payload
= str(self
.headers
).encode()
81 self
.send_response(200)
82 self
.send_header('Content-Type', 'application/json')
83 self
.send_header('Content-Length', str(len(payload
)))
85 self
.wfile
.write(payload
)
88 self
.send_response(int(self
.path
[len('/redirect_'):]))
89 self
.send_header('Location', '/method')
90 self
.send_header('Content-Length', '0')
93 def _method(self
, method
, payload
=None):
94 self
.send_response(200)
95 self
.send_header('Content-Length', str(len(payload
or '')))
96 self
.send_header('Method', method
)
99 self
.wfile
.write(payload
)
101 def _status(self
, status
):
102 payload
= f
'<html>{status} NOT FOUND</html>'.encode()
103 self
.send_response(int(status
))
104 self
.send_header('Content-Type', 'text/html; charset=utf-8')
105 self
.send_header('Content-Length', str(len(payload
)))
107 self
.wfile
.write(payload
)
109 def _read_data(self
):
110 if 'Content-Length' in self
.headers
:
111 return self
.rfile
.read(int(self
.headers
['Content-Length']))
114 data
= self
._read
_data
() + str(self
.headers
).encode()
115 if self
.path
.startswith('/redirect_'):
117 elif self
.path
.startswith('/method'):
118 self
._method
('POST', data
)
119 elif self
.path
.startswith('/headers'):
125 if self
.path
.startswith('/redirect_'):
127 elif self
.path
.startswith('/method'):
133 data
= self
._read
_data
() + str(self
.headers
).encode()
134 if self
.path
.startswith('/redirect_'):
136 elif self
.path
.startswith('/method'):
137 self
._method
('PUT', data
)
142 if self
.path
== '/video.html':
143 payload
= b
'<html><video src="/vid.mp4" /></html>'
144 self
.send_response(200)
145 self
.send_header('Content-Type', 'text/html; charset=utf-8')
146 self
.send_header('Content-Length', str(len(payload
)))
148 self
.wfile
.write(payload
)
149 elif self
.path
== '/vid.mp4':
150 payload
= b
'\x00\x00\x00\x00\x20\x66\x74[video]'
151 self
.send_response(200)
152 self
.send_header('Content-Type', 'video/mp4')
153 self
.send_header('Content-Length', str(len(payload
)))
155 self
.wfile
.write(payload
)
156 elif self
.path
== '/%E4%B8%AD%E6%96%87.html':
157 payload
= b
'<html><video src="/vid.mp4" /></html>'
158 self
.send_response(200)
159 self
.send_header('Content-Type', 'text/html; charset=utf-8')
160 self
.send_header('Content-Length', str(len(payload
)))
162 self
.wfile
.write(payload
)
163 elif self
.path
== '/%c7%9f':
164 payload
= b
'<html><video src="/vid.mp4" /></html>'
165 self
.send_response(200)
166 self
.send_header('Content-Type', 'text/html; charset=utf-8')
167 self
.send_header('Content-Length', str(len(payload
)))
169 self
.wfile
.write(payload
)
170 elif self
.path
.startswith('/redirect_loop'):
171 self
.send_response(301)
172 self
.send_header('Location', self
.path
)
173 self
.send_header('Content-Length', '0')
175 elif self
.path
== '/redirect_dotsegments':
176 self
.send_response(301)
177 # redirect to /headers but with dot segments before
178 self
.send_header('Location', '/a/b/./../../headers')
179 self
.send_header('Content-Length', '0')
181 elif self
.path
.startswith('/redirect_'):
183 elif self
.path
.startswith('/method'):
184 self
._method
('GET', str(self
.headers
).encode())
185 elif self
.path
.startswith('/headers'):
187 elif self
.path
.startswith('/308-to-headers'):
188 self
.send_response(308)
189 self
.send_header('Location', '/headers')
190 self
.send_header('Content-Length', '0')
192 elif self
.path
== '/trailing_garbage':
193 payload
= b
'<html><video src="/vid.mp4" /></html>'
194 self
.send_response(200)
195 self
.send_header('Content-Type', 'text/html; charset=utf-8')
196 self
.send_header('Content-Encoding', 'gzip')
198 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
200 compressed
= buf
.getvalue() + b
'trailing garbage'
201 self
.send_header('Content-Length', str(len(compressed
)))
203 self
.wfile
.write(compressed
)
204 elif self
.path
== '/302-non-ascii-redirect':
205 new_url
= f
'http://127.0.0.1:{http_server_port(self.server)}/中文.html'
206 self
.send_response(301)
207 self
.send_header('Location', new_url
)
208 self
.send_header('Content-Length', '0')
210 elif self
.path
== '/content-encoding':
211 encodings
= self
.headers
.get('ytdl-encoding', '')
212 payload
= b
'<html><video src="/vid.mp4" /></html>'
213 for encoding
in filter(None, (e
.strip() for e
in encodings
.split(','))):
214 if encoding
== 'br' and brotli
:
215 payload
= brotli
.compress(payload
)
216 elif encoding
== 'gzip':
218 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
220 payload
= buf
.getvalue()
221 elif encoding
== 'deflate':
222 payload
= zlib
.compress(payload
)
223 elif encoding
== 'unsupported':
229 self
.send_response(200)
230 self
.send_header('Content-Encoding', encodings
)
231 self
.send_header('Content-Length', str(len(payload
)))
233 self
.wfile
.write(payload
)
234 elif self
.path
.startswith('/gen_'):
235 payload
= b
'<html></html>'
236 self
.send_response(int(self
.path
[len('/gen_'):]))
237 self
.send_header('Content-Type', 'text/html; charset=utf-8')
238 self
.send_header('Content-Length', str(len(payload
)))
240 self
.wfile
.write(payload
)
241 elif self
.path
.startswith('/incompleteread'):
242 payload
= b
'<html></html>'
243 self
.send_response(200)
244 self
.send_header('Content-Type', 'text/html; charset=utf-8')
245 self
.send_header('Content-Length', '234234')
247 self
.wfile
.write(payload
)
249 elif self
.path
.startswith('/timeout_'):
250 time
.sleep(int(self
.path
[len('/timeout_'):]))
252 elif self
.path
== '/source_address':
253 payload
= str(self
.client_address
[0]).encode()
254 self
.send_response(200)
255 self
.send_header('Content-Type', 'text/html; charset=utf-8')
256 self
.send_header('Content-Length', str(len(payload
)))
258 self
.wfile
.write(payload
)
263 def send_header(self
, keyword
, value
):
265 Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
266 This is against what is defined in RFC 3986, however we need to test we support this
267 since some sites incorrectly do this.
269 if keyword
.lower() == 'connection':
270 return super().send_header(keyword
, value
)
272 if not hasattr(self
, '_headers_buffer'):
273 self
._headers
_buffer
= []
275 self
._headers
_buffer
.append(f
'{keyword}: {value}\r\n'.encode())
278 def validate_and_send(rh
, req
):
283 class TestRequestHandlerBase
:
285 def setup_class(cls
):
286 cls
.http_httpd
= http
.server
.ThreadingHTTPServer(
287 ('127.0.0.1', 0), HTTPTestRequestHandler
)
288 cls
.http_port
= http_server_port(cls
.http_httpd
)
289 cls
.http_server_thread
= threading
.Thread(target
=cls
.http_httpd
.serve_forever
)
290 # FIXME: we should probably stop the http server thread after each test
291 # See: https://github.com/yt-dlp/yt-dlp/pull/7094#discussion_r1199746041
292 cls
.http_server_thread
.daemon
= True
293 cls
.http_server_thread
.start()
296 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
297 cls
.https_httpd
= http
.server
.ThreadingHTTPServer(
298 ('127.0.0.1', 0), HTTPTestRequestHandler
)
299 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
300 sslctx
.load_cert_chain(certfn
, None)
301 cls
.https_httpd
.socket
= sslctx
.wrap_socket(cls
.https_httpd
.socket
, server_side
=True)
302 cls
.https_port
= http_server_port(cls
.https_httpd
)
303 cls
.https_server_thread
= threading
.Thread(target
=cls
.https_httpd
.serve_forever
)
304 cls
.https_server_thread
.daemon
= True
305 cls
.https_server_thread
.start()
308 class TestHTTPRequestHandler(TestRequestHandlerBase
):
309 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
310 def test_verify_cert(self
, handler
):
311 with handler() as rh
:
312 with pytest
.raises(CertificateVerifyError
):
313 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
315 with handler(verify
=False) as rh
:
316 r
= validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
317 assert r
.status
== 200
320 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
321 def test_ssl_error(self
, handler
):
322 # HTTPS server with too old TLS version
323 # XXX: is there a better way to test this than to create a new server?
324 https_httpd
= http
.server
.ThreadingHTTPServer(
325 ('127.0.0.1', 0), HTTPTestRequestHandler
)
326 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
327 https_httpd
.socket
= sslctx
.wrap_socket(https_httpd
.socket
, server_side
=True)
328 https_port
= http_server_port(https_httpd
)
329 https_server_thread
= threading
.Thread(target
=https_httpd
.serve_forever
)
330 https_server_thread
.daemon
= True
331 https_server_thread
.start()
333 with handler(verify
=False) as rh
:
334 with pytest
.raises(SSLError
, match
='sslv3 alert handshake failure') as exc_info
:
335 validate_and_send(rh
, Request(f
'https://127.0.0.1:{https_port}/headers'))
336 assert not issubclass(exc_info
.type, CertificateVerifyError
)
338 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
339 def test_percent_encode(self
, handler
):
340 with handler() as rh
:
341 # Unicode characters should be encoded with uppercase percent-encoding
342 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/中文.html'))
343 assert res
.status
== 200
345 # don't normalize existing percent encodings
346 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/%c7%9f'))
347 assert res
.status
== 200
350 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
351 def test_remove_dot_segments(self
, handler
):
352 with handler() as rh
:
353 # This isn't a comprehensive test,
354 # but it should be enough to check whether the handler is removing dot segments
355 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/a/b/./../../headers'))
356 assert res
.status
== 200
357 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
360 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_dotsegments'))
361 assert res
.status
== 200
362 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
365 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
366 def test_unicode_path_redirection(self
, handler
):
367 with handler() as rh
:
368 r
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
369 assert r
.url
== f
'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
372 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
373 def test_raise_http_error(self
, handler
):
374 with handler() as rh
:
375 for bad_status
in (400, 500, 599, 302):
376 with pytest
.raises(HTTPError
):
377 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_%d' % (self
.http_port
, bad_status
)))
379 # Should not raise an error
380 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_200' % self
.http_port
)).close()
382 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
383 def test_response_url(self
, handler
):
384 with handler() as rh
:
385 # Response url should be that of the last url in redirect chain
386 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_301'))
387 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/method'
389 res2
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_200'))
390 assert res2
.url
== f
'http://127.0.0.1:{self.http_port}/gen_200'
393 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
394 def test_redirect(self
, handler
):
395 with handler() as rh
:
396 def do_req(redirect_status
, method
, assert_no_content
=False):
397 data
= b
'testdata' if method
in ('POST', 'PUT') else None
398 res
= validate_and_send(
399 rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method
=method
, data
=data
))
404 data_sent
+= res
.read(len(data
))
405 if data_sent
!= data
:
409 headers
+= res
.read()
411 if assert_no_content
or data
is None:
412 assert b
'Content-Type' not in headers
413 assert b
'Content-Length' not in headers
415 assert b
'Content-Type' in headers
416 assert b
'Content-Length' in headers
418 return data_sent
.decode(), res
.headers
.get('method', '')
420 # A 303 must either use GET or HEAD for subsequent request
421 assert do_req(303, 'POST', True) == ('', 'GET')
422 assert do_req(303, 'HEAD') == ('', 'HEAD')
424 assert do_req(303, 'PUT', True) == ('', 'GET')
426 # 301 and 302 turn POST only into a GET
427 assert do_req(301, 'POST', True) == ('', 'GET')
428 assert do_req(301, 'HEAD') == ('', 'HEAD')
429 assert do_req(302, 'POST', True) == ('', 'GET')
430 assert do_req(302, 'HEAD') == ('', 'HEAD')
432 assert do_req(301, 'PUT') == ('testdata', 'PUT')
433 assert do_req(302, 'PUT') == ('testdata', 'PUT')
435 # 307 and 308 should not change method
436 for m
in ('POST', 'PUT'):
437 assert do_req(307, m
) == ('testdata', m
)
438 assert do_req(308, m
) == ('testdata', m
)
440 assert do_req(307, 'HEAD') == ('', 'HEAD')
441 assert do_req(308, 'HEAD') == ('', 'HEAD')
443 # These should not redirect and instead raise an HTTPError
444 for code
in (300, 304, 305, 306):
445 with pytest
.raises(HTTPError
):
448 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
449 def test_request_cookie_header(self
, handler
):
450 # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
451 with handler() as rh
:
452 # Specified Cookie header should be used
453 res
= validate_and_send(
455 f
'http://127.0.0.1:{self.http_port}/headers',
456 headers
={'Cookie': 'test=test'}
)).read().decode()
457 assert 'Cookie: test=test' in res
459 # Specified Cookie header should be removed on any redirect
460 res
= validate_and_send(
462 f
'http://127.0.0.1:{self.http_port}/308-to-headers',
463 headers
={'Cookie': 'test=test'}
)).read().decode()
464 assert 'Cookie: test=test' not in res
466 # Specified Cookie header should override global cookiejar for that request
467 cookiejar
= YoutubeDLCookieJar()
468 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
469 version
=0, name
='test', value
='ytdlp', port
=None, port_specified
=False,
470 domain
='127.0.0.1', domain_specified
=True, domain_initial_dot
=False, path
='/',
471 path_specified
=True, secure
=False, expires
=None, discard
=False, comment
=None,
472 comment_url
=None, rest
={}))
474 with handler(cookiejar
=cookiejar
) as rh
:
475 data
= validate_and_send(
476 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', headers
={'cookie': 'test=test'}
)).read()
477 assert b
'Cookie: test=ytdlp' not in data
478 assert b
'Cookie: test=test' in data
480 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
481 def test_redirect_loop(self
, handler
):
482 with handler() as rh
:
483 with pytest
.raises(HTTPError
, match
='redirect loop'):
484 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_loop'))
486 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
487 def test_incompleteread(self
, handler
):
488 with handler(timeout
=2) as rh
:
489 with pytest
.raises(IncompleteRead
):
490 validate_and_send(rh
, Request('http://127.0.0.1:%d/incompleteread' % self
.http_port
)).read()
492 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
493 def test_cookies(self
, handler
):
494 cookiejar
= YoutubeDLCookieJar()
495 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
496 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
497 False, '/headers', True, False, None, False, None, None, {}))
499 with handler(cookiejar
=cookiejar
) as rh
:
500 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
501 assert b
'Cookie: test=ytdlp' in data
504 with handler() as rh
:
505 data
= validate_and_send(
506 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', extensions
={'cookiejar': cookiejar}
)).read()
507 assert b
'Cookie: test=ytdlp' in data
509 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
510 def test_headers(self
, handler
):
512 with handler(headers
=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'}
)) as rh
:
514 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
515 assert b
'Test1: test' in data
517 # Per request headers, merged with global
518 data
= validate_and_send(rh
, Request(
519 f
'http://127.0.0.1:{self.http_port}/headers', headers
={'test2': 'changed', 'test3': 'test3'}
)).read()
520 assert b
'Test1: test' in data
521 assert b
'Test2: changed' in data
522 assert b
'Test2: test2' not in data
523 assert b
'Test3: test3' in data
525 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
526 def test_timeout(self
, handler
):
527 with handler() as rh
:
528 # Default timeout is 20 seconds, so this should go through
530 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_3'))
532 with handler(timeout
=0.5) as rh
:
533 with pytest
.raises(TransportError
):
535 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1'))
537 # Per request timeout, should override handler timeout
539 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1', extensions
={'timeout': 4}
))
541 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
542 def test_source_address(self
, handler
):
543 source_address
= f
'127.0.0.{random.randint(5, 255)}'
544 with handler(source_address
=source_address
) as rh
:
545 data
= validate_and_send(
546 rh
, Request(f
'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
547 assert source_address
== data
549 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
550 def test_gzip_trailing_garbage(self
, handler
):
551 with handler() as rh
:
552 data
= validate_and_send(rh
, Request(f
'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
553 assert data
== '<html><video src="/vid.mp4" /></html>'
555 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
556 @pytest.mark.skipif(not brotli
, reason
='brotli support is not installed')
557 def test_brotli(self
, handler
):
558 with handler() as rh
:
559 res
= validate_and_send(
561 f
'http://127.0.0.1:{self.http_port}/content-encoding',
562 headers
={'ytdl-encoding': 'br'}
))
563 assert res
.headers
.get('Content-Encoding') == 'br'
564 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
566 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
567 def test_deflate(self
, handler
):
568 with handler() as rh
:
569 res
= validate_and_send(
571 f
'http://127.0.0.1:{self.http_port}/content-encoding',
572 headers
={'ytdl-encoding': 'deflate'}
))
573 assert res
.headers
.get('Content-Encoding') == 'deflate'
574 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
576 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
577 def test_gzip(self
, handler
):
578 with handler() as rh
:
579 res
= validate_and_send(
581 f
'http://127.0.0.1:{self.http_port}/content-encoding',
582 headers
={'ytdl-encoding': 'gzip'}
))
583 assert res
.headers
.get('Content-Encoding') == 'gzip'
584 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
586 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
587 def test_multiple_encodings(self
, handler
):
588 with handler() as rh
:
589 for pair
in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
590 res
= validate_and_send(
592 f
'http://127.0.0.1:{self.http_port}/content-encoding',
593 headers
={'ytdl-encoding': pair}
))
594 assert res
.headers
.get('Content-Encoding') == pair
595 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
597 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
598 def test_unsupported_encoding(self
, handler
):
599 with handler() as rh
:
600 res
= validate_and_send(
602 f
'http://127.0.0.1:{self.http_port}/content-encoding',
603 headers
={'ytdl-encoding': 'unsupported'}
))
604 assert res
.headers
.get('Content-Encoding') == 'unsupported'
605 assert res
.read() == b
'raw'
607 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
608 def test_read(self
, handler
):
609 with handler() as rh
:
610 res
= validate_and_send(
611 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers'))
612 assert res
.readable()
613 assert res
.read(1) == b
'H'
614 assert res
.read(3) == b
'ost'
617 class TestHTTPProxy(TestRequestHandlerBase
):
619 def setup_class(cls
):
620 super().setup_class()
622 cls
.proxy
= http
.server
.ThreadingHTTPServer(
623 ('127.0.0.1', 0), _build_proxy_handler('normal'))
624 cls
.proxy_port
= http_server_port(cls
.proxy
)
625 cls
.proxy_thread
= threading
.Thread(target
=cls
.proxy
.serve_forever
)
626 cls
.proxy_thread
.daemon
= True
627 cls
.proxy_thread
.start()
630 cls
.geo_proxy
= http
.server
.ThreadingHTTPServer(
631 ('127.0.0.1', 0), _build_proxy_handler('geo'))
632 cls
.geo_port
= http_server_port(cls
.geo_proxy
)
633 cls
.geo_proxy_thread
= threading
.Thread(target
=cls
.geo_proxy
.serve_forever
)
634 cls
.geo_proxy_thread
.daemon
= True
635 cls
.geo_proxy_thread
.start()
637 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
638 def test_http_proxy(self
, handler
):
639 http_proxy
= f
'http://127.0.0.1:{self.proxy_port}'
640 geo_proxy
= f
'http://127.0.0.1:{self.geo_port}'
642 # Test global http proxy
643 # Test per request http proxy
644 # Test per request http proxy disables proxy
645 url
= 'http://foo.com/bar'
648 with handler(proxies
={'http': http_proxy}
) as rh
:
649 res
= validate_and_send(rh
, Request(url
)).read().decode()
650 assert res
== f
'normal: {url}'
652 # Per request proxy overrides global
653 res
= validate_and_send(rh
, Request(url
, proxies
={'http': geo_proxy}
)).read().decode()
654 assert res
== f
'geo: {url}'
656 # and setting to None disables all proxies for that request
657 real_url
= f
'http://127.0.0.1:{self.http_port}/headers'
658 res
= validate_and_send(
659 rh
, Request(real_url
, proxies
={'http': None}
)).read().decode()
660 assert res
!= f
'normal: {real_url}'
661 assert 'Accept' in res
663 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
664 def test_noproxy(self
, handler
):
665 with handler(proxies
={'proxy': f'http://127.0.0.1:{self.proxy_port}
'}) as rh:
667 for no_proxy in (f'127.0.0.1:{self.http_port}
', '127.0.0.1', 'localhost
'):
668 nop_response = validate_and_send(
669 rh, Request(f'http
://127.0.0.1:{self.http_port}
/headers
', proxies={'no': no_proxy})).read().decode(
671 assert 'Accept
' in nop_response
673 @pytest.mark.parametrize('handler
', ['Urllib
', 'Requests
'], indirect=True)
674 def test_allproxy(self, handler):
675 url = 'http
://foo
.com
/bar
'
676 with handler() as rh:
677 response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
679 assert response
== f
'normal: {url}'
681 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
682 def test_http_proxy_with_idn(self
, handler
):
683 with handler(proxies
={
684 'http': f
'http://127.0.0.1:{self.proxy_port}',
686 url
= 'http://中文.tw/'
687 response
= rh
.send(Request(url
)).read().decode()
688 # b'xn--fiq228c' is '中文'.encode('idna')
689 assert response
== 'normal: http://xn--fiq228c.tw/'
692 class TestClientCertificate
:
695 def setup_class(cls
):
696 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
697 cls
.certdir
= os
.path
.join(TEST_DIR
, 'testdata', 'certificate')
698 cacertfn
= os
.path
.join(cls
.certdir
, 'ca.crt')
699 cls
.httpd
= http
.server
.ThreadingHTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler
)
700 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
701 sslctx
.verify_mode
= ssl
.CERT_REQUIRED
702 sslctx
.load_verify_locations(cafile
=cacertfn
)
703 sslctx
.load_cert_chain(certfn
, None)
704 cls
.httpd
.socket
= sslctx
.wrap_socket(cls
.httpd
.socket
, server_side
=True)
705 cls
.port
= http_server_port(cls
.httpd
)
706 cls
.server_thread
= threading
.Thread(target
=cls
.httpd
.serve_forever
)
707 cls
.server_thread
.daemon
= True
708 cls
.server_thread
.start()
710 def _run_test(self
, handler
, **handler_kwargs
):
712 # Disable client-side validation of unacceptable self-signed testcert.pem
713 # The test is of a check on the server side, so unaffected
717 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.port}/video.html')).read().decode()
719 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
720 def test_certificate_combined_nopass(self
, handler
):
721 self
._run
_test
(handler
, client_cert
={
722 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithkey.crt'),
725 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
726 def test_certificate_nocombined_nopass(self
, handler
):
727 self
._run
_test
(handler
, client_cert
={
728 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
729 'client_certificate_key': os
.path
.join(self
.certdir
, 'client.key'),
732 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
733 def test_certificate_combined_pass(self
, handler
):
734 self
._run
_test
(handler
, client_cert
={
735 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithencryptedkey.crt'),
736 'client_certificate_password': 'foobar',
739 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
740 def test_certificate_nocombined_pass(self
, handler
):
741 self
._run
_test
(handler
, client_cert
={
742 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
743 'client_certificate_key': os
.path
.join(self
.certdir
, 'clientencrypted.key'),
744 'client_certificate_password': 'foobar',
748 class TestUrllibRequestHandler(TestRequestHandlerBase
):
749 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
750 def test_file_urls(self
, handler
):
751 # See https://github.com/ytdl-org/youtube-dl/issues/8227
752 tf
= tempfile
.NamedTemporaryFile(delete
=False)
755 req
= Request(pathlib
.Path(tf
.name
).as_uri())
756 with handler() as rh
:
757 with pytest
.raises(UnsupportedRequest
):
760 # Test that urllib never loaded FileHandler
761 with pytest
.raises(TransportError
):
764 with handler(enable_file_urls
=True) as rh
:
765 res
= validate_and_send(rh
, req
)
766 assert res
.read() == b
'foobar'
771 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
772 def test_http_error_returns_content(self
, handler
):
773 # urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
775 with handler() as rh
:
778 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_404'))
779 except HTTPError
as e
:
782 assert get_response().read() == b
'<html></html>'
784 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
785 def test_verify_cert_error_text(self
, handler
):
786 # Check the output of the error message
787 with handler() as rh
:
789 CertificateVerifyError
,
790 match
=r
'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate'
792 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
794 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
795 @pytest.mark.parametrize('req,match,version_check', [
796 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
797 # bpo-39603: Check implemented in 3.7.9+, 3.8.5+
799 Request('http://127.0.0.1', method
='GET\n'),
800 'method can\'t contain control characters',
801 lambda v
: v
< (3, 7, 9) or (3, 8, 0) <= v
< (3, 8, 5)
803 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
804 # bpo-38576: Check implemented in 3.7.8+, 3.8.3+
806 Request('http://127.0.0. 1', method
='GET'),
807 'URL can\'t contain control characters',
808 lambda v
: v
< (3, 7, 8) or (3, 8, 0) <= v
< (3, 8, 3)
810 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
811 (Request('http://127.0.0.1', headers
={'foo\n': 'bar'}
), 'Invalid header name', None),
813 def test_httplib_validation_errors(self
, handler
, req
, match
, version_check
):
814 if version_check
and version_check(sys
.version_info
):
815 pytest
.skip(f
'Python {sys.version} version does not have the required validation for this test.')
817 with handler() as rh
:
818 with pytest
.raises(RequestError
, match
=match
) as exc_info
:
819 validate_and_send(rh
, req
)
820 assert not isinstance(exc_info
.value
, TransportError
)
823 class TestRequestsRequestHandler(TestRequestHandlerBase
):
824 @pytest.mark.parametrize('raised,expected', [
825 (lambda: requests
.exceptions
.ConnectTimeout(), TransportError
),
826 (lambda: requests
.exceptions
.ReadTimeout(), TransportError
),
827 (lambda: requests
.exceptions
.Timeout(), TransportError
),
828 (lambda: requests
.exceptions
.ConnectionError(), TransportError
),
829 (lambda: requests
.exceptions
.ProxyError(), ProxyError
),
830 (lambda: requests
.exceptions
.SSLError('12[CERTIFICATE_VERIFY_FAILED]34'), CertificateVerifyError
),
831 (lambda: requests
.exceptions
.SSLError(), SSLError
),
832 (lambda: requests
.exceptions
.InvalidURL(), RequestError
),
833 (lambda: requests
.exceptions
.InvalidHeader(), RequestError
),
834 # catch-all: https://github.com/psf/requests/blob/main/src/requests/adapters.py#L535
835 (lambda: urllib3
.exceptions
.HTTPError(), TransportError
),
836 (lambda: requests
.exceptions
.RequestException(), RequestError
)
837 # (lambda: requests.exceptions.TooManyRedirects(), HTTPError) - Needs a response object
839 @pytest.mark.parametrize('handler', ['Requests'], indirect
=True)
840 def test_request_error_mapping(self
, handler
, monkeypatch
, raised
, expected
):
841 with handler() as rh
:
842 def mock_get_instance(*args
, **kwargs
):
844 def request(self
, *args
, **kwargs
):
848 monkeypatch
.setattr(rh
, '_get_instance', mock_get_instance
)
850 with pytest
.raises(expected
) as exc_info
:
851 rh
.send(Request('http://fake'))
853 assert exc_info
.type is expected
855 @pytest.mark.parametrize('raised,expected,match', [
856 (lambda: urllib3
.exceptions
.SSLError(), SSLError
, None),
857 (lambda: urllib3
.exceptions
.TimeoutError(), TransportError
, None),
858 (lambda: urllib3
.exceptions
.ReadTimeoutError(None, None, None), TransportError
, None),
859 (lambda: urllib3
.exceptions
.ProtocolError(), TransportError
, None),
860 (lambda: urllib3
.exceptions
.DecodeError(), TransportError
, None),
861 (lambda: urllib3
.exceptions
.HTTPError(), TransportError
, None), # catch-all
863 lambda: urllib3
.exceptions
.ProtocolError('error', http
.client
.IncompleteRead(partial
=b
'abc', expected
=4)),
865 '3 bytes read, 4 more expected'
868 lambda: urllib3
.exceptions
.ProtocolError('error', urllib3
.exceptions
.IncompleteRead(partial
=3, expected
=5)),
870 '3 bytes read, 5 more expected'
873 @pytest.mark.parametrize('handler', ['Requests'], indirect
=True)
874 def test_response_error_mapping(self
, handler
, monkeypatch
, raised
, expected
, match
):
875 from urllib3
.response
import HTTPResponse
as Urllib3Response
876 from requests
.models
import Response
as RequestsResponse
877 from yt_dlp
.networking
._requests
import RequestsResponseAdapter
878 requests_res
= RequestsResponse()
879 requests_res
.raw
= Urllib3Response(body
=b
'', status
=200)
880 res
= RequestsResponseAdapter(requests_res
)
882 def mock_read(*args
, **kwargs
):
884 monkeypatch
.setattr(res
.fp
, 'read', mock_read
)
886 with pytest
.raises(expected
, match
=match
) as exc_info
:
889 assert exc_info
.type is expected
892 def run_validation(handler
, error
, req
, **handler_kwargs
):
893 with handler(**handler_kwargs
) as rh
:
895 with pytest
.raises(error
):
901 class TestRequestHandlerValidation
:
903 class ValidationRH(RequestHandler
):
904 def _send(self
, request
):
905 raise RequestError('test')
907 class NoCheckRH(ValidationRH
):
908 _SUPPORTED_FEATURES
= None
909 _SUPPORTED_PROXY_SCHEMES
= None
910 _SUPPORTED_URL_SCHEMES
= None
912 def _check_extensions(self
, extensions
):
915 class HTTPSupportedRH(ValidationRH
):
916 _SUPPORTED_URL_SCHEMES
= ('http',)
919 # scheme, expected to fail, handler kwargs
922 ('https', False, {}),
925 ('file', UnsupportedRequest
, {}),
926 ('file', False, {'enable_file_urls': True}
),
930 ('https', False, {}),
932 (NoCheckRH
, [('http', False, {})]),
933 (ValidationRH
, [('http', UnsupportedRequest
, {})])
936 PROXY_SCHEME_TESTS
= [
937 # scheme, expected to fail
940 ('https', UnsupportedRequest
),
945 ('socks', UnsupportedRequest
),
955 (NoCheckRH
, [('http', False)]),
956 (HTTPSupportedRH
, [('http', UnsupportedRequest
)]),
960 # key, expected to fail
963 ('unrelated', False),
967 ('unrelated', False),
969 (NoCheckRH
, [('all', False)]),
970 (HTTPSupportedRH
, [('all', UnsupportedRequest
)]),
971 (HTTPSupportedRH
, [('no', UnsupportedRequest
)]),
976 ({'cookiejar': 'notacookiejar'}
, AssertionError),
977 ({'cookiejar': YoutubeDLCookieJar()}
, False),
978 ({'cookiejar': CookieJar()}
, AssertionError),
979 ({'timeout': 1}
, False),
980 ({'timeout': 'notatimeout'}
, AssertionError),
981 ({'unsupported': 'value'}
, UnsupportedRequest
),
984 ({'cookiejar': 'notacookiejar'}
, AssertionError),
985 ({'cookiejar': YoutubeDLCookieJar()}
, False),
986 ({'timeout': 1}
, False),
987 ({'timeout': 'notatimeout'}
, AssertionError),
988 ({'unsupported': 'value'}
, UnsupportedRequest
),
991 ({'cookiejar': 'notacookiejar'}
, False),
992 ({'somerandom': 'test'}
, False), # but any extension is allowed through
996 @pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
997 (handler_tests
[0], scheme
, fail
, handler_kwargs
)
998 for handler_tests
in URL_SCHEME_TESTS
999 for scheme
, fail
, handler_kwargs
in handler_tests
[1]
1001 ], indirect
=['handler'])
1002 def test_url_scheme(self
, handler
, scheme
, fail
, handler_kwargs
):
1003 run_validation(handler
, fail
, Request(f
'{scheme}://'), **(handler_kwargs
or {}))
1005 @pytest.mark.parametrize('handler,fail', [('Urllib', False), ('Requests', False)], indirect
=['handler'])
1006 def test_no_proxy(self
, handler
, fail
):
1007 run_validation(handler
, fail
, Request('http://', proxies
={'no': '127.0.0.1,github.com'}
))
1008 run_validation(handler
, fail
, Request('http://'), proxies
={'no': '127.0.0.1,github.com'}
)
1010 @pytest.mark.parametrize('handler,proxy_key,fail', [
1011 (handler_tests
[0], proxy_key
, fail
)
1012 for handler_tests
in PROXY_KEY_TESTS
1013 for proxy_key
, fail
in handler_tests
[1]
1014 ], indirect
=['handler'])
1015 def test_proxy_key(self
, handler
, proxy_key
, fail
):
1016 run_validation(handler
, fail
, Request('http://', proxies
={proxy_key: 'http://example.com'}
))
1017 run_validation(handler
, fail
, Request('http://'), proxies
={proxy_key: 'http://example.com'}
)
1019 @pytest.mark.parametrize('handler,scheme,fail', [
1020 (handler_tests
[0], scheme
, fail
)
1021 for handler_tests
in PROXY_SCHEME_TESTS
1022 for scheme
, fail
in handler_tests
[1]
1023 ], indirect
=['handler'])
1024 def test_proxy_scheme(self
, handler
, scheme
, fail
):
1025 run_validation(handler
, fail
, Request('http://', proxies
={'http': f'{scheme}
://example
.com
'}))
1026 run_validation(handler, fail, Request('http
://'), proxies={'http': f'{scheme}://example.com'})
1028 @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH
, 'Requests'], indirect
=True)
1029 def test_empty_proxy(self
, handler
):
1030 run_validation(handler
, False, Request('http://', proxies
={'http': None}
))
1031 run_validation(handler
, False, Request('http://'), proxies
={'http': None}
)
1033 @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
1034 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect
=True)
1035 def test_invalid_proxy_url(self
, handler
, proxy_url
):
1036 run_validation(handler
, UnsupportedRequest
, Request('http://', proxies
={'http': proxy_url}
))
1038 @pytest.mark.parametrize('handler,extensions,fail', [
1039 (handler_tests
[0], extensions
, fail
)
1040 for handler_tests
in EXTENSION_TESTS
1041 for extensions
, fail
in handler_tests
[1]
1042 ], indirect
=['handler'])
1043 def test_extension(self
, handler
, extensions
, fail
):
1045 handler
, fail
, Request('http://', extensions
=extensions
))
1047 def test_invalid_request_type(self
):
1048 rh
= self
.ValidationRH(logger
=FakeLogger())
1049 for method
in (rh
.validate
, rh
.send
):
1050 with pytest
.raises(TypeError, match
='Expected an instance of Request'):
1051 method('not a request')
1054 class FakeResponse(Response
):
1055 def __init__(self
, request
):
1056 # XXX: we could make request part of standard response interface
1057 self
.request
= request
1058 super().__init
__(fp
=io
.BytesIO(b
''), headers
={}, url
=request
.url
)
1061 class FakeRH(RequestHandler
):
1063 def _validate(self
, request
):
1066 def _send(self
, request
: Request
):
1067 if request
.url
.startswith('ssl://'):
1068 raise SSLError(request
.url
[len('ssl://'):])
1069 return FakeResponse(request
)
1072 class FakeRHYDL(FakeYDL
):
1073 def __init__(self
, *args
, **kwargs
):
1074 super().__init
__(*args
, **kwargs
)
1075 self
._request
_director
= self
.build_request_director([FakeRH
])
1078 class TestRequestDirector
:
1080 def test_handler_operations(self
):
1081 director
= RequestDirector(logger
=FakeLogger())
1082 handler
= FakeRH(logger
=FakeLogger())
1083 director
.add_handler(handler
)
1084 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler
1086 # Handler should overwrite
1087 handler2
= FakeRH(logger
=FakeLogger())
1088 director
.add_handler(handler2
)
1089 assert director
.handlers
.get(FakeRH
.RH_KEY
) is not handler
1090 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler2
1091 assert len(director
.handlers
) == 1
1093 class AnotherFakeRH(FakeRH
):
1095 director
.add_handler(AnotherFakeRH(logger
=FakeLogger()))
1096 assert len(director
.handlers
) == 2
1097 assert director
.handlers
.get(AnotherFakeRH
.RH_KEY
).RH_KEY
== AnotherFakeRH
.RH_KEY
1099 director
.handlers
.pop(FakeRH
.RH_KEY
, None)
1100 assert director
.handlers
.get(FakeRH
.RH_KEY
) is None
1101 assert len(director
.handlers
) == 1
1103 # RequestErrors should passthrough
1104 with pytest
.raises(SSLError
):
1105 director
.send(Request('ssl://something'))
1107 def test_send(self
):
1108 director
= RequestDirector(logger
=FakeLogger())
1109 with pytest
.raises(RequestError
):
1110 director
.send(Request('any://'))
1111 director
.add_handler(FakeRH(logger
=FakeLogger()))
1112 assert isinstance(director
.send(Request('http://')), FakeResponse
)
1114 def test_unsupported_handlers(self
):
1115 class SupportedRH(RequestHandler
):
1116 _SUPPORTED_URL_SCHEMES
= ['http']
1118 def _send(self
, request
: Request
):
1119 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
1121 director
= RequestDirector(logger
=FakeLogger())
1122 director
.add_handler(SupportedRH(logger
=FakeLogger()))
1123 director
.add_handler(FakeRH(logger
=FakeLogger()))
1125 # First should take preference
1126 assert director
.send(Request('http://')).read() == b
'supported'
1127 assert director
.send(Request('any://')).read() == b
''
1129 director
.handlers
.pop(FakeRH
.RH_KEY
)
1130 with pytest
.raises(NoSupportingHandlers
):
1131 director
.send(Request('any://'))
1133 def test_unexpected_error(self
):
1134 director
= RequestDirector(logger
=FakeLogger())
1136 class UnexpectedRH(FakeRH
):
1137 def _send(self
, request
: Request
):
1138 raise TypeError('something')
1140 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1141 with pytest
.raises(NoSupportingHandlers
, match
=r
'1 unexpected error'):
1142 director
.send(Request('any://'))
1144 director
.handlers
.clear()
1145 assert len(director
.handlers
) == 0
1147 # Should not be fatal
1148 director
.add_handler(FakeRH(logger
=FakeLogger()))
1149 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1150 assert director
.send(Request('any://'))
1152 def test_preference(self
):
1153 director
= RequestDirector(logger
=FakeLogger())
1154 director
.add_handler(FakeRH(logger
=FakeLogger()))
1156 class SomeRH(RequestHandler
):
1157 _SUPPORTED_URL_SCHEMES
= ['http']
1159 def _send(self
, request
: Request
):
1160 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
1162 def some_preference(rh
, request
):
1163 return (0 if not isinstance(rh
, SomeRH
)
1164 else 100 if 'prefer' in request
.headers
1167 director
.add_handler(SomeRH(logger
=FakeLogger()))
1168 director
.preferences
.add(some_preference
)
1170 assert director
.send(Request('http://')).read() == b
''
1171 assert director
.send(Request('http://', headers
={'prefer': '1'}
)).read() == b
'supported'
1174 # XXX: do we want to move this to test_YoutubeDL.py?
1175 class TestYoutubeDLNetworking
:
1178 def build_handler(ydl
, handler
: RequestHandler
= FakeRH
):
1179 return ydl
.build_request_director([handler
]).handlers
.get(handler
.RH_KEY
)
1181 def test_compat_opener(self
):
1182 with FakeYDL() as ydl
:
1183 with warnings
.catch_warnings():
1184 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1185 assert isinstance(ydl
._opener
, urllib
.request
.OpenerDirector
)
1187 @pytest.mark.parametrize('proxy,expected', [
1188 ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}
),
1189 ('', {'all': '__noproxy__'}
),
1190 (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}
) # env, set https
1192 def test_proxy(self
, proxy
, expected
):
1193 old_http_proxy
= os
.environ
.get('HTTP_PROXY')
1195 os
.environ
['HTTP_PROXY'] = 'http://127.0.0.1:8081' # ensure that provided proxies override env
1196 with FakeYDL({'proxy': proxy}
) as ydl
:
1197 assert ydl
.proxies
== expected
1200 os
.environ
['HTTP_PROXY'] = old_http_proxy
1202 def test_compat_request(self
):
1203 with FakeRHYDL() as ydl
:
1204 assert ydl
.urlopen('test://')
1205 urllib_req
= urllib
.request
.Request('http://foo.bar', data
=b
'test', method
='PUT', headers
={'X-Test': '1'}
)
1206 urllib_req
.add_unredirected_header('Cookie', 'bob=bob')
1207 urllib_req
.timeout
= 2
1208 with warnings
.catch_warnings():
1209 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1210 req
= ydl
.urlopen(urllib_req
).request
1211 assert req
.url
== urllib_req
.get_full_url()
1212 assert req
.data
== urllib_req
.data
1213 assert req
.method
== urllib_req
.get_method()
1214 assert 'X-Test' in req
.headers
1215 assert 'Cookie' in req
.headers
1216 assert req
.extensions
.get('timeout') == 2
1218 with pytest
.raises(AssertionError):
1221 def test_extract_basic_auth(self
):
1222 with FakeRHYDL() as ydl
:
1223 res
= ydl
.urlopen(Request('http://user:pass@foo.bar'))
1224 assert res
.request
.headers
['Authorization'] == 'Basic dXNlcjpwYXNz'
1226 def test_sanitize_url(self
):
1227 with FakeRHYDL() as ydl
:
1228 res
= ydl
.urlopen(Request('httpss://foo.bar'))
1229 assert res
.request
.url
== 'https://foo.bar'
1231 def test_file_urls_error(self
):
1232 # use urllib handler
1233 with FakeYDL() as ydl
:
1234 with pytest
.raises(RequestError
, match
=r
'file:// URLs are disabled by default'):
1235 ydl
.urlopen('file://')
1237 def test_legacy_server_connect_error(self
):
1238 with FakeRHYDL() as ydl
:
1239 for error
in ('UNSAFE_LEGACY_RENEGOTIATION_DISABLED', 'SSLV3_ALERT_HANDSHAKE_FAILURE'):
1240 with pytest
.raises(RequestError
, match
=r
'Try using --legacy-server-connect'):
1241 ydl
.urlopen(f
'ssl://{error}')
1243 with pytest
.raises(SSLError
, match
='testerror'):
1244 ydl
.urlopen('ssl://testerror')
1246 @pytest.mark.parametrize('proxy_key,proxy_url,expected', [
1247 ('http', '__noproxy__', None),
1248 ('no', '127.0.0.1,foo.bar', '127.0.0.1,foo.bar'),
1249 ('https', 'example.com', 'http://example.com'),
1250 ('https', '//example.com', 'http://example.com'),
1251 ('https', 'socks5://example.com', 'socks5h://example.com'),
1252 ('http', 'socks://example.com', 'socks4://example.com'),
1253 ('http', 'socks4://example.com', 'socks4://example.com'),
1254 ('unrelated', '/bad/proxy', '/bad/proxy'), # clean_proxies should ignore bad proxies
1256 def test_clean_proxy(self
, proxy_key
, proxy_url
, expected
):
1257 # proxies should be cleaned in urlopen()
1258 with FakeRHYDL() as ydl
:
1259 req
= ydl
.urlopen(Request('test://', proxies
={proxy_key: proxy_url}
)).request
1260 assert req
.proxies
[proxy_key
] == expected
1262 # and should also be cleaned when building the handler
1263 env_key
= f
'{proxy_key.upper()}_PROXY'
1264 old_env_proxy
= os
.environ
.get(env_key
)
1266 os
.environ
[env_key
] = proxy_url
# ensure that provided proxies override env
1267 with FakeYDL() as ydl
:
1268 rh
= self
.build_handler(ydl
)
1269 assert rh
.proxies
[proxy_key
] == expected
1272 os
.environ
[env_key
] = old_env_proxy
1274 def test_clean_proxy_header(self
):
1275 with FakeRHYDL() as ydl
:
1276 req
= ydl
.urlopen(Request('test://', headers
={'ytdl-request-proxy': '//foo.bar'}
)).request
1277 assert 'ytdl-request-proxy' not in req
.headers
1278 assert req
.proxies
== {'all': 'http://foo.bar'}
1280 with FakeYDL({'http_headers': {'ytdl-request-proxy': '//foo.bar'}
}) as ydl
:
1281 rh
= self
.build_handler(ydl
)
1282 assert 'ytdl-request-proxy' not in rh
.headers
1283 assert rh
.proxies
== {'all': 'http://foo.bar'}
1285 def test_clean_header(self
):
1286 with FakeRHYDL() as ydl
:
1287 res
= ydl
.urlopen(Request('test://', headers
={'Youtubedl-no-compression': True}
))
1288 assert 'Youtubedl-no-compression' not in res
.request
.headers
1289 assert res
.request
.headers
.get('Accept-Encoding') == 'identity'
1291 with FakeYDL({'http_headers': {'Youtubedl-no-compression': True}
}) as ydl
:
1292 rh
= self
.build_handler(ydl
)
1293 assert 'Youtubedl-no-compression' not in rh
.headers
1294 assert rh
.headers
.get('Accept-Encoding') == 'identity'
1296 with FakeYDL({'http_headers': {'Ytdl-socks-proxy': 'socks://localhost:1080'}
}) as ydl
:
1297 rh
= self
.build_handler(ydl
)
1298 assert 'Ytdl-socks-proxy' not in rh
.headers
1300 def test_build_handler_params(self
):
1302 'http_headers': {'test': 'testtest'}
,
1303 'socket_timeout': 2,
1304 'proxy': 'http://127.0.0.1:8080',
1305 'source_address': '127.0.0.45',
1306 'debug_printtraffic': True,
1307 'compat_opts': ['no-certifi'],
1308 'nocheckcertificate': True,
1309 'legacyserverconnect': True,
1311 rh
= self
.build_handler(ydl
)
1312 assert rh
.headers
.get('test') == 'testtest'
1313 assert 'Accept' in rh
.headers
# ensure std_headers are still there
1314 assert rh
.timeout
== 2
1315 assert rh
.proxies
.get('all') == 'http://127.0.0.1:8080'
1316 assert rh
.source_address
== '127.0.0.45'
1317 assert rh
.verbose
is True
1318 assert rh
.prefer_system_certs
is True
1319 assert rh
.verify
is False
1320 assert rh
.legacy_ssl_support
is True
1322 @pytest.mark.parametrize('ydl_params', [
1323 {'client_certificate': 'fakecert.crt'}
,
1324 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key'}
,
1325 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1326 {'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1328 def test_client_certificate(self
, ydl_params
):
1329 with FakeYDL(ydl_params
) as ydl
:
1330 rh
= self
.build_handler(ydl
)
1331 assert rh
._client
_cert
== ydl_params
# XXX: Too bound to implementation
1333 def test_urllib_file_urls(self
):
1334 with FakeYDL({'enable_file_urls': False}
) as ydl
:
1335 rh
= self
.build_handler(ydl
, UrllibRH
)
1336 assert rh
.enable_file_urls
is False
1338 with FakeYDL({'enable_file_urls': True}
) as ydl
:
1339 rh
= self
.build_handler(ydl
, UrllibRH
)
1340 assert rh
.enable_file_urls
is True
1342 def test_compat_opt_prefer_urllib(self
):
1343 # This assumes urllib only has a preference when this compat opt is given
1344 with FakeYDL({'compat_opts': ['prefer-legacy-http-handler']}
) as ydl
:
1345 director
= ydl
.build_request_director([UrllibRH
])
1346 assert len(director
.preferences
) == 1
1347 assert director
.preferences
.pop()(UrllibRH
, None)
1352 def test_query(self
):
1353 req
= Request('http://example.com?q=something', query
={'v': 'xyz'}
)
1354 assert req
.url
== 'http://example.com?q=something&v=xyz'
1356 req
.update(query
={'v': '123'}
)
1357 assert req
.url
== 'http://example.com?q=something&v=123'
1358 req
.update(url
='http://example.com', query
={'v': 'xyz'}
)
1359 assert req
.url
== 'http://example.com?v=xyz'
1361 def test_method(self
):
1362 req
= Request('http://example.com')
1363 assert req
.method
== 'GET'
1365 assert req
.method
== 'POST'
1367 assert req
.method
== 'GET'
1370 assert req
.method
== 'PUT'
1372 assert req
.method
== 'PUT'
1373 with pytest
.raises(TypeError):
1376 def test_request_helpers(self
):
1377 assert HEADRequest('http://example.com').method
== 'HEAD'
1378 assert PUTRequest('http://example.com').method
== 'PUT'
1380 def test_headers(self
):
1381 req
= Request('http://example.com', headers
={'tesT': 'test'}
)
1382 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1383 req
.update(headers
={'teSt2': 'test2'}
)
1384 assert req
.headers
== HTTPHeaderDict({'test': 'test', 'test2': 'test2'}
)
1386 req
.headers
= new_headers
= HTTPHeaderDict({'test': 'test'}
)
1387 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1388 assert req
.headers
is new_headers
1390 # test converts dict to case insensitive dict
1391 req
.headers
= new_headers
= {'test2': 'test2'}
1392 assert isinstance(req
.headers
, HTTPHeaderDict
)
1393 assert req
.headers
is not new_headers
1395 with pytest
.raises(TypeError):
1398 def test_data_type(self
):
1399 req
= Request('http://example.com')
1400 assert req
.data
is None
1401 # test bytes is allowed
1403 assert req
.data
== b
'test'
1404 # test iterable of bytes is allowed
1405 i
= [b
'test', b
'test2']
1407 assert req
.data
== i
1409 # test file-like object is allowed
1410 f
= io
.BytesIO(b
'test')
1412 assert req
.data
== f
1414 # common mistake: test str not allowed
1415 with pytest
.raises(TypeError):
1417 assert req
.data
!= 'test'
1419 # common mistake: test dict is not allowed
1420 with pytest
.raises(TypeError):
1421 req
.data
= {'test': 'test'}
1422 assert req
.data
!= {'test': 'test'}
1424 def test_content_length_header(self
):
1425 req
= Request('http://example.com', headers
={'Content-Length': '0'}
, data
=b
'')
1426 assert req
.headers
.get('Content-Length') == '0'
1429 assert 'Content-Length' not in req
.headers
1431 req
= Request('http://example.com', headers
={'Content-Length': '10'}
)
1432 assert 'Content-Length' not in req
.headers
1434 def test_content_type_header(self
):
1435 req
= Request('http://example.com', headers
={'Content-Type': 'test'}
, data
=b
'test')
1436 assert req
.headers
.get('Content-Type') == 'test'
1438 assert req
.headers
.get('Content-Type') == 'test'
1440 assert 'Content-Type' not in req
.headers
1442 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1444 def test_update_req(self
):
1445 req
= Request('http://example.com')
1446 assert req
.data
is None
1447 assert req
.method
== 'GET'
1448 assert 'Content-Type' not in req
.headers
1449 # Test that zero-byte payloads will be sent
1450 req
.update(data
=b
'')
1451 assert req
.data
== b
''
1452 assert req
.method
== 'POST'
1453 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1455 def test_proxies(self
):
1456 req
= Request(url
='http://example.com', proxies
={'http': 'http://127.0.0.1:8080'}
)
1457 assert req
.proxies
== {'http': 'http://127.0.0.1:8080'}
1459 def test_extensions(self
):
1460 req
= Request(url
='http://example.com', extensions
={'timeout': 2}
)
1461 assert req
.extensions
== {'timeout': 2}
1463 def test_copy(self
):
1465 url
='http://example.com',
1466 extensions
={'cookiejar': CookieJar()}
,
1467 headers
={'Accept-Encoding': 'br'}
,
1468 proxies
={'http': 'http://127.0.0.1'}
,
1471 req_copy
= req
.copy()
1472 assert req_copy
is not req
1473 assert req_copy
.url
== req
.url
1474 assert req_copy
.headers
== req
.headers
1475 assert req_copy
.headers
is not req
.headers
1476 assert req_copy
.proxies
== req
.proxies
1477 assert req_copy
.proxies
is not req
.proxies
1479 # Data is not able to be copied
1480 assert req_copy
.data
== req
.data
1481 assert req_copy
.data
is req
.data
1483 # Shallow copy extensions
1484 assert req_copy
.extensions
is not req
.extensions
1485 assert req_copy
.extensions
['cookiejar'] == req
.extensions
['cookiejar']
1487 # Subclasses are copied by default
1488 class AnotherRequest(Request
):
1491 req
= AnotherRequest(url
='http://127.0.0.1')
1492 assert isinstance(req
.copy(), AnotherRequest
)
1495 req
= Request(url
='https://фtest.example.com/ some spaceв?ä=c',)
1496 assert req
.url
== 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c'
1498 assert Request(url
='//example.com').url
== 'http://example.com'
1500 with pytest
.raises(TypeError):
1501 Request(url
='https://').url
= None
1506 @pytest.mark.parametrize('reason,status,expected', [
1507 ('custom', 200, 'custom'),
1508 (None, 404, 'Not Found'), # fallback status
1509 ('', 403, 'Forbidden'),
1512 def test_reason(self
, reason
, status
, expected
):
1513 res
= Response(io
.BytesIO(b
''), url
='test://', headers
={}, status
=status
, reason
=reason
)
1514 assert res
.reason
== expected
1516 def test_headers(self
):
1518 headers
.add_header('Test', 'test')
1519 headers
.add_header('Test', 'test2')
1520 headers
.add_header('content-encoding', 'br')
1521 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1522 assert res
.headers
.get_all('test') == ['test', 'test2']
1523 assert 'Content-Encoding' in res
.headers
1525 def test_get_header(self
):
1527 headers
.add_header('Set-Cookie', 'cookie1')
1528 headers
.add_header('Set-cookie', 'cookie2')
1529 headers
.add_header('Test', 'test')
1530 headers
.add_header('Test', 'test2')
1531 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1532 assert res
.get_header('test') == 'test, test2'
1533 assert res
.get_header('set-Cookie') == 'cookie1'
1534 assert res
.get_header('notexist', 'default') == 'default'
1536 def test_compat(self
):
1537 res
= Response(io
.BytesIO(b
''), url
='test://', status
=404, headers
={'test': 'test'}
)
1538 with warnings
.catch_warnings():
1539 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1540 assert res
.code
== res
.getcode() == res
.status
1541 assert res
.geturl() == res
.url
1542 assert res
.info() is res
.headers
1543 assert res
.getheader('test') == res
.get_header('test')