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
== '/redirect_dotsegments':
177 self
.send_response(301)
178 # redirect to /headers but with dot segments before
179 self
.send_header('Location', '/a/b/./../../headers')
180 self
.send_header('Content-Length', '0')
182 elif self
.path
.startswith('/redirect_'):
184 elif self
.path
.startswith('/method'):
185 self
._method
('GET', str(self
.headers
).encode())
186 elif self
.path
.startswith('/headers'):
188 elif self
.path
.startswith('/308-to-headers'):
189 self
.send_response(308)
190 self
.send_header('Location', '/headers')
191 self
.send_header('Content-Length', '0')
193 elif self
.path
== '/trailing_garbage':
194 payload
= b
'<html><video src="/vid.mp4" /></html>'
195 self
.send_response(200)
196 self
.send_header('Content-Type', 'text/html; charset=utf-8')
197 self
.send_header('Content-Encoding', 'gzip')
199 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
201 compressed
= buf
.getvalue() + b
'trailing garbage'
202 self
.send_header('Content-Length', str(len(compressed
)))
204 self
.wfile
.write(compressed
)
205 elif self
.path
== '/302-non-ascii-redirect':
206 new_url
= f
'http://127.0.0.1:{http_server_port(self.server)}/中文.html'
207 self
.send_response(301)
208 self
.send_header('Location', new_url
)
209 self
.send_header('Content-Length', '0')
211 elif self
.path
== '/content-encoding':
212 encodings
= self
.headers
.get('ytdl-encoding', '')
213 payload
= b
'<html><video src="/vid.mp4" /></html>'
214 for encoding
in filter(None, (e
.strip() for e
in encodings
.split(','))):
215 if encoding
== 'br' and brotli
:
216 payload
= brotli
.compress(payload
)
217 elif encoding
== 'gzip':
219 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
221 payload
= buf
.getvalue()
222 elif encoding
== 'deflate':
223 payload
= zlib
.compress(payload
)
224 elif encoding
== 'unsupported':
230 self
.send_response(200)
231 self
.send_header('Content-Encoding', encodings
)
232 self
.send_header('Content-Length', str(len(payload
)))
234 self
.wfile
.write(payload
)
235 elif self
.path
.startswith('/gen_'):
236 payload
= b
'<html></html>'
237 self
.send_response(int(self
.path
[len('/gen_'):]))
238 self
.send_header('Content-Type', 'text/html; charset=utf-8')
239 self
.send_header('Content-Length', str(len(payload
)))
241 self
.wfile
.write(payload
)
242 elif self
.path
.startswith('/incompleteread'):
243 payload
= b
'<html></html>'
244 self
.send_response(200)
245 self
.send_header('Content-Type', 'text/html; charset=utf-8')
246 self
.send_header('Content-Length', '234234')
248 self
.wfile
.write(payload
)
250 elif self
.path
.startswith('/timeout_'):
251 time
.sleep(int(self
.path
[len('/timeout_'):]))
253 elif self
.path
== '/source_address':
254 payload
= str(self
.client_address
[0]).encode()
255 self
.send_response(200)
256 self
.send_header('Content-Type', 'text/html; charset=utf-8')
257 self
.send_header('Content-Length', str(len(payload
)))
259 self
.wfile
.write(payload
)
264 def send_header(self
, keyword
, value
):
266 Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
267 This is against what is defined in RFC 3986, however we need to test we support this
268 since some sites incorrectly do this.
270 if keyword
.lower() == 'connection':
271 return super().send_header(keyword
, value
)
273 if not hasattr(self
, '_headers_buffer'):
274 self
._headers
_buffer
= []
276 self
._headers
_buffer
.append(f
'{keyword}: {value}\r\n'.encode())
279 def validate_and_send(rh
, req
):
284 class TestRequestHandlerBase
:
286 def setup_class(cls
):
287 cls
.http_httpd
= http
.server
.ThreadingHTTPServer(
288 ('127.0.0.1', 0), HTTPTestRequestHandler
)
289 cls
.http_port
= http_server_port(cls
.http_httpd
)
290 cls
.http_server_thread
= threading
.Thread(target
=cls
.http_httpd
.serve_forever
)
291 # FIXME: we should probably stop the http server thread after each test
292 # See: https://github.com/yt-dlp/yt-dlp/pull/7094#discussion_r1199746041
293 cls
.http_server_thread
.daemon
= True
294 cls
.http_server_thread
.start()
297 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
298 cls
.https_httpd
= http
.server
.ThreadingHTTPServer(
299 ('127.0.0.1', 0), HTTPTestRequestHandler
)
300 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
301 sslctx
.load_cert_chain(certfn
, None)
302 cls
.https_httpd
.socket
= sslctx
.wrap_socket(cls
.https_httpd
.socket
, server_side
=True)
303 cls
.https_port
= http_server_port(cls
.https_httpd
)
304 cls
.https_server_thread
= threading
.Thread(target
=cls
.https_httpd
.serve_forever
)
305 cls
.https_server_thread
.daemon
= True
306 cls
.https_server_thread
.start()
310 def handler(request
):
311 RH_KEY
= request
.param
312 if inspect
.isclass(RH_KEY
) and issubclass(RH_KEY
, RequestHandler
):
314 elif RH_KEY
in _REQUEST_HANDLERS
:
315 handler
= _REQUEST_HANDLERS
[RH_KEY
]
317 pytest
.skip(f
'{RH_KEY} request handler is not available')
319 return functools
.partial(handler
, logger
=FakeLogger
)
322 class TestHTTPRequestHandler(TestRequestHandlerBase
):
323 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
324 def test_verify_cert(self
, handler
):
325 with handler() as rh
:
326 with pytest
.raises(CertificateVerifyError
):
327 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
329 with handler(verify
=False) as rh
:
330 r
= validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
331 assert r
.status
== 200
334 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
335 def test_ssl_error(self
, handler
):
336 # HTTPS server with too old TLS version
337 # XXX: is there a better way to test this than to create a new server?
338 https_httpd
= http
.server
.ThreadingHTTPServer(
339 ('127.0.0.1', 0), HTTPTestRequestHandler
)
340 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
341 https_httpd
.socket
= sslctx
.wrap_socket(https_httpd
.socket
, server_side
=True)
342 https_port
= http_server_port(https_httpd
)
343 https_server_thread
= threading
.Thread(target
=https_httpd
.serve_forever
)
344 https_server_thread
.daemon
= True
345 https_server_thread
.start()
347 with handler(verify
=False) as rh
:
348 with pytest
.raises(SSLError
, match
='sslv3 alert handshake failure') as exc_info
:
349 validate_and_send(rh
, Request(f
'https://127.0.0.1:{https_port}/headers'))
350 assert not issubclass(exc_info
.type, CertificateVerifyError
)
352 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
353 def test_percent_encode(self
, handler
):
354 with handler() as rh
:
355 # Unicode characters should be encoded with uppercase percent-encoding
356 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/中文.html'))
357 assert res
.status
== 200
359 # don't normalize existing percent encodings
360 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/%c7%9f'))
361 assert res
.status
== 200
364 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
365 def test_remove_dot_segments(self
, handler
):
366 with handler() as rh
:
367 # This isn't a comprehensive test,
368 # but it should be enough to check whether the handler is removing dot segments
369 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/a/b/./../../headers'))
370 assert res
.status
== 200
371 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
374 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_dotsegments'))
375 assert res
.status
== 200
376 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
379 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
380 def test_unicode_path_redirection(self
, handler
):
381 with handler() as rh
:
382 r
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
383 assert r
.url
== f
'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
386 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
387 def test_raise_http_error(self
, handler
):
388 with handler() as rh
:
389 for bad_status
in (400, 500, 599, 302):
390 with pytest
.raises(HTTPError
):
391 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_%d' % (self
.http_port
, bad_status
)))
393 # Should not raise an error
394 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_200' % self
.http_port
)).close()
396 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
397 def test_response_url(self
, handler
):
398 with handler() as rh
:
399 # Response url should be that of the last url in redirect chain
400 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_301'))
401 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/method'
403 res2
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_200'))
404 assert res2
.url
== f
'http://127.0.0.1:{self.http_port}/gen_200'
407 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
408 def test_redirect(self
, handler
):
409 with handler() as rh
:
410 def do_req(redirect_status
, method
, assert_no_content
=False):
411 data
= b
'testdata' if method
in ('POST', 'PUT') else None
412 res
= validate_and_send(
413 rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method
=method
, data
=data
))
418 data_sent
+= res
.read(len(data
))
419 if data_sent
!= data
:
423 headers
+= res
.read()
425 if assert_no_content
or data
is None:
426 assert b
'Content-Type' not in headers
427 assert b
'Content-Length' not in headers
429 assert b
'Content-Type' in headers
430 assert b
'Content-Length' in headers
432 return data_sent
.decode(), res
.headers
.get('method', '')
434 # A 303 must either use GET or HEAD for subsequent request
435 assert do_req(303, 'POST', True) == ('', 'GET')
436 assert do_req(303, 'HEAD') == ('', 'HEAD')
438 assert do_req(303, 'PUT', True) == ('', 'GET')
440 # 301 and 302 turn POST only into a GET
441 assert do_req(301, 'POST', True) == ('', 'GET')
442 assert do_req(301, 'HEAD') == ('', 'HEAD')
443 assert do_req(302, 'POST', True) == ('', 'GET')
444 assert do_req(302, 'HEAD') == ('', 'HEAD')
446 assert do_req(301, 'PUT') == ('testdata', 'PUT')
447 assert do_req(302, 'PUT') == ('testdata', 'PUT')
449 # 307 and 308 should not change method
450 for m
in ('POST', 'PUT'):
451 assert do_req(307, m
) == ('testdata', m
)
452 assert do_req(308, m
) == ('testdata', m
)
454 assert do_req(307, 'HEAD') == ('', 'HEAD')
455 assert do_req(308, 'HEAD') == ('', 'HEAD')
457 # These should not redirect and instead raise an HTTPError
458 for code
in (300, 304, 305, 306):
459 with pytest
.raises(HTTPError
):
462 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
463 def test_request_cookie_header(self
, handler
):
464 # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
465 with handler() as rh
:
466 # Specified Cookie header should be used
467 res
= validate_and_send(
469 f
'http://127.0.0.1:{self.http_port}/headers',
470 headers
={'Cookie': 'test=test'}
)).read().decode()
471 assert 'Cookie: test=test' in res
473 # Specified Cookie header should be removed on any redirect
474 res
= validate_and_send(
476 f
'http://127.0.0.1:{self.http_port}/308-to-headers',
477 headers
={'Cookie': 'test=test'}
)).read().decode()
478 assert 'Cookie: test=test' not in res
480 # Specified Cookie header should override global cookiejar for that request
481 cookiejar
= http
.cookiejar
.CookieJar()
482 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
483 version
=0, name
='test', value
='ytdlp', port
=None, port_specified
=False,
484 domain
='127.0.0.1', domain_specified
=True, domain_initial_dot
=False, path
='/',
485 path_specified
=True, secure
=False, expires
=None, discard
=False, comment
=None,
486 comment_url
=None, rest
={}))
488 with handler(cookiejar
=cookiejar
) as rh
:
489 data
= validate_and_send(
490 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', headers
={'cookie': 'test=test'}
)).read()
491 assert b
'Cookie: test=ytdlp' not in data
492 assert b
'Cookie: test=test' in data
494 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
495 def test_redirect_loop(self
, handler
):
496 with handler() as rh
:
497 with pytest
.raises(HTTPError
, match
='redirect loop'):
498 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_loop'))
500 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
501 def test_incompleteread(self
, handler
):
502 with handler(timeout
=2) as rh
:
503 with pytest
.raises(IncompleteRead
):
504 validate_and_send(rh
, Request('http://127.0.0.1:%d/incompleteread' % self
.http_port
)).read()
506 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
507 def test_cookies(self
, handler
):
508 cookiejar
= http
.cookiejar
.CookieJar()
509 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
510 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
511 False, '/headers', True, False, None, False, None, None, {}))
513 with handler(cookiejar
=cookiejar
) as rh
:
514 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
515 assert b
'Cookie: test=ytdlp' in data
518 with handler() as rh
:
519 data
= validate_and_send(
520 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', extensions
={'cookiejar': cookiejar}
)).read()
521 assert b
'Cookie: test=ytdlp' in data
523 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
524 def test_headers(self
, handler
):
526 with handler(headers
=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'}
)) as rh
:
528 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
529 assert b
'Test1: test' in data
531 # Per request headers, merged with global
532 data
= validate_and_send(rh
, Request(
533 f
'http://127.0.0.1:{self.http_port}/headers', headers
={'test2': 'changed', 'test3': 'test3'}
)).read()
534 assert b
'Test1: test' in data
535 assert b
'Test2: changed' in data
536 assert b
'Test2: test2' not in data
537 assert b
'Test3: test3' in data
539 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
540 def test_timeout(self
, handler
):
541 with handler() as rh
:
542 # Default timeout is 20 seconds, so this should go through
544 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_3'))
546 with handler(timeout
=0.5) as rh
:
547 with pytest
.raises(TransportError
):
549 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1'))
551 # Per request timeout, should override handler timeout
553 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1', extensions
={'timeout': 4}
))
555 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
556 def test_source_address(self
, handler
):
557 source_address
= f
'127.0.0.{random.randint(5, 255)}'
558 with handler(source_address
=source_address
) as rh
:
559 data
= validate_and_send(
560 rh
, Request(f
'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
561 assert source_address
== data
563 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
564 def test_gzip_trailing_garbage(self
, handler
):
565 with handler() as rh
:
566 data
= validate_and_send(rh
, Request(f
'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
567 assert data
== '<html><video src="/vid.mp4" /></html>'
569 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
570 @pytest.mark.skipif(not brotli
, reason
='brotli support is not installed')
571 def test_brotli(self
, handler
):
572 with handler() as rh
:
573 res
= validate_and_send(
575 f
'http://127.0.0.1:{self.http_port}/content-encoding',
576 headers
={'ytdl-encoding': 'br'}
))
577 assert res
.headers
.get('Content-Encoding') == 'br'
578 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
580 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
581 def test_deflate(self
, handler
):
582 with handler() as rh
:
583 res
= validate_and_send(
585 f
'http://127.0.0.1:{self.http_port}/content-encoding',
586 headers
={'ytdl-encoding': 'deflate'}
))
587 assert res
.headers
.get('Content-Encoding') == 'deflate'
588 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
590 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
591 def test_gzip(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': 'gzip'}
))
597 assert res
.headers
.get('Content-Encoding') == 'gzip'
598 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
600 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
601 def test_multiple_encodings(self
, handler
):
602 with handler() as rh
:
603 for pair
in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
604 res
= validate_and_send(
606 f
'http://127.0.0.1:{self.http_port}/content-encoding',
607 headers
={'ytdl-encoding': pair}
))
608 assert res
.headers
.get('Content-Encoding') == pair
609 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
611 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
612 def test_unsupported_encoding(self
, handler
):
613 with handler() as rh
:
614 res
= validate_and_send(
616 f
'http://127.0.0.1:{self.http_port}/content-encoding',
617 headers
={'ytdl-encoding': 'unsupported'}
))
618 assert res
.headers
.get('Content-Encoding') == 'unsupported'
619 assert res
.read() == b
'raw'
621 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
622 def test_read(self
, handler
):
623 with handler() as rh
:
624 res
= validate_and_send(
625 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers'))
626 assert res
.readable()
627 assert res
.read(1) == b
'H'
628 assert res
.read(3) == b
'ost'
631 class TestHTTPProxy(TestRequestHandlerBase
):
633 def setup_class(cls
):
634 super().setup_class()
636 cls
.proxy
= http
.server
.ThreadingHTTPServer(
637 ('127.0.0.1', 0), _build_proxy_handler('normal'))
638 cls
.proxy_port
= http_server_port(cls
.proxy
)
639 cls
.proxy_thread
= threading
.Thread(target
=cls
.proxy
.serve_forever
)
640 cls
.proxy_thread
.daemon
= True
641 cls
.proxy_thread
.start()
644 cls
.geo_proxy
= http
.server
.ThreadingHTTPServer(
645 ('127.0.0.1', 0), _build_proxy_handler('geo'))
646 cls
.geo_port
= http_server_port(cls
.geo_proxy
)
647 cls
.geo_proxy_thread
= threading
.Thread(target
=cls
.geo_proxy
.serve_forever
)
648 cls
.geo_proxy_thread
.daemon
= True
649 cls
.geo_proxy_thread
.start()
651 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
652 def test_http_proxy(self
, handler
):
653 http_proxy
= f
'http://127.0.0.1:{self.proxy_port}'
654 geo_proxy
= f
'http://127.0.0.1:{self.geo_port}'
656 # Test global http proxy
657 # Test per request http proxy
658 # Test per request http proxy disables proxy
659 url
= 'http://foo.com/bar'
662 with handler(proxies
={'http': http_proxy}
) as rh
:
663 res
= validate_and_send(rh
, Request(url
)).read().decode()
664 assert res
== f
'normal: {url}'
666 # Per request proxy overrides global
667 res
= validate_and_send(rh
, Request(url
, proxies
={'http': geo_proxy}
)).read().decode()
668 assert res
== f
'geo: {url}'
670 # and setting to None disables all proxies for that request
671 real_url
= f
'http://127.0.0.1:{self.http_port}/headers'
672 res
= validate_and_send(
673 rh
, Request(real_url
, proxies
={'http': None}
)).read().decode()
674 assert res
!= f
'normal: {real_url}'
675 assert 'Accept' in res
677 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
678 def test_noproxy(self
, handler
):
679 with handler(proxies
={'proxy': f'http://127.0.0.1:{self.proxy_port}
'}) as rh:
681 for no_proxy in (f'127.0.0.1:{self.http_port}
', '127.0.0.1', 'localhost
'):
682 nop_response = validate_and_send(
683 rh, Request(f'http
://127.0.0.1:{self.http_port}
/headers
', proxies={'no': no_proxy})).read().decode(
685 assert 'Accept
' in nop_response
687 @pytest.mark.parametrize('handler
', ['Urllib
'], indirect=True)
688 def test_allproxy(self, handler):
689 url = 'http
://foo
.com
/bar
'
690 with handler() as rh:
691 response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
693 assert response
== f
'normal: {url}'
695 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
696 def test_http_proxy_with_idn(self
, handler
):
697 with handler(proxies
={
698 'http': f
'http://127.0.0.1:{self.proxy_port}',
700 url
= 'http://中文.tw/'
701 response
= rh
.send(Request(url
)).read().decode()
702 # b'xn--fiq228c' is '中文'.encode('idna')
703 assert response
== 'normal: http://xn--fiq228c.tw/'
706 class TestClientCertificate
:
709 def setup_class(cls
):
710 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
711 cls
.certdir
= os
.path
.join(TEST_DIR
, 'testdata', 'certificate')
712 cacertfn
= os
.path
.join(cls
.certdir
, 'ca.crt')
713 cls
.httpd
= http
.server
.ThreadingHTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler
)
714 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
715 sslctx
.verify_mode
= ssl
.CERT_REQUIRED
716 sslctx
.load_verify_locations(cafile
=cacertfn
)
717 sslctx
.load_cert_chain(certfn
, None)
718 cls
.httpd
.socket
= sslctx
.wrap_socket(cls
.httpd
.socket
, server_side
=True)
719 cls
.port
= http_server_port(cls
.httpd
)
720 cls
.server_thread
= threading
.Thread(target
=cls
.httpd
.serve_forever
)
721 cls
.server_thread
.daemon
= True
722 cls
.server_thread
.start()
724 def _run_test(self
, handler
, **handler_kwargs
):
726 # Disable client-side validation of unacceptable self-signed testcert.pem
727 # The test is of a check on the server side, so unaffected
731 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.port}/video.html')).read().decode()
733 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
734 def test_certificate_combined_nopass(self
, handler
):
735 self
._run
_test
(handler
, client_cert
={
736 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithkey.crt'),
739 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
740 def test_certificate_nocombined_nopass(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
, 'client.key'),
746 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
747 def test_certificate_combined_pass(self
, handler
):
748 self
._run
_test
(handler
, client_cert
={
749 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithencryptedkey.crt'),
750 'client_certificate_password': 'foobar',
753 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
754 def test_certificate_nocombined_pass(self
, handler
):
755 self
._run
_test
(handler
, client_cert
={
756 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
757 'client_certificate_key': os
.path
.join(self
.certdir
, 'clientencrypted.key'),
758 'client_certificate_password': 'foobar',
762 class TestUrllibRequestHandler(TestRequestHandlerBase
):
763 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
764 def test_file_urls(self
, handler
):
765 # See https://github.com/ytdl-org/youtube-dl/issues/8227
766 tf
= tempfile
.NamedTemporaryFile(delete
=False)
769 req
= Request(pathlib
.Path(tf
.name
).as_uri())
770 with handler() as rh
:
771 with pytest
.raises(UnsupportedRequest
):
774 # Test that urllib never loaded FileHandler
775 with pytest
.raises(TransportError
):
778 with handler(enable_file_urls
=True) as rh
:
779 res
= validate_and_send(rh
, req
)
780 assert res
.read() == b
'foobar'
785 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
786 def test_http_error_returns_content(self
, handler
):
787 # urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
789 with handler() as rh
:
792 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_404'))
793 except HTTPError
as e
:
796 assert get_response().read() == b
'<html></html>'
798 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
799 def test_verify_cert_error_text(self
, handler
):
800 # Check the output of the error message
801 with handler() as rh
:
803 CertificateVerifyError
,
804 match
=r
'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate'
806 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
808 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
809 @pytest.mark.parametrize('req,match,version_check', [
810 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
811 # bpo-39603: Check implemented in 3.7.9+, 3.8.5+
813 Request('http://127.0.0.1', method
='GET\n'),
814 'method can\'t contain control characters',
815 lambda v
: v
< (3, 7, 9) or (3, 8, 0) <= v
< (3, 8, 5)
817 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
818 # bpo-38576: Check implemented in 3.7.8+, 3.8.3+
820 Request('http://127.0.0. 1', method
='GET'),
821 'URL can\'t contain control characters',
822 lambda v
: v
< (3, 7, 8) or (3, 8, 0) <= v
< (3, 8, 3)
824 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
825 (Request('http://127.0.0.1', headers
={'foo\n': 'bar'}
), 'Invalid header name', None),
827 def test_httplib_validation_errors(self
, handler
, req
, match
, version_check
):
828 if version_check
and version_check(sys
.version_info
):
829 pytest
.skip(f
'Python {sys.version} version does not have the required validation for this test.')
831 with handler() as rh
:
832 with pytest
.raises(RequestError
, match
=match
) as exc_info
:
833 validate_and_send(rh
, req
)
834 assert not isinstance(exc_info
.value
, TransportError
)
837 def run_validation(handler
, error
, req
, **handler_kwargs
):
838 with handler(**handler_kwargs
) as rh
:
840 with pytest
.raises(error
):
846 class TestRequestHandlerValidation
:
848 class ValidationRH(RequestHandler
):
849 def _send(self
, request
):
850 raise RequestError('test')
852 class NoCheckRH(ValidationRH
):
853 _SUPPORTED_FEATURES
= None
854 _SUPPORTED_PROXY_SCHEMES
= None
855 _SUPPORTED_URL_SCHEMES
= None
857 def _check_extensions(self
, extensions
):
860 class HTTPSupportedRH(ValidationRH
):
861 _SUPPORTED_URL_SCHEMES
= ('http',)
864 # scheme, expected to fail, handler kwargs
867 ('https', False, {}),
870 ('file', UnsupportedRequest
, {}),
871 ('file', False, {'enable_file_urls': True}
),
873 (NoCheckRH
, [('http', False, {})]),
874 (ValidationRH
, [('http', UnsupportedRequest
, {})])
877 PROXY_SCHEME_TESTS
= [
878 # scheme, expected to fail
881 ('https', UnsupportedRequest
),
886 ('socks', UnsupportedRequest
),
888 (NoCheckRH
, [('http', False)]),
889 (HTTPSupportedRH
, [('http', UnsupportedRequest
)]),
893 # key, expected to fail
896 ('unrelated', False),
898 (NoCheckRH
, [('all', False)]),
899 (HTTPSupportedRH
, [('all', UnsupportedRequest
)]),
900 (HTTPSupportedRH
, [('no', UnsupportedRequest
)]),
905 ({'cookiejar': 'notacookiejar'}
, AssertionError),
906 ({'cookiejar': CookieJar()}
, False),
907 ({'timeout': 1}
, False),
908 ({'timeout': 'notatimeout'}
, AssertionError),
909 ({'unsupported': 'value'}
, UnsupportedRequest
),
912 ({'cookiejar': 'notacookiejar'}
, False),
913 ({'somerandom': 'test'}
, False), # but any extension is allowed through
917 @pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
918 (handler_tests
[0], scheme
, fail
, handler_kwargs
)
919 for handler_tests
in URL_SCHEME_TESTS
920 for scheme
, fail
, handler_kwargs
in handler_tests
[1]
922 ], indirect
=['handler'])
923 def test_url_scheme(self
, handler
, scheme
, fail
, handler_kwargs
):
924 run_validation(handler
, fail
, Request(f
'{scheme}://'), **(handler_kwargs
or {}))
926 @pytest.mark.parametrize('handler,fail', [('Urllib', False)], indirect
=['handler'])
927 def test_no_proxy(self
, handler
, fail
):
928 run_validation(handler
, fail
, Request('http://', proxies
={'no': '127.0.0.1,github.com'}
))
929 run_validation(handler
, fail
, Request('http://'), proxies
={'no': '127.0.0.1,github.com'}
)
931 @pytest.mark.parametrize('handler,proxy_key,fail', [
932 (handler_tests
[0], proxy_key
, fail
)
933 for handler_tests
in PROXY_KEY_TESTS
934 for proxy_key
, fail
in handler_tests
[1]
935 ], indirect
=['handler'])
936 def test_proxy_key(self
, handler
, proxy_key
, fail
):
937 run_validation(handler
, fail
, Request('http://', proxies
={proxy_key: 'http://example.com'}
))
938 run_validation(handler
, fail
, Request('http://'), proxies
={proxy_key: 'http://example.com'}
)
940 @pytest.mark.parametrize('handler,scheme,fail', [
941 (handler_tests
[0], scheme
, fail
)
942 for handler_tests
in PROXY_SCHEME_TESTS
943 for scheme
, fail
in handler_tests
[1]
944 ], indirect
=['handler'])
945 def test_proxy_scheme(self
, handler
, scheme
, fail
):
946 run_validation(handler
, fail
, Request('http://', proxies
={'http': f'{scheme}
://example
.com
'}))
947 run_validation(handler, fail, Request('http
://'), proxies={'http': f'{scheme}://example.com'})
949 @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH
], indirect
=True)
950 def test_empty_proxy(self
, handler
):
951 run_validation(handler
, False, Request('http://', proxies
={'http': None}
))
952 run_validation(handler
, False, Request('http://'), proxies
={'http': None}
)
954 @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
955 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
956 def test_invalid_proxy_url(self
, handler
, proxy_url
):
957 run_validation(handler
, UnsupportedRequest
, Request('http://', proxies
={'http': proxy_url}
))
959 @pytest.mark.parametrize('handler,extensions,fail', [
960 (handler_tests
[0], extensions
, fail
)
961 for handler_tests
in EXTENSION_TESTS
962 for extensions
, fail
in handler_tests
[1]
963 ], indirect
=['handler'])
964 def test_extension(self
, handler
, extensions
, fail
):
966 handler
, fail
, Request('http://', extensions
=extensions
))
968 def test_invalid_request_type(self
):
969 rh
= self
.ValidationRH(logger
=FakeLogger())
970 for method
in (rh
.validate
, rh
.send
):
971 with pytest
.raises(TypeError, match
='Expected an instance of Request'):
972 method('not a request')
975 class FakeResponse(Response
):
976 def __init__(self
, request
):
977 # XXX: we could make request part of standard response interface
978 self
.request
= request
979 super().__init
__(fp
=io
.BytesIO(b
''), headers
={}, url
=request
.url
)
982 class FakeRH(RequestHandler
):
984 def _validate(self
, request
):
987 def _send(self
, request
: Request
):
988 if request
.url
.startswith('ssl://'):
989 raise SSLError(request
.url
[len('ssl://'):])
990 return FakeResponse(request
)
993 class FakeRHYDL(FakeYDL
):
994 def __init__(self
, *args
, **kwargs
):
995 super().__init
__(*args
, **kwargs
)
996 self
._request
_director
= self
.build_request_director([FakeRH
])
999 class TestRequestDirector
:
1001 def test_handler_operations(self
):
1002 director
= RequestDirector(logger
=FakeLogger())
1003 handler
= FakeRH(logger
=FakeLogger())
1004 director
.add_handler(handler
)
1005 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler
1007 # Handler should overwrite
1008 handler2
= FakeRH(logger
=FakeLogger())
1009 director
.add_handler(handler2
)
1010 assert director
.handlers
.get(FakeRH
.RH_KEY
) is not handler
1011 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler2
1012 assert len(director
.handlers
) == 1
1014 class AnotherFakeRH(FakeRH
):
1016 director
.add_handler(AnotherFakeRH(logger
=FakeLogger()))
1017 assert len(director
.handlers
) == 2
1018 assert director
.handlers
.get(AnotherFakeRH
.RH_KEY
).RH_KEY
== AnotherFakeRH
.RH_KEY
1020 director
.handlers
.pop(FakeRH
.RH_KEY
, None)
1021 assert director
.handlers
.get(FakeRH
.RH_KEY
) is None
1022 assert len(director
.handlers
) == 1
1024 # RequestErrors should passthrough
1025 with pytest
.raises(SSLError
):
1026 director
.send(Request('ssl://something'))
1028 def test_send(self
):
1029 director
= RequestDirector(logger
=FakeLogger())
1030 with pytest
.raises(RequestError
):
1031 director
.send(Request('any://'))
1032 director
.add_handler(FakeRH(logger
=FakeLogger()))
1033 assert isinstance(director
.send(Request('http://')), FakeResponse
)
1035 def test_unsupported_handlers(self
):
1036 director
= RequestDirector(logger
=FakeLogger())
1037 director
.add_handler(FakeRH(logger
=FakeLogger()))
1039 class SupportedRH(RequestHandler
):
1040 _SUPPORTED_URL_SCHEMES
= ['http']
1042 def _send(self
, request
: Request
):
1043 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
1045 # This handler should by default take preference over FakeRH
1046 director
.add_handler(SupportedRH(logger
=FakeLogger()))
1047 assert director
.send(Request('http://')).read() == b
'supported'
1048 assert director
.send(Request('any://')).read() == b
''
1050 director
.handlers
.pop(FakeRH
.RH_KEY
)
1051 with pytest
.raises(NoSupportingHandlers
):
1052 director
.send(Request('any://'))
1054 def test_unexpected_error(self
):
1055 director
= RequestDirector(logger
=FakeLogger())
1057 class UnexpectedRH(FakeRH
):
1058 def _send(self
, request
: Request
):
1059 raise TypeError('something')
1061 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1062 with pytest
.raises(NoSupportingHandlers
, match
=r
'1 unexpected error'):
1063 director
.send(Request('any://'))
1065 director
.handlers
.clear()
1066 assert len(director
.handlers
) == 0
1068 # Should not be fatal
1069 director
.add_handler(FakeRH(logger
=FakeLogger()))
1070 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1071 assert director
.send(Request('any://'))
1074 # XXX: do we want to move this to test_YoutubeDL.py?
1075 class TestYoutubeDLNetworking
:
1078 def build_handler(ydl
, handler
: RequestHandler
= FakeRH
):
1079 return ydl
.build_request_director([handler
]).handlers
.get(handler
.RH_KEY
)
1081 def test_compat_opener(self
):
1082 with FakeYDL() as ydl
:
1083 with warnings
.catch_warnings():
1084 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1085 assert isinstance(ydl
._opener
, urllib
.request
.OpenerDirector
)
1087 @pytest.mark.parametrize('proxy,expected', [
1088 ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}
),
1089 ('', {'all': '__noproxy__'}
),
1090 (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}
) # env, set https
1092 def test_proxy(self
, proxy
, expected
):
1093 old_http_proxy
= os
.environ
.get('HTTP_PROXY')
1095 os
.environ
['HTTP_PROXY'] = 'http://127.0.0.1:8081' # ensure that provided proxies override env
1096 with FakeYDL({'proxy': proxy}
) as ydl
:
1097 assert ydl
.proxies
== expected
1100 os
.environ
['HTTP_PROXY'] = old_http_proxy
1102 def test_compat_request(self
):
1103 with FakeRHYDL() as ydl
:
1104 assert ydl
.urlopen('test://')
1105 urllib_req
= urllib
.request
.Request('http://foo.bar', data
=b
'test', method
='PUT', headers
={'X-Test': '1'}
)
1106 urllib_req
.add_unredirected_header('Cookie', 'bob=bob')
1107 urllib_req
.timeout
= 2
1108 with warnings
.catch_warnings():
1109 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1110 req
= ydl
.urlopen(urllib_req
).request
1111 assert req
.url
== urllib_req
.get_full_url()
1112 assert req
.data
== urllib_req
.data
1113 assert req
.method
== urllib_req
.get_method()
1114 assert 'X-Test' in req
.headers
1115 assert 'Cookie' in req
.headers
1116 assert req
.extensions
.get('timeout') == 2
1118 with pytest
.raises(AssertionError):
1121 def test_extract_basic_auth(self
):
1122 with FakeRHYDL() as ydl
:
1123 res
= ydl
.urlopen(Request('http://user:pass@foo.bar'))
1124 assert res
.request
.headers
['Authorization'] == 'Basic dXNlcjpwYXNz'
1126 def test_sanitize_url(self
):
1127 with FakeRHYDL() as ydl
:
1128 res
= ydl
.urlopen(Request('httpss://foo.bar'))
1129 assert res
.request
.url
== 'https://foo.bar'
1131 def test_file_urls_error(self
):
1132 # use urllib handler
1133 with FakeYDL() as ydl
:
1134 with pytest
.raises(RequestError
, match
=r
'file:// URLs are disabled by default'):
1135 ydl
.urlopen('file://')
1137 def test_legacy_server_connect_error(self
):
1138 with FakeRHYDL() as ydl
:
1139 for error
in ('UNSAFE_LEGACY_RENEGOTIATION_DISABLED', 'SSLV3_ALERT_HANDSHAKE_FAILURE'):
1140 with pytest
.raises(RequestError
, match
=r
'Try using --legacy-server-connect'):
1141 ydl
.urlopen(f
'ssl://{error}')
1143 with pytest
.raises(SSLError
, match
='testerror'):
1144 ydl
.urlopen('ssl://testerror')
1146 @pytest.mark.parametrize('proxy_key,proxy_url,expected', [
1147 ('http', '__noproxy__', None),
1148 ('no', '127.0.0.1,foo.bar', '127.0.0.1,foo.bar'),
1149 ('https', 'example.com', 'http://example.com'),
1150 ('https', '//example.com', 'http://example.com'),
1151 ('https', 'socks5://example.com', 'socks5h://example.com'),
1152 ('http', 'socks://example.com', 'socks4://example.com'),
1153 ('http', 'socks4://example.com', 'socks4://example.com'),
1154 ('unrelated', '/bad/proxy', '/bad/proxy'), # clean_proxies should ignore bad proxies
1156 def test_clean_proxy(self
, proxy_key
, proxy_url
, expected
):
1157 # proxies should be cleaned in urlopen()
1158 with FakeRHYDL() as ydl
:
1159 req
= ydl
.urlopen(Request('test://', proxies
={proxy_key: proxy_url}
)).request
1160 assert req
.proxies
[proxy_key
] == expected
1162 # and should also be cleaned when building the handler
1163 env_key
= f
'{proxy_key.upper()}_PROXY'
1164 old_env_proxy
= os
.environ
.get(env_key
)
1166 os
.environ
[env_key
] = proxy_url
# ensure that provided proxies override env
1167 with FakeYDL() as ydl
:
1168 rh
= self
.build_handler(ydl
)
1169 assert rh
.proxies
[proxy_key
] == expected
1172 os
.environ
[env_key
] = old_env_proxy
1174 def test_clean_proxy_header(self
):
1175 with FakeRHYDL() as ydl
:
1176 req
= ydl
.urlopen(Request('test://', headers
={'ytdl-request-proxy': '//foo.bar'}
)).request
1177 assert 'ytdl-request-proxy' not in req
.headers
1178 assert req
.proxies
== {'all': 'http://foo.bar'}
1180 with FakeYDL({'http_headers': {'ytdl-request-proxy': '//foo.bar'}
}) as ydl
:
1181 rh
= self
.build_handler(ydl
)
1182 assert 'ytdl-request-proxy' not in rh
.headers
1183 assert rh
.proxies
== {'all': 'http://foo.bar'}
1185 def test_clean_header(self
):
1186 with FakeRHYDL() as ydl
:
1187 res
= ydl
.urlopen(Request('test://', headers
={'Youtubedl-no-compression': True}
))
1188 assert 'Youtubedl-no-compression' not in res
.request
.headers
1189 assert res
.request
.headers
.get('Accept-Encoding') == 'identity'
1191 with FakeYDL({'http_headers': {'Youtubedl-no-compression': True}
}) as ydl
:
1192 rh
= self
.build_handler(ydl
)
1193 assert 'Youtubedl-no-compression' not in rh
.headers
1194 assert rh
.headers
.get('Accept-Encoding') == 'identity'
1196 def test_build_handler_params(self
):
1198 'http_headers': {'test': 'testtest'}
,
1199 'socket_timeout': 2,
1200 'proxy': 'http://127.0.0.1:8080',
1201 'source_address': '127.0.0.45',
1202 'debug_printtraffic': True,
1203 'compat_opts': ['no-certifi'],
1204 'nocheckcertificate': True,
1205 'legacyserverconnect': True,
1207 rh
= self
.build_handler(ydl
)
1208 assert rh
.headers
.get('test') == 'testtest'
1209 assert 'Accept' in rh
.headers
# ensure std_headers are still there
1210 assert rh
.timeout
== 2
1211 assert rh
.proxies
.get('all') == 'http://127.0.0.1:8080'
1212 assert rh
.source_address
== '127.0.0.45'
1213 assert rh
.verbose
is True
1214 assert rh
.prefer_system_certs
is True
1215 assert rh
.verify
is False
1216 assert rh
.legacy_ssl_support
is True
1218 @pytest.mark.parametrize('ydl_params', [
1219 {'client_certificate': 'fakecert.crt'}
,
1220 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key'}
,
1221 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1222 {'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1224 def test_client_certificate(self
, ydl_params
):
1225 with FakeYDL(ydl_params
) as ydl
:
1226 rh
= self
.build_handler(ydl
)
1227 assert rh
._client
_cert
== ydl_params
# XXX: Too bound to implementation
1229 def test_urllib_file_urls(self
):
1230 with FakeYDL({'enable_file_urls': False}
) as ydl
:
1231 rh
= self
.build_handler(ydl
, UrllibRH
)
1232 assert rh
.enable_file_urls
is False
1234 with FakeYDL({'enable_file_urls': True}
) as ydl
:
1235 rh
= self
.build_handler(ydl
, UrllibRH
)
1236 assert rh
.enable_file_urls
is True
1241 def test_query(self
):
1242 req
= Request('http://example.com?q=something', query
={'v': 'xyz'}
)
1243 assert req
.url
== 'http://example.com?q=something&v=xyz'
1245 req
.update(query
={'v': '123'}
)
1246 assert req
.url
== 'http://example.com?q=something&v=123'
1247 req
.update(url
='http://example.com', query
={'v': 'xyz'}
)
1248 assert req
.url
== 'http://example.com?v=xyz'
1250 def test_method(self
):
1251 req
= Request('http://example.com')
1252 assert req
.method
== 'GET'
1254 assert req
.method
== 'POST'
1256 assert req
.method
== 'GET'
1259 assert req
.method
== 'PUT'
1261 assert req
.method
== 'PUT'
1262 with pytest
.raises(TypeError):
1265 def test_request_helpers(self
):
1266 assert HEADRequest('http://example.com').method
== 'HEAD'
1267 assert PUTRequest('http://example.com').method
== 'PUT'
1269 def test_headers(self
):
1270 req
= Request('http://example.com', headers
={'tesT': 'test'}
)
1271 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1272 req
.update(headers
={'teSt2': 'test2'}
)
1273 assert req
.headers
== HTTPHeaderDict({'test': 'test', 'test2': 'test2'}
)
1275 req
.headers
= new_headers
= HTTPHeaderDict({'test': 'test'}
)
1276 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1277 assert req
.headers
is new_headers
1279 # test converts dict to case insensitive dict
1280 req
.headers
= new_headers
= {'test2': 'test2'}
1281 assert isinstance(req
.headers
, HTTPHeaderDict
)
1282 assert req
.headers
is not new_headers
1284 with pytest
.raises(TypeError):
1287 def test_data_type(self
):
1288 req
= Request('http://example.com')
1289 assert req
.data
is None
1290 # test bytes is allowed
1292 assert req
.data
== b
'test'
1293 # test iterable of bytes is allowed
1294 i
= [b
'test', b
'test2']
1296 assert req
.data
== i
1298 # test file-like object is allowed
1299 f
= io
.BytesIO(b
'test')
1301 assert req
.data
== f
1303 # common mistake: test str not allowed
1304 with pytest
.raises(TypeError):
1306 assert req
.data
!= 'test'
1308 # common mistake: test dict is not allowed
1309 with pytest
.raises(TypeError):
1310 req
.data
= {'test': 'test'}
1311 assert req
.data
!= {'test': 'test'}
1313 def test_content_length_header(self
):
1314 req
= Request('http://example.com', headers
={'Content-Length': '0'}
, data
=b
'')
1315 assert req
.headers
.get('Content-Length') == '0'
1318 assert 'Content-Length' not in req
.headers
1320 req
= Request('http://example.com', headers
={'Content-Length': '10'}
)
1321 assert 'Content-Length' not in req
.headers
1323 def test_content_type_header(self
):
1324 req
= Request('http://example.com', headers
={'Content-Type': 'test'}
, data
=b
'test')
1325 assert req
.headers
.get('Content-Type') == 'test'
1327 assert req
.headers
.get('Content-Type') == 'test'
1329 assert 'Content-Type' not in req
.headers
1331 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1333 def test_update_req(self
):
1334 req
= Request('http://example.com')
1335 assert req
.data
is None
1336 assert req
.method
== 'GET'
1337 assert 'Content-Type' not in req
.headers
1338 # Test that zero-byte payloads will be sent
1339 req
.update(data
=b
'')
1340 assert req
.data
== b
''
1341 assert req
.method
== 'POST'
1342 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1344 def test_proxies(self
):
1345 req
= Request(url
='http://example.com', proxies
={'http': 'http://127.0.0.1:8080'}
)
1346 assert req
.proxies
== {'http': 'http://127.0.0.1:8080'}
1348 def test_extensions(self
):
1349 req
= Request(url
='http://example.com', extensions
={'timeout': 2}
)
1350 assert req
.extensions
== {'timeout': 2}
1352 def test_copy(self
):
1354 url
='http://example.com',
1355 extensions
={'cookiejar': CookieJar()}
,
1356 headers
={'Accept-Encoding': 'br'}
,
1357 proxies
={'http': 'http://127.0.0.1'}
,
1360 req_copy
= req
.copy()
1361 assert req_copy
is not req
1362 assert req_copy
.url
== req
.url
1363 assert req_copy
.headers
== req
.headers
1364 assert req_copy
.headers
is not req
.headers
1365 assert req_copy
.proxies
== req
.proxies
1366 assert req_copy
.proxies
is not req
.proxies
1368 # Data is not able to be copied
1369 assert req_copy
.data
== req
.data
1370 assert req_copy
.data
is req
.data
1372 # Shallow copy extensions
1373 assert req_copy
.extensions
is not req
.extensions
1374 assert req_copy
.extensions
['cookiejar'] == req
.extensions
['cookiejar']
1376 # Subclasses are copied by default
1377 class AnotherRequest(Request
):
1380 req
= AnotherRequest(url
='http://127.0.0.1')
1381 assert isinstance(req
.copy(), AnotherRequest
)
1384 req
= Request(url
='https://фtest.example.com/ some spaceв?ä=c',)
1385 assert req
.url
== 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c'
1387 assert Request(url
='//example.com').url
== 'http://example.com'
1389 with pytest
.raises(TypeError):
1390 Request(url
='https://').url
= None
1395 @pytest.mark.parametrize('reason,status,expected', [
1396 ('custom', 200, 'custom'),
1397 (None, 404, 'Not Found'), # fallback status
1398 ('', 403, 'Forbidden'),
1401 def test_reason(self
, reason
, status
, expected
):
1402 res
= Response(io
.BytesIO(b
''), url
='test://', headers
={}, status
=status
, reason
=reason
)
1403 assert res
.reason
== expected
1405 def test_headers(self
):
1407 headers
.add_header('Test', 'test')
1408 headers
.add_header('Test', 'test2')
1409 headers
.add_header('content-encoding', 'br')
1410 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1411 assert res
.headers
.get_all('test') == ['test', 'test2']
1412 assert 'Content-Encoding' in res
.headers
1414 def test_get_header(self
):
1416 headers
.add_header('Set-Cookie', 'cookie1')
1417 headers
.add_header('Set-cookie', 'cookie2')
1418 headers
.add_header('Test', 'test')
1419 headers
.add_header('Test', 'test2')
1420 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1421 assert res
.get_header('test') == 'test, test2'
1422 assert res
.get_header('set-Cookie') == 'cookie1'
1423 assert res
.get_header('notexist', 'default') == 'default'
1425 def test_compat(self
):
1426 res
= Response(io
.BytesIO(b
''), url
='test://', status
=404, headers
={'test': 'test'}
)
1427 with warnings
.catch_warnings():
1428 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1429 assert res
.code
== res
.getcode() == res
.status
1430 assert res
.geturl() == res
.url
1431 assert res
.info() is res
.headers
1432 assert res
.getheader('test') == res
.get_header('test')