3 # Allow direct execution
9 sys
.path
.insert(0, os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))))
28 from email
.message
import Message
29 from http
.cookiejar
import CookieJar
31 from test
.helper
import FakeYDL
, http_server_port
32 from yt_dlp
.dependencies
import brotli
33 from yt_dlp
.networking
import (
41 from yt_dlp
.networking
._urllib
import UrllibRH
42 from yt_dlp
.networking
.common
import _REQUEST_HANDLERS
43 from yt_dlp
.networking
.exceptions
import (
44 CertificateVerifyError
,
53 from yt_dlp
.utils
._utils
import _YDLLogger
as FakeLogger
54 from yt_dlp
.utils
.networking
import HTTPHeaderDict
56 TEST_DIR
= os
.path
.dirname(os
.path
.abspath(__file__
))
59 def _build_proxy_handler(name
):
60 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
63 def log_message(self
, format
, *args
):
67 self
.send_response(200)
68 self
.send_header('Content-Type', 'text/plain; charset=utf-8')
70 self
.wfile
.write('{self.proxy_name}: {self.path}'.format(self
=self
).encode())
71 return HTTPTestRequestHandler
74 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
75 protocol_version
= 'HTTP/1.1'
77 def log_message(self
, format
, *args
):
81 payload
= str(self
.headers
).encode()
82 self
.send_response(200)
83 self
.send_header('Content-Type', 'application/json')
84 self
.send_header('Content-Length', str(len(payload
)))
86 self
.wfile
.write(payload
)
89 self
.send_response(int(self
.path
[len('/redirect_'):]))
90 self
.send_header('Location', '/method')
91 self
.send_header('Content-Length', '0')
94 def _method(self
, method
, payload
=None):
95 self
.send_response(200)
96 self
.send_header('Content-Length', str(len(payload
or '')))
97 self
.send_header('Method', method
)
100 self
.wfile
.write(payload
)
102 def _status(self
, status
):
103 payload
= f
'<html>{status} NOT FOUND</html>'.encode()
104 self
.send_response(int(status
))
105 self
.send_header('Content-Type', 'text/html; charset=utf-8')
106 self
.send_header('Content-Length', str(len(payload
)))
108 self
.wfile
.write(payload
)
110 def _read_data(self
):
111 if 'Content-Length' in self
.headers
:
112 return self
.rfile
.read(int(self
.headers
['Content-Length']))
115 data
= self
._read
_data
() + str(self
.headers
).encode()
116 if self
.path
.startswith('/redirect_'):
118 elif self
.path
.startswith('/method'):
119 self
._method
('POST', data
)
120 elif self
.path
.startswith('/headers'):
126 if self
.path
.startswith('/redirect_'):
128 elif self
.path
.startswith('/method'):
134 data
= self
._read
_data
() + str(self
.headers
).encode()
135 if self
.path
.startswith('/redirect_'):
137 elif self
.path
.startswith('/method'):
138 self
._method
('PUT', data
)
143 if self
.path
== '/video.html':
144 payload
= b
'<html><video src="/vid.mp4" /></html>'
145 self
.send_response(200)
146 self
.send_header('Content-Type', 'text/html; charset=utf-8')
147 self
.send_header('Content-Length', str(len(payload
)))
149 self
.wfile
.write(payload
)
150 elif self
.path
== '/vid.mp4':
151 payload
= b
'\x00\x00\x00\x00\x20\x66\x74[video]'
152 self
.send_response(200)
153 self
.send_header('Content-Type', 'video/mp4')
154 self
.send_header('Content-Length', str(len(payload
)))
156 self
.wfile
.write(payload
)
157 elif self
.path
== '/%E4%B8%AD%E6%96%87.html':
158 payload
= b
'<html><video src="/vid.mp4" /></html>'
159 self
.send_response(200)
160 self
.send_header('Content-Type', 'text/html; charset=utf-8')
161 self
.send_header('Content-Length', str(len(payload
)))
163 self
.wfile
.write(payload
)
164 elif self
.path
== '/%c7%9f':
165 payload
= b
'<html><video src="/vid.mp4" /></html>'
166 self
.send_response(200)
167 self
.send_header('Content-Type', 'text/html; charset=utf-8')
168 self
.send_header('Content-Length', str(len(payload
)))
170 self
.wfile
.write(payload
)
171 elif self
.path
.startswith('/redirect_loop'):
172 self
.send_response(301)
173 self
.send_header('Location', self
.path
)
174 self
.send_header('Content-Length', '0')
176 elif self
.path
.startswith('/redirect_'):
178 elif self
.path
.startswith('/method'):
179 self
._method
('GET', str(self
.headers
).encode())
180 elif self
.path
.startswith('/headers'):
182 elif self
.path
.startswith('/308-to-headers'):
183 self
.send_response(308)
184 self
.send_header('Location', '/headers')
185 self
.send_header('Content-Length', '0')
187 elif self
.path
== '/trailing_garbage':
188 payload
= b
'<html><video src="/vid.mp4" /></html>'
189 self
.send_response(200)
190 self
.send_header('Content-Type', 'text/html; charset=utf-8')
191 self
.send_header('Content-Encoding', 'gzip')
193 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
195 compressed
= buf
.getvalue() + b
'trailing garbage'
196 self
.send_header('Content-Length', str(len(compressed
)))
198 self
.wfile
.write(compressed
)
199 elif self
.path
== '/302-non-ascii-redirect':
200 new_url
= f
'http://127.0.0.1:{http_server_port(self.server)}/中文.html'
201 self
.send_response(301)
202 self
.send_header('Location', new_url
)
203 self
.send_header('Content-Length', '0')
205 elif self
.path
== '/content-encoding':
206 encodings
= self
.headers
.get('ytdl-encoding', '')
207 payload
= b
'<html><video src="/vid.mp4" /></html>'
208 for encoding
in filter(None, (e
.strip() for e
in encodings
.split(','))):
209 if encoding
== 'br' and brotli
:
210 payload
= brotli
.compress(payload
)
211 elif encoding
== 'gzip':
213 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
215 payload
= buf
.getvalue()
216 elif encoding
== 'deflate':
217 payload
= zlib
.compress(payload
)
218 elif encoding
== 'unsupported':
224 self
.send_response(200)
225 self
.send_header('Content-Encoding', encodings
)
226 self
.send_header('Content-Length', str(len(payload
)))
228 self
.wfile
.write(payload
)
229 elif self
.path
.startswith('/gen_'):
230 payload
= b
'<html></html>'
231 self
.send_response(int(self
.path
[len('/gen_'):]))
232 self
.send_header('Content-Type', 'text/html; charset=utf-8')
233 self
.send_header('Content-Length', str(len(payload
)))
235 self
.wfile
.write(payload
)
236 elif self
.path
.startswith('/incompleteread'):
237 payload
= b
'<html></html>'
238 self
.send_response(200)
239 self
.send_header('Content-Type', 'text/html; charset=utf-8')
240 self
.send_header('Content-Length', '234234')
242 self
.wfile
.write(payload
)
244 elif self
.path
.startswith('/timeout_'):
245 time
.sleep(int(self
.path
[len('/timeout_'):]))
247 elif self
.path
== '/source_address':
248 payload
= str(self
.client_address
[0]).encode()
249 self
.send_response(200)
250 self
.send_header('Content-Type', 'text/html; charset=utf-8')
251 self
.send_header('Content-Length', str(len(payload
)))
253 self
.wfile
.write(payload
)
258 def send_header(self
, keyword
, value
):
260 Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
261 This is against what is defined in RFC 3986, however we need to test we support this
262 since some sites incorrectly do this.
264 if keyword
.lower() == 'connection':
265 return super().send_header(keyword
, value
)
267 if not hasattr(self
, '_headers_buffer'):
268 self
._headers
_buffer
= []
270 self
._headers
_buffer
.append(f
'{keyword}: {value}\r\n'.encode())
273 def validate_and_send(rh
, req
):
278 class TestRequestHandlerBase
:
280 def setup_class(cls
):
281 cls
.http_httpd
= http
.server
.ThreadingHTTPServer(
282 ('127.0.0.1', 0), HTTPTestRequestHandler
)
283 cls
.http_port
= http_server_port(cls
.http_httpd
)
284 cls
.http_server_thread
= threading
.Thread(target
=cls
.http_httpd
.serve_forever
)
285 # FIXME: we should probably stop the http server thread after each test
286 # See: https://github.com/yt-dlp/yt-dlp/pull/7094#discussion_r1199746041
287 cls
.http_server_thread
.daemon
= True
288 cls
.http_server_thread
.start()
291 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
292 cls
.https_httpd
= http
.server
.ThreadingHTTPServer(
293 ('127.0.0.1', 0), HTTPTestRequestHandler
)
294 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
295 sslctx
.load_cert_chain(certfn
, None)
296 cls
.https_httpd
.socket
= sslctx
.wrap_socket(cls
.https_httpd
.socket
, server_side
=True)
297 cls
.https_port
= http_server_port(cls
.https_httpd
)
298 cls
.https_server_thread
= threading
.Thread(target
=cls
.https_httpd
.serve_forever
)
299 cls
.https_server_thread
.daemon
= True
300 cls
.https_server_thread
.start()
304 def handler(request
):
305 RH_KEY
= request
.param
306 if inspect
.isclass(RH_KEY
) and issubclass(RH_KEY
, RequestHandler
):
308 elif RH_KEY
in _REQUEST_HANDLERS
:
309 handler
= _REQUEST_HANDLERS
[RH_KEY
]
311 pytest
.skip(f
'{RH_KEY} request handler is not available')
313 return functools
.partial(handler
, logger
=FakeLogger
)
316 class TestHTTPRequestHandler(TestRequestHandlerBase
):
317 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
318 def test_verify_cert(self
, handler
):
319 with handler() as rh
:
320 with pytest
.raises(CertificateVerifyError
):
321 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
323 with handler(verify
=False) as rh
:
324 r
= validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
325 assert r
.status
== 200
328 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
329 def test_ssl_error(self
, handler
):
330 # HTTPS server with too old TLS version
331 # XXX: is there a better way to test this than to create a new server?
332 https_httpd
= http
.server
.ThreadingHTTPServer(
333 ('127.0.0.1', 0), HTTPTestRequestHandler
)
334 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
335 https_httpd
.socket
= sslctx
.wrap_socket(https_httpd
.socket
, server_side
=True)
336 https_port
= http_server_port(https_httpd
)
337 https_server_thread
= threading
.Thread(target
=https_httpd
.serve_forever
)
338 https_server_thread
.daemon
= True
339 https_server_thread
.start()
341 with handler(verify
=False) as rh
:
342 with pytest
.raises(SSLError
, match
='sslv3 alert handshake failure') as exc_info
:
343 validate_and_send(rh
, Request(f
'https://127.0.0.1:{https_port}/headers'))
344 assert not issubclass(exc_info
.type, CertificateVerifyError
)
346 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
347 def test_percent_encode(self
, handler
):
348 with handler() as rh
:
349 # Unicode characters should be encoded with uppercase percent-encoding
350 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/中文.html'))
351 assert res
.status
== 200
353 # don't normalize existing percent encodings
354 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/%c7%9f'))
355 assert res
.status
== 200
358 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
359 def test_unicode_path_redirection(self
, handler
):
360 with handler() as rh
:
361 r
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
362 assert r
.url
== f
'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
365 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
366 def test_raise_http_error(self
, handler
):
367 with handler() as rh
:
368 for bad_status
in (400, 500, 599, 302):
369 with pytest
.raises(HTTPError
):
370 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_%d' % (self
.http_port
, bad_status
)))
372 # Should not raise an error
373 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_200' % self
.http_port
)).close()
375 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
376 def test_response_url(self
, handler
):
377 with handler() as rh
:
378 # Response url should be that of the last url in redirect chain
379 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_301'))
380 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/method'
382 res2
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_200'))
383 assert res2
.url
== f
'http://127.0.0.1:{self.http_port}/gen_200'
386 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
387 def test_redirect(self
, handler
):
388 with handler() as rh
:
389 def do_req(redirect_status
, method
, assert_no_content
=False):
390 data
= b
'testdata' if method
in ('POST', 'PUT') else None
391 res
= validate_and_send(
392 rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method
=method
, data
=data
))
397 data_sent
+= res
.read(len(data
))
398 if data_sent
!= data
:
402 headers
+= res
.read()
404 if assert_no_content
or data
is None:
405 assert b
'Content-Type' not in headers
406 assert b
'Content-Length' not in headers
408 assert b
'Content-Type' in headers
409 assert b
'Content-Length' in headers
411 return data_sent
.decode(), res
.headers
.get('method', '')
413 # A 303 must either use GET or HEAD for subsequent request
414 assert do_req(303, 'POST', True) == ('', 'GET')
415 assert do_req(303, 'HEAD') == ('', 'HEAD')
417 assert do_req(303, 'PUT', True) == ('', 'GET')
419 # 301 and 302 turn POST only into a GET
420 assert do_req(301, 'POST', True) == ('', 'GET')
421 assert do_req(301, 'HEAD') == ('', 'HEAD')
422 assert do_req(302, 'POST', True) == ('', 'GET')
423 assert do_req(302, 'HEAD') == ('', 'HEAD')
425 assert do_req(301, 'PUT') == ('testdata', 'PUT')
426 assert do_req(302, 'PUT') == ('testdata', 'PUT')
428 # 307 and 308 should not change method
429 for m
in ('POST', 'PUT'):
430 assert do_req(307, m
) == ('testdata', m
)
431 assert do_req(308, m
) == ('testdata', m
)
433 assert do_req(307, 'HEAD') == ('', 'HEAD')
434 assert do_req(308, 'HEAD') == ('', 'HEAD')
436 # These should not redirect and instead raise an HTTPError
437 for code
in (300, 304, 305, 306):
438 with pytest
.raises(HTTPError
):
441 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
442 def test_request_cookie_header(self
, handler
):
443 # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
444 with handler() as rh
:
445 # Specified Cookie header should be used
446 res
= validate_and_send(
448 f
'http://127.0.0.1:{self.http_port}/headers',
449 headers
={'Cookie': 'test=test'}
)).read().decode()
450 assert 'Cookie: test=test' in res
452 # Specified Cookie header should be removed on any redirect
453 res
= validate_and_send(
455 f
'http://127.0.0.1:{self.http_port}/308-to-headers',
456 headers
={'Cookie': 'test=test'}
)).read().decode()
457 assert 'Cookie: test=test' not in res
459 # Specified Cookie header should override global cookiejar for that request
460 cookiejar
= http
.cookiejar
.CookieJar()
461 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
462 version
=0, name
='test', value
='ytdlp', port
=None, port_specified
=False,
463 domain
='127.0.0.1', domain_specified
=True, domain_initial_dot
=False, path
='/',
464 path_specified
=True, secure
=False, expires
=None, discard
=False, comment
=None,
465 comment_url
=None, rest
={}))
467 with handler(cookiejar
=cookiejar
) as rh
:
468 data
= validate_and_send(
469 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', headers
={'cookie': 'test=test'}
)).read()
470 assert b
'Cookie: test=ytdlp' not in data
471 assert b
'Cookie: test=test' in data
473 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
474 def test_redirect_loop(self
, handler
):
475 with handler() as rh
:
476 with pytest
.raises(HTTPError
, match
='redirect loop'):
477 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_loop'))
479 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
480 def test_incompleteread(self
, handler
):
481 with handler(timeout
=2) as rh
:
482 with pytest
.raises(IncompleteRead
):
483 validate_and_send(rh
, Request('http://127.0.0.1:%d/incompleteread' % self
.http_port
)).read()
485 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
486 def test_cookies(self
, handler
):
487 cookiejar
= http
.cookiejar
.CookieJar()
488 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
489 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
490 False, '/headers', True, False, None, False, None, None, {}))
492 with handler(cookiejar
=cookiejar
) as rh
:
493 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
494 assert b
'Cookie: test=ytdlp' in data
497 with handler() as rh
:
498 data
= validate_and_send(
499 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', extensions
={'cookiejar': cookiejar}
)).read()
500 assert b
'Cookie: test=ytdlp' in data
502 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
503 def test_headers(self
, handler
):
505 with handler(headers
=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'}
)) as rh
:
507 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
508 assert b
'Test1: test' in data
510 # Per request headers, merged with global
511 data
= validate_and_send(rh
, Request(
512 f
'http://127.0.0.1:{self.http_port}/headers', headers
={'test2': 'changed', 'test3': 'test3'}
)).read()
513 assert b
'Test1: test' in data
514 assert b
'Test2: changed' in data
515 assert b
'Test2: test2' not in data
516 assert b
'Test3: test3' in data
518 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
519 def test_timeout(self
, handler
):
520 with handler() as rh
:
521 # Default timeout is 20 seconds, so this should go through
523 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_3'))
525 with handler(timeout
=0.5) as rh
:
526 with pytest
.raises(TransportError
):
528 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1'))
530 # Per request timeout, should override handler timeout
532 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1', extensions
={'timeout': 4}
))
534 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
535 def test_source_address(self
, handler
):
536 source_address
= f
'127.0.0.{random.randint(5, 255)}'
537 with handler(source_address
=source_address
) as rh
:
538 data
= validate_and_send(
539 rh
, Request(f
'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
540 assert source_address
== data
542 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
543 def test_gzip_trailing_garbage(self
, handler
):
544 with handler() as rh
:
545 data
= validate_and_send(rh
, Request(f
'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
546 assert data
== '<html><video src="/vid.mp4" /></html>'
548 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
549 @pytest.mark.skipif(not brotli
, reason
='brotli support is not installed')
550 def test_brotli(self
, handler
):
551 with handler() as rh
:
552 res
= validate_and_send(
554 f
'http://127.0.0.1:{self.http_port}/content-encoding',
555 headers
={'ytdl-encoding': 'br'}
))
556 assert res
.headers
.get('Content-Encoding') == 'br'
557 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
559 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
560 def test_deflate(self
, handler
):
561 with handler() as rh
:
562 res
= validate_and_send(
564 f
'http://127.0.0.1:{self.http_port}/content-encoding',
565 headers
={'ytdl-encoding': 'deflate'}
))
566 assert res
.headers
.get('Content-Encoding') == 'deflate'
567 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
569 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
570 def test_gzip(self
, handler
):
571 with handler() as rh
:
572 res
= validate_and_send(
574 f
'http://127.0.0.1:{self.http_port}/content-encoding',
575 headers
={'ytdl-encoding': 'gzip'}
))
576 assert res
.headers
.get('Content-Encoding') == 'gzip'
577 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
579 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
580 def test_multiple_encodings(self
, handler
):
581 with handler() as rh
:
582 for pair
in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
583 res
= validate_and_send(
585 f
'http://127.0.0.1:{self.http_port}/content-encoding',
586 headers
={'ytdl-encoding': pair}
))
587 assert res
.headers
.get('Content-Encoding') == pair
588 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
590 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
591 def test_unsupported_encoding(self
, handler
):
592 with handler() as rh
:
593 res
= validate_and_send(
595 f
'http://127.0.0.1:{self.http_port}/content-encoding',
596 headers
={'ytdl-encoding': 'unsupported'}
))
597 assert res
.headers
.get('Content-Encoding') == 'unsupported'
598 assert res
.read() == b
'raw'
600 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
601 def test_read(self
, handler
):
602 with handler() as rh
:
603 res
= validate_and_send(
604 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers'))
605 assert res
.readable()
606 assert res
.read(1) == b
'H'
607 assert res
.read(3) == b
'ost'
610 class TestHTTPProxy(TestRequestHandlerBase
):
612 def setup_class(cls
):
613 super().setup_class()
615 cls
.proxy
= http
.server
.ThreadingHTTPServer(
616 ('127.0.0.1', 0), _build_proxy_handler('normal'))
617 cls
.proxy_port
= http_server_port(cls
.proxy
)
618 cls
.proxy_thread
= threading
.Thread(target
=cls
.proxy
.serve_forever
)
619 cls
.proxy_thread
.daemon
= True
620 cls
.proxy_thread
.start()
623 cls
.geo_proxy
= http
.server
.ThreadingHTTPServer(
624 ('127.0.0.1', 0), _build_proxy_handler('geo'))
625 cls
.geo_port
= http_server_port(cls
.geo_proxy
)
626 cls
.geo_proxy_thread
= threading
.Thread(target
=cls
.geo_proxy
.serve_forever
)
627 cls
.geo_proxy_thread
.daemon
= True
628 cls
.geo_proxy_thread
.start()
630 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
631 def test_http_proxy(self
, handler
):
632 http_proxy
= f
'http://127.0.0.1:{self.proxy_port}'
633 geo_proxy
= f
'http://127.0.0.1:{self.geo_port}'
635 # Test global http proxy
636 # Test per request http proxy
637 # Test per request http proxy disables proxy
638 url
= 'http://foo.com/bar'
641 with handler(proxies
={'http': http_proxy}
) as rh
:
642 res
= validate_and_send(rh
, Request(url
)).read().decode()
643 assert res
== f
'normal: {url}'
645 # Per request proxy overrides global
646 res
= validate_and_send(rh
, Request(url
, proxies
={'http': geo_proxy}
)).read().decode()
647 assert res
== f
'geo: {url}'
649 # and setting to None disables all proxies for that request
650 real_url
= f
'http://127.0.0.1:{self.http_port}/headers'
651 res
= validate_and_send(
652 rh
, Request(real_url
, proxies
={'http': None}
)).read().decode()
653 assert res
!= f
'normal: {real_url}'
654 assert 'Accept' in res
656 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
657 def test_noproxy(self
, handler
):
658 with handler(proxies
={'proxy': f'http://127.0.0.1:{self.proxy_port}
'}) as rh:
660 for no_proxy in (f'127.0.0.1:{self.http_port}
', '127.0.0.1', 'localhost
'):
661 nop_response = validate_and_send(
662 rh, Request(f'http
://127.0.0.1:{self.http_port}
/headers
', proxies={'no': no_proxy})).read().decode(
664 assert 'Accept
' in nop_response
666 @pytest.mark.parametrize('handler
', ['Urllib
'], indirect=True)
667 def test_allproxy(self, handler):
668 url = 'http
://foo
.com
/bar
'
669 with handler() as rh:
670 response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
672 assert response
== f
'normal: {url}'
674 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
675 def test_http_proxy_with_idn(self
, handler
):
676 with handler(proxies
={
677 'http': f
'http://127.0.0.1:{self.proxy_port}',
679 url
= 'http://中文.tw/'
680 response
= rh
.send(Request(url
)).read().decode()
681 # b'xn--fiq228c' is '中文'.encode('idna')
682 assert response
== 'normal: http://xn--fiq228c.tw/'
685 class TestClientCertificate
:
688 def setup_class(cls
):
689 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
690 cls
.certdir
= os
.path
.join(TEST_DIR
, 'testdata', 'certificate')
691 cacertfn
= os
.path
.join(cls
.certdir
, 'ca.crt')
692 cls
.httpd
= http
.server
.ThreadingHTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler
)
693 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
694 sslctx
.verify_mode
= ssl
.CERT_REQUIRED
695 sslctx
.load_verify_locations(cafile
=cacertfn
)
696 sslctx
.load_cert_chain(certfn
, None)
697 cls
.httpd
.socket
= sslctx
.wrap_socket(cls
.httpd
.socket
, server_side
=True)
698 cls
.port
= http_server_port(cls
.httpd
)
699 cls
.server_thread
= threading
.Thread(target
=cls
.httpd
.serve_forever
)
700 cls
.server_thread
.daemon
= True
701 cls
.server_thread
.start()
703 def _run_test(self
, handler
, **handler_kwargs
):
705 # Disable client-side validation of unacceptable self-signed testcert.pem
706 # The test is of a check on the server side, so unaffected
710 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.port}/video.html')).read().decode()
712 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
713 def test_certificate_combined_nopass(self
, handler
):
714 self
._run
_test
(handler
, client_cert
={
715 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithkey.crt'),
718 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
719 def test_certificate_nocombined_nopass(self
, handler
):
720 self
._run
_test
(handler
, client_cert
={
721 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
722 'client_certificate_key': os
.path
.join(self
.certdir
, 'client.key'),
725 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
726 def test_certificate_combined_pass(self
, handler
):
727 self
._run
_test
(handler
, client_cert
={
728 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithencryptedkey.crt'),
729 'client_certificate_password': 'foobar',
732 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
733 def test_certificate_nocombined_pass(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
, 'clientencrypted.key'),
737 'client_certificate_password': 'foobar',
741 class TestUrllibRequestHandler(TestRequestHandlerBase
):
742 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
743 def test_file_urls(self
, handler
):
744 # See https://github.com/ytdl-org/youtube-dl/issues/8227
745 tf
= tempfile
.NamedTemporaryFile(delete
=False)
748 req
= Request(pathlib
.Path(tf
.name
).as_uri())
749 with handler() as rh
:
750 with pytest
.raises(UnsupportedRequest
):
753 # Test that urllib never loaded FileHandler
754 with pytest
.raises(TransportError
):
757 with handler(enable_file_urls
=True) as rh
:
758 res
= validate_and_send(rh
, req
)
759 assert res
.read() == b
'foobar'
764 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
765 def test_http_error_returns_content(self
, handler
):
766 # urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
768 with handler() as rh
:
771 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_404'))
772 except HTTPError
as e
:
775 assert get_response().read() == b
'<html></html>'
777 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
778 def test_verify_cert_error_text(self
, handler
):
779 # Check the output of the error message
780 with handler() as rh
:
782 CertificateVerifyError
,
783 match
=r
'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate'
785 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
787 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
788 def test_httplib_validation_errors(self
, handler
):
789 with handler() as rh
:
791 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
792 with pytest
.raises(RequestError
, match
='method can\'t contain control characters') as exc_info
:
793 validate_and_send(rh
, Request('http://127.0.0.1', method
='GET\n'))
794 assert not isinstance(exc_info
.value
, TransportError
)
796 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
797 with pytest
.raises(RequestError
, match
='URL can\'t contain control characters') as exc_info
:
798 validate_and_send(rh
, Request('http://127.0.0. 1', method
='GET\n'))
799 assert not isinstance(exc_info
.value
, TransportError
)
801 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
802 with pytest
.raises(RequestError
, match
='Invalid header name') as exc_info
:
803 validate_and_send(rh
, Request('http://127.0.0.1', headers
={'foo\n': 'bar'}
))
804 assert not isinstance(exc_info
.value
, TransportError
)
807 def run_validation(handler
, fail
, req
, **handler_kwargs
):
808 with handler(**handler_kwargs
) as rh
:
810 with pytest
.raises(UnsupportedRequest
):
816 class TestRequestHandlerValidation
:
818 class ValidationRH(RequestHandler
):
819 def _send(self
, request
):
820 raise RequestError('test')
822 class NoCheckRH(ValidationRH
):
823 _SUPPORTED_FEATURES
= None
824 _SUPPORTED_PROXY_SCHEMES
= None
825 _SUPPORTED_URL_SCHEMES
= None
827 class HTTPSupportedRH(ValidationRH
):
828 _SUPPORTED_URL_SCHEMES
= ('http',)
831 # scheme, expected to fail, handler kwargs
834 ('https', False, {}),
838 ('file', False, {'enable_file_urls': True}
),
840 (NoCheckRH
, [('http', False, {})]),
841 (ValidationRH
, [('http', True, {})])
844 PROXY_SCHEME_TESTS
= [
845 # scheme, expected to fail
855 (NoCheckRH
, [('http', False)]),
856 (HTTPSupportedRH
, [('http', True)]),
860 # key, expected to fail
863 ('unrelated', False),
865 (NoCheckRH
, [('all', False)]),
866 (HTTPSupportedRH
, [('all', True)]),
867 (HTTPSupportedRH
, [('no', True)]),
870 @pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
871 (handler_tests
[0], scheme
, fail
, handler_kwargs
)
872 for handler_tests
in URL_SCHEME_TESTS
873 for scheme
, fail
, handler_kwargs
in handler_tests
[1]
875 ], indirect
=['handler'])
876 def test_url_scheme(self
, handler
, scheme
, fail
, handler_kwargs
):
877 run_validation(handler
, fail
, Request(f
'{scheme}://'), **(handler_kwargs
or {}))
879 @pytest.mark.parametrize('handler,fail', [('Urllib', False)], indirect
=['handler'])
880 def test_no_proxy(self
, handler
, fail
):
881 run_validation(handler
, fail
, Request('http://', proxies
={'no': '127.0.0.1,github.com'}
))
882 run_validation(handler
, fail
, Request('http://'), proxies
={'no': '127.0.0.1,github.com'}
)
884 @pytest.mark.parametrize('handler,proxy_key,fail', [
885 (handler_tests
[0], proxy_key
, fail
)
886 for handler_tests
in PROXY_KEY_TESTS
887 for proxy_key
, fail
in handler_tests
[1]
888 ], indirect
=['handler'])
889 def test_proxy_key(self
, handler
, proxy_key
, fail
):
890 run_validation(handler
, fail
, Request('http://', proxies
={proxy_key: 'http://example.com'}
))
891 run_validation(handler
, fail
, Request('http://'), proxies
={proxy_key: 'http://example.com'}
)
893 @pytest.mark.parametrize('handler,scheme,fail', [
894 (handler_tests
[0], scheme
, fail
)
895 for handler_tests
in PROXY_SCHEME_TESTS
896 for scheme
, fail
in handler_tests
[1]
897 ], indirect
=['handler'])
898 def test_proxy_scheme(self
, handler
, scheme
, fail
):
899 run_validation(handler
, fail
, Request('http://', proxies
={'http': f'{scheme}
://example
.com
'}))
900 run_validation(handler, fail, Request('http
://'), proxies={'http': f'{scheme}://example.com'})
902 @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH
], indirect
=True)
903 def test_empty_proxy(self
, handler
):
904 run_validation(handler
, False, Request('http://', proxies
={'http': None}
))
905 run_validation(handler
, False, Request('http://'), proxies
={'http': None}
)
907 @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1'])
908 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
909 def test_missing_proxy_scheme(self
, handler
, proxy_url
):
910 run_validation(handler
, True, Request('http://', proxies
={'http': 'example.com'}
))
912 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
913 def test_cookiejar_extension(self
, handler
):
914 run_validation(handler
, True, Request('http://', extensions
={'cookiejar': 'notacookiejar'}
))
916 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
917 def test_timeout_extension(self
, handler
):
918 run_validation(handler
, True, Request('http://', extensions
={'timeout': 'notavalidtimeout'}
))
920 def test_invalid_request_type(self
):
921 rh
= self
.ValidationRH(logger
=FakeLogger())
922 for method
in (rh
.validate
, rh
.send
):
923 with pytest
.raises(TypeError, match
='Expected an instance of Request'):
924 method('not a request')
927 class FakeResponse(Response
):
928 def __init__(self
, request
):
929 # XXX: we could make request part of standard response interface
930 self
.request
= request
931 super().__init
__(fp
=io
.BytesIO(b
''), headers
={}, url
=request
.url
)
934 class FakeRH(RequestHandler
):
936 def _validate(self
, request
):
939 def _send(self
, request
: Request
):
940 if request
.url
.startswith('ssl://'):
941 raise SSLError(request
.url
[len('ssl://'):])
942 return FakeResponse(request
)
945 class FakeRHYDL(FakeYDL
):
946 def __init__(self
, *args
, **kwargs
):
947 super().__init
__(*args
, **kwargs
)
948 self
._request
_director
= self
.build_request_director([FakeRH
])
951 class TestRequestDirector
:
953 def test_handler_operations(self
):
954 director
= RequestDirector(logger
=FakeLogger())
955 handler
= FakeRH(logger
=FakeLogger())
956 director
.add_handler(handler
)
957 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler
959 # Handler should overwrite
960 handler2
= FakeRH(logger
=FakeLogger())
961 director
.add_handler(handler2
)
962 assert director
.handlers
.get(FakeRH
.RH_KEY
) is not handler
963 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler2
964 assert len(director
.handlers
) == 1
966 class AnotherFakeRH(FakeRH
):
968 director
.add_handler(AnotherFakeRH(logger
=FakeLogger()))
969 assert len(director
.handlers
) == 2
970 assert director
.handlers
.get(AnotherFakeRH
.RH_KEY
).RH_KEY
== AnotherFakeRH
.RH_KEY
972 director
.handlers
.pop(FakeRH
.RH_KEY
, None)
973 assert director
.handlers
.get(FakeRH
.RH_KEY
) is None
974 assert len(director
.handlers
) == 1
976 # RequestErrors should passthrough
977 with pytest
.raises(SSLError
):
978 director
.send(Request('ssl://something'))
981 director
= RequestDirector(logger
=FakeLogger())
982 with pytest
.raises(RequestError
):
983 director
.send(Request('any://'))
984 director
.add_handler(FakeRH(logger
=FakeLogger()))
985 assert isinstance(director
.send(Request('http://')), FakeResponse
)
987 def test_unsupported_handlers(self
):
988 director
= RequestDirector(logger
=FakeLogger())
989 director
.add_handler(FakeRH(logger
=FakeLogger()))
991 class SupportedRH(RequestHandler
):
992 _SUPPORTED_URL_SCHEMES
= ['http']
994 def _send(self
, request
: Request
):
995 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
997 # This handler should by default take preference over FakeRH
998 director
.add_handler(SupportedRH(logger
=FakeLogger()))
999 assert director
.send(Request('http://')).read() == b
'supported'
1000 assert director
.send(Request('any://')).read() == b
''
1002 director
.handlers
.pop(FakeRH
.RH_KEY
)
1003 with pytest
.raises(NoSupportingHandlers
):
1004 director
.send(Request('any://'))
1006 def test_unexpected_error(self
):
1007 director
= RequestDirector(logger
=FakeLogger())
1009 class UnexpectedRH(FakeRH
):
1010 def _send(self
, request
: Request
):
1011 raise TypeError('something')
1013 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1014 with pytest
.raises(NoSupportingHandlers
, match
=r
'1 unexpected error'):
1015 director
.send(Request('any://'))
1017 director
.handlers
.clear()
1018 assert len(director
.handlers
) == 0
1020 # Should not be fatal
1021 director
.add_handler(FakeRH(logger
=FakeLogger()))
1022 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1023 assert director
.send(Request('any://'))
1026 # XXX: do we want to move this to test_YoutubeDL.py?
1027 class TestYoutubeDLNetworking
:
1030 def build_handler(ydl
, handler
: RequestHandler
= FakeRH
):
1031 return ydl
.build_request_director([handler
]).handlers
.get(handler
.RH_KEY
)
1033 def test_compat_opener(self
):
1034 with FakeYDL() as ydl
:
1035 with warnings
.catch_warnings():
1036 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1037 assert isinstance(ydl
._opener
, urllib
.request
.OpenerDirector
)
1039 @pytest.mark.parametrize('proxy,expected', [
1040 ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}
),
1041 ('', {'all': '__noproxy__'}
),
1042 (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}
) # env, set https
1044 def test_proxy(self
, proxy
, expected
):
1045 old_http_proxy
= os
.environ
.get('HTTP_PROXY')
1047 os
.environ
['HTTP_PROXY'] = 'http://127.0.0.1:8081' # ensure that provided proxies override env
1048 with FakeYDL({'proxy': proxy}
) as ydl
:
1049 assert ydl
.proxies
== expected
1052 os
.environ
['HTTP_PROXY'] = old_http_proxy
1054 def test_compat_request(self
):
1055 with FakeRHYDL() as ydl
:
1056 assert ydl
.urlopen('test://')
1057 urllib_req
= urllib
.request
.Request('http://foo.bar', data
=b
'test', method
='PUT', headers
={'X-Test': '1'}
)
1058 urllib_req
.add_unredirected_header('Cookie', 'bob=bob')
1059 urllib_req
.timeout
= 2
1060 with warnings
.catch_warnings():
1061 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1062 req
= ydl
.urlopen(urllib_req
).request
1063 assert req
.url
== urllib_req
.get_full_url()
1064 assert req
.data
== urllib_req
.data
1065 assert req
.method
== urllib_req
.get_method()
1066 assert 'X-Test' in req
.headers
1067 assert 'Cookie' in req
.headers
1068 assert req
.extensions
.get('timeout') == 2
1070 with pytest
.raises(AssertionError):
1073 def test_extract_basic_auth(self
):
1074 with FakeRHYDL() as ydl
:
1075 res
= ydl
.urlopen(Request('http://user:pass@foo.bar'))
1076 assert res
.request
.headers
['Authorization'] == 'Basic dXNlcjpwYXNz'
1078 def test_sanitize_url(self
):
1079 with FakeRHYDL() as ydl
:
1080 res
= ydl
.urlopen(Request('httpss://foo.bar'))
1081 assert res
.request
.url
== 'https://foo.bar'
1083 def test_file_urls_error(self
):
1084 # use urllib handler
1085 with FakeYDL() as ydl
:
1086 with pytest
.raises(RequestError
, match
=r
'file:// URLs are disabled by default'):
1087 ydl
.urlopen('file://')
1089 def test_legacy_server_connect_error(self
):
1090 with FakeRHYDL() as ydl
:
1091 for error
in ('UNSAFE_LEGACY_RENEGOTIATION_DISABLED', 'SSLV3_ALERT_HANDSHAKE_FAILURE'):
1092 with pytest
.raises(RequestError
, match
=r
'Try using --legacy-server-connect'):
1093 ydl
.urlopen(f
'ssl://{error}')
1095 with pytest
.raises(SSLError
, match
='testerror'):
1096 ydl
.urlopen('ssl://testerror')
1098 @pytest.mark.parametrize('proxy_key,proxy_url,expected', [
1099 ('http', '__noproxy__', None),
1100 ('no', '127.0.0.1,foo.bar', '127.0.0.1,foo.bar'),
1101 ('https', 'example.com', 'http://example.com'),
1102 ('https', 'socks5://example.com', 'socks5h://example.com'),
1103 ('http', 'socks://example.com', 'socks4://example.com'),
1104 ('http', 'socks4://example.com', 'socks4://example.com'),
1106 def test_clean_proxy(self
, proxy_key
, proxy_url
, expected
):
1107 # proxies should be cleaned in urlopen()
1108 with FakeRHYDL() as ydl
:
1109 req
= ydl
.urlopen(Request('test://', proxies
={proxy_key: proxy_url}
)).request
1110 assert req
.proxies
[proxy_key
] == expected
1112 # and should also be cleaned when building the handler
1113 env_key
= f
'{proxy_key.upper()}_PROXY'
1114 old_env_proxy
= os
.environ
.get(env_key
)
1116 os
.environ
[env_key
] = proxy_url
# ensure that provided proxies override env
1117 with FakeYDL() as ydl
:
1118 rh
= self
.build_handler(ydl
)
1119 assert rh
.proxies
[proxy_key
] == expected
1122 os
.environ
[env_key
] = old_env_proxy
1124 def test_clean_proxy_header(self
):
1125 with FakeRHYDL() as ydl
:
1126 req
= ydl
.urlopen(Request('test://', headers
={'ytdl-request-proxy': '//foo.bar'}
)).request
1127 assert 'ytdl-request-proxy' not in req
.headers
1128 assert req
.proxies
== {'all': 'http://foo.bar'}
1130 with FakeYDL({'http_headers': {'ytdl-request-proxy': '//foo.bar'}
}) as ydl
:
1131 rh
= self
.build_handler(ydl
)
1132 assert 'ytdl-request-proxy' not in rh
.headers
1133 assert rh
.proxies
== {'all': 'http://foo.bar'}
1135 def test_clean_header(self
):
1136 with FakeRHYDL() as ydl
:
1137 res
= ydl
.urlopen(Request('test://', headers
={'Youtubedl-no-compression': True}
))
1138 assert 'Youtubedl-no-compression' not in res
.request
.headers
1139 assert res
.request
.headers
.get('Accept-Encoding') == 'identity'
1141 with FakeYDL({'http_headers': {'Youtubedl-no-compression': True}
}) as ydl
:
1142 rh
= self
.build_handler(ydl
)
1143 assert 'Youtubedl-no-compression' not in rh
.headers
1144 assert rh
.headers
.get('Accept-Encoding') == 'identity'
1146 def test_build_handler_params(self
):
1148 'http_headers': {'test': 'testtest'}
,
1149 'socket_timeout': 2,
1150 'proxy': 'http://127.0.0.1:8080',
1151 'source_address': '127.0.0.45',
1152 'debug_printtraffic': True,
1153 'compat_opts': ['no-certifi'],
1154 'nocheckcertificate': True,
1155 'legacyserverconnect': True,
1157 rh
= self
.build_handler(ydl
)
1158 assert rh
.headers
.get('test') == 'testtest'
1159 assert 'Accept' in rh
.headers
# ensure std_headers are still there
1160 assert rh
.timeout
== 2
1161 assert rh
.proxies
.get('all') == 'http://127.0.0.1:8080'
1162 assert rh
.source_address
== '127.0.0.45'
1163 assert rh
.verbose
is True
1164 assert rh
.prefer_system_certs
is True
1165 assert rh
.verify
is False
1166 assert rh
.legacy_ssl_support
is True
1168 @pytest.mark.parametrize('ydl_params', [
1169 {'client_certificate': 'fakecert.crt'}
,
1170 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key'}
,
1171 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1172 {'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1174 def test_client_certificate(self
, ydl_params
):
1175 with FakeYDL(ydl_params
) as ydl
:
1176 rh
= self
.build_handler(ydl
)
1177 assert rh
._client
_cert
== ydl_params
# XXX: Too bound to implementation
1179 def test_urllib_file_urls(self
):
1180 with FakeYDL({'enable_file_urls': False}
) as ydl
:
1181 rh
= self
.build_handler(ydl
, UrllibRH
)
1182 assert rh
.enable_file_urls
is False
1184 with FakeYDL({'enable_file_urls': True}
) as ydl
:
1185 rh
= self
.build_handler(ydl
, UrllibRH
)
1186 assert rh
.enable_file_urls
is True
1191 def test_query(self
):
1192 req
= Request('http://example.com?q=something', query
={'v': 'xyz'}
)
1193 assert req
.url
== 'http://example.com?q=something&v=xyz'
1195 req
.update(query
={'v': '123'}
)
1196 assert req
.url
== 'http://example.com?q=something&v=123'
1197 req
.update(url
='http://example.com', query
={'v': 'xyz'}
)
1198 assert req
.url
== 'http://example.com?v=xyz'
1200 def test_method(self
):
1201 req
= Request('http://example.com')
1202 assert req
.method
== 'GET'
1204 assert req
.method
== 'POST'
1206 assert req
.method
== 'GET'
1209 assert req
.method
== 'PUT'
1211 assert req
.method
== 'PUT'
1212 with pytest
.raises(TypeError):
1215 def test_request_helpers(self
):
1216 assert HEADRequest('http://example.com').method
== 'HEAD'
1217 assert PUTRequest('http://example.com').method
== 'PUT'
1219 def test_headers(self
):
1220 req
= Request('http://example.com', headers
={'tesT': 'test'}
)
1221 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1222 req
.update(headers
={'teSt2': 'test2'}
)
1223 assert req
.headers
== HTTPHeaderDict({'test': 'test', 'test2': 'test2'}
)
1225 req
.headers
= new_headers
= HTTPHeaderDict({'test': 'test'}
)
1226 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1227 assert req
.headers
is new_headers
1229 # test converts dict to case insensitive dict
1230 req
.headers
= new_headers
= {'test2': 'test2'}
1231 assert isinstance(req
.headers
, HTTPHeaderDict
)
1232 assert req
.headers
is not new_headers
1234 with pytest
.raises(TypeError):
1237 def test_data_type(self
):
1238 req
= Request('http://example.com')
1239 assert req
.data
is None
1240 # test bytes is allowed
1242 assert req
.data
== b
'test'
1243 # test iterable of bytes is allowed
1244 i
= [b
'test', b
'test2']
1246 assert req
.data
== i
1248 # test file-like object is allowed
1249 f
= io
.BytesIO(b
'test')
1251 assert req
.data
== f
1253 # common mistake: test str not allowed
1254 with pytest
.raises(TypeError):
1256 assert req
.data
!= 'test'
1258 # common mistake: test dict is not allowed
1259 with pytest
.raises(TypeError):
1260 req
.data
= {'test': 'test'}
1261 assert req
.data
!= {'test': 'test'}
1263 def test_content_length_header(self
):
1264 req
= Request('http://example.com', headers
={'Content-Length': '0'}
, data
=b
'')
1265 assert req
.headers
.get('Content-Length') == '0'
1268 assert 'Content-Length' not in req
.headers
1270 req
= Request('http://example.com', headers
={'Content-Length': '10'}
)
1271 assert 'Content-Length' not in req
.headers
1273 def test_content_type_header(self
):
1274 req
= Request('http://example.com', headers
={'Content-Type': 'test'}
, data
=b
'test')
1275 assert req
.headers
.get('Content-Type') == 'test'
1277 assert req
.headers
.get('Content-Type') == 'test'
1279 assert 'Content-Type' not in req
.headers
1281 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1283 def test_update_req(self
):
1284 req
= Request('http://example.com')
1285 assert req
.data
is None
1286 assert req
.method
== 'GET'
1287 assert 'Content-Type' not in req
.headers
1288 # Test that zero-byte payloads will be sent
1289 req
.update(data
=b
'')
1290 assert req
.data
== b
''
1291 assert req
.method
== 'POST'
1292 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1294 def test_proxies(self
):
1295 req
= Request(url
='http://example.com', proxies
={'http': 'http://127.0.0.1:8080'}
)
1296 assert req
.proxies
== {'http': 'http://127.0.0.1:8080'}
1298 def test_extensions(self
):
1299 req
= Request(url
='http://example.com', extensions
={'timeout': 2}
)
1300 assert req
.extensions
== {'timeout': 2}
1302 def test_copy(self
):
1304 url
='http://example.com',
1305 extensions
={'cookiejar': CookieJar()}
,
1306 headers
={'Accept-Encoding': 'br'}
,
1307 proxies
={'http': 'http://127.0.0.1'}
,
1310 req_copy
= req
.copy()
1311 assert req_copy
is not req
1312 assert req_copy
.url
== req
.url
1313 assert req_copy
.headers
== req
.headers
1314 assert req_copy
.headers
is not req
.headers
1315 assert req_copy
.proxies
== req
.proxies
1316 assert req_copy
.proxies
is not req
.proxies
1318 # Data is not able to be copied
1319 assert req_copy
.data
== req
.data
1320 assert req_copy
.data
is req
.data
1322 # Shallow copy extensions
1323 assert req_copy
.extensions
is not req
.extensions
1324 assert req_copy
.extensions
['cookiejar'] == req
.extensions
['cookiejar']
1326 # Subclasses are copied by default
1327 class AnotherRequest(Request
):
1330 req
= AnotherRequest(url
='http://127.0.0.1')
1331 assert isinstance(req
.copy(), AnotherRequest
)
1334 req
= Request(url
='https://фtest.example.com/ some spaceв?ä=c',)
1335 assert req
.url
== 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c'
1337 assert Request(url
='//example.com').url
== 'http://example.com'
1339 with pytest
.raises(TypeError):
1340 Request(url
='https://').url
= None
1345 @pytest.mark.parametrize('reason,status,expected', [
1346 ('custom', 200, 'custom'),
1347 (None, 404, 'Not Found'), # fallback status
1348 ('', 403, 'Forbidden'),
1351 def test_reason(self
, reason
, status
, expected
):
1352 res
= Response(io
.BytesIO(b
''), url
='test://', headers
={}, status
=status
, reason
=reason
)
1353 assert res
.reason
== expected
1355 def test_headers(self
):
1357 headers
.add_header('Test', 'test')
1358 headers
.add_header('Test', 'test2')
1359 headers
.add_header('content-encoding', 'br')
1360 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1361 assert res
.headers
.get_all('test') == ['test', 'test2']
1362 assert 'Content-Encoding' in res
.headers
1364 def test_get_header(self
):
1366 headers
.add_header('Set-Cookie', 'cookie1')
1367 headers
.add_header('Set-cookie', 'cookie2')
1368 headers
.add_header('Test', 'test')
1369 headers
.add_header('Test', 'test2')
1370 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1371 assert res
.get_header('test') == 'test, test2'
1372 assert res
.get_header('set-Cookie') == 'cookie1'
1373 assert res
.get_header('notexist', 'default') == 'default'
1375 def test_compat(self
):
1376 res
= Response(io
.BytesIO(b
''), url
='test://', status
=404, headers
={'test': 'test'}
)
1377 with warnings
.catch_warnings():
1378 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1379 assert res
.code
== res
.getcode() == res
.status
1380 assert res
.geturl() == res
.url
1381 assert res
.info() is res
.headers
1382 assert res
.getheader('test') == res
.get_header('test')