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
.cookies
import YoutubeDLCookieJar
33 from yt_dlp
.dependencies
import brotli
34 from yt_dlp
.networking
import (
42 from yt_dlp
.networking
._urllib
import UrllibRH
43 from yt_dlp
.networking
.common
import _REQUEST_HANDLERS
44 from yt_dlp
.networking
.exceptions
import (
45 CertificateVerifyError
,
54 from yt_dlp
.utils
._utils
import _YDLLogger
as FakeLogger
55 from yt_dlp
.utils
.networking
import HTTPHeaderDict
57 TEST_DIR
= os
.path
.dirname(os
.path
.abspath(__file__
))
60 def _build_proxy_handler(name
):
61 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
64 def log_message(self
, format
, *args
):
68 self
.send_response(200)
69 self
.send_header('Content-Type', 'text/plain; charset=utf-8')
71 self
.wfile
.write('{self.proxy_name}: {self.path}'.format(self
=self
).encode())
72 return HTTPTestRequestHandler
75 class HTTPTestRequestHandler(http
.server
.BaseHTTPRequestHandler
):
76 protocol_version
= 'HTTP/1.1'
78 def log_message(self
, format
, *args
):
82 payload
= str(self
.headers
).encode()
83 self
.send_response(200)
84 self
.send_header('Content-Type', 'application/json')
85 self
.send_header('Content-Length', str(len(payload
)))
87 self
.wfile
.write(payload
)
90 self
.send_response(int(self
.path
[len('/redirect_'):]))
91 self
.send_header('Location', '/method')
92 self
.send_header('Content-Length', '0')
95 def _method(self
, method
, payload
=None):
96 self
.send_response(200)
97 self
.send_header('Content-Length', str(len(payload
or '')))
98 self
.send_header('Method', method
)
101 self
.wfile
.write(payload
)
103 def _status(self
, status
):
104 payload
= f
'<html>{status} NOT FOUND</html>'.encode()
105 self
.send_response(int(status
))
106 self
.send_header('Content-Type', 'text/html; charset=utf-8')
107 self
.send_header('Content-Length', str(len(payload
)))
109 self
.wfile
.write(payload
)
111 def _read_data(self
):
112 if 'Content-Length' in self
.headers
:
113 return self
.rfile
.read(int(self
.headers
['Content-Length']))
116 data
= self
._read
_data
() + str(self
.headers
).encode()
117 if self
.path
.startswith('/redirect_'):
119 elif self
.path
.startswith('/method'):
120 self
._method
('POST', data
)
121 elif self
.path
.startswith('/headers'):
127 if self
.path
.startswith('/redirect_'):
129 elif self
.path
.startswith('/method'):
135 data
= self
._read
_data
() + str(self
.headers
).encode()
136 if self
.path
.startswith('/redirect_'):
138 elif self
.path
.startswith('/method'):
139 self
._method
('PUT', data
)
144 if self
.path
== '/video.html':
145 payload
= b
'<html><video src="/vid.mp4" /></html>'
146 self
.send_response(200)
147 self
.send_header('Content-Type', 'text/html; charset=utf-8')
148 self
.send_header('Content-Length', str(len(payload
)))
150 self
.wfile
.write(payload
)
151 elif self
.path
== '/vid.mp4':
152 payload
= b
'\x00\x00\x00\x00\x20\x66\x74[video]'
153 self
.send_response(200)
154 self
.send_header('Content-Type', 'video/mp4')
155 self
.send_header('Content-Length', str(len(payload
)))
157 self
.wfile
.write(payload
)
158 elif self
.path
== '/%E4%B8%AD%E6%96%87.html':
159 payload
= b
'<html><video src="/vid.mp4" /></html>'
160 self
.send_response(200)
161 self
.send_header('Content-Type', 'text/html; charset=utf-8')
162 self
.send_header('Content-Length', str(len(payload
)))
164 self
.wfile
.write(payload
)
165 elif self
.path
== '/%c7%9f':
166 payload
= b
'<html><video src="/vid.mp4" /></html>'
167 self
.send_response(200)
168 self
.send_header('Content-Type', 'text/html; charset=utf-8')
169 self
.send_header('Content-Length', str(len(payload
)))
171 self
.wfile
.write(payload
)
172 elif self
.path
.startswith('/redirect_loop'):
173 self
.send_response(301)
174 self
.send_header('Location', self
.path
)
175 self
.send_header('Content-Length', '0')
177 elif self
.path
== '/redirect_dotsegments':
178 self
.send_response(301)
179 # redirect to /headers but with dot segments before
180 self
.send_header('Location', '/a/b/./../../headers')
181 self
.send_header('Content-Length', '0')
183 elif self
.path
.startswith('/redirect_'):
185 elif self
.path
.startswith('/method'):
186 self
._method
('GET', str(self
.headers
).encode())
187 elif self
.path
.startswith('/headers'):
189 elif self
.path
.startswith('/308-to-headers'):
190 self
.send_response(308)
191 self
.send_header('Location', '/headers')
192 self
.send_header('Content-Length', '0')
194 elif self
.path
== '/trailing_garbage':
195 payload
= b
'<html><video src="/vid.mp4" /></html>'
196 self
.send_response(200)
197 self
.send_header('Content-Type', 'text/html; charset=utf-8')
198 self
.send_header('Content-Encoding', 'gzip')
200 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
202 compressed
= buf
.getvalue() + b
'trailing garbage'
203 self
.send_header('Content-Length', str(len(compressed
)))
205 self
.wfile
.write(compressed
)
206 elif self
.path
== '/302-non-ascii-redirect':
207 new_url
= f
'http://127.0.0.1:{http_server_port(self.server)}/中文.html'
208 self
.send_response(301)
209 self
.send_header('Location', new_url
)
210 self
.send_header('Content-Length', '0')
212 elif self
.path
== '/content-encoding':
213 encodings
= self
.headers
.get('ytdl-encoding', '')
214 payload
= b
'<html><video src="/vid.mp4" /></html>'
215 for encoding
in filter(None, (e
.strip() for e
in encodings
.split(','))):
216 if encoding
== 'br' and brotli
:
217 payload
= brotli
.compress(payload
)
218 elif encoding
== 'gzip':
220 with gzip
.GzipFile(fileobj
=buf
, mode
='wb') as f
:
222 payload
= buf
.getvalue()
223 elif encoding
== 'deflate':
224 payload
= zlib
.compress(payload
)
225 elif encoding
== 'unsupported':
231 self
.send_response(200)
232 self
.send_header('Content-Encoding', encodings
)
233 self
.send_header('Content-Length', str(len(payload
)))
235 self
.wfile
.write(payload
)
236 elif self
.path
.startswith('/gen_'):
237 payload
= b
'<html></html>'
238 self
.send_response(int(self
.path
[len('/gen_'):]))
239 self
.send_header('Content-Type', 'text/html; charset=utf-8')
240 self
.send_header('Content-Length', str(len(payload
)))
242 self
.wfile
.write(payload
)
243 elif self
.path
.startswith('/incompleteread'):
244 payload
= b
'<html></html>'
245 self
.send_response(200)
246 self
.send_header('Content-Type', 'text/html; charset=utf-8')
247 self
.send_header('Content-Length', '234234')
249 self
.wfile
.write(payload
)
251 elif self
.path
.startswith('/timeout_'):
252 time
.sleep(int(self
.path
[len('/timeout_'):]))
254 elif self
.path
== '/source_address':
255 payload
= str(self
.client_address
[0]).encode()
256 self
.send_response(200)
257 self
.send_header('Content-Type', 'text/html; charset=utf-8')
258 self
.send_header('Content-Length', str(len(payload
)))
260 self
.wfile
.write(payload
)
265 def send_header(self
, keyword
, value
):
267 Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
268 This is against what is defined in RFC 3986, however we need to test we support this
269 since some sites incorrectly do this.
271 if keyword
.lower() == 'connection':
272 return super().send_header(keyword
, value
)
274 if not hasattr(self
, '_headers_buffer'):
275 self
._headers
_buffer
= []
277 self
._headers
_buffer
.append(f
'{keyword}: {value}\r\n'.encode())
280 def validate_and_send(rh
, req
):
285 class TestRequestHandlerBase
:
287 def setup_class(cls
):
288 cls
.http_httpd
= http
.server
.ThreadingHTTPServer(
289 ('127.0.0.1', 0), HTTPTestRequestHandler
)
290 cls
.http_port
= http_server_port(cls
.http_httpd
)
291 cls
.http_server_thread
= threading
.Thread(target
=cls
.http_httpd
.serve_forever
)
292 # FIXME: we should probably stop the http server thread after each test
293 # See: https://github.com/yt-dlp/yt-dlp/pull/7094#discussion_r1199746041
294 cls
.http_server_thread
.daemon
= True
295 cls
.http_server_thread
.start()
298 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
299 cls
.https_httpd
= http
.server
.ThreadingHTTPServer(
300 ('127.0.0.1', 0), HTTPTestRequestHandler
)
301 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
302 sslctx
.load_cert_chain(certfn
, None)
303 cls
.https_httpd
.socket
= sslctx
.wrap_socket(cls
.https_httpd
.socket
, server_side
=True)
304 cls
.https_port
= http_server_port(cls
.https_httpd
)
305 cls
.https_server_thread
= threading
.Thread(target
=cls
.https_httpd
.serve_forever
)
306 cls
.https_server_thread
.daemon
= True
307 cls
.https_server_thread
.start()
311 def handler(request
):
312 RH_KEY
= request
.param
313 if inspect
.isclass(RH_KEY
) and issubclass(RH_KEY
, RequestHandler
):
315 elif RH_KEY
in _REQUEST_HANDLERS
:
316 handler
= _REQUEST_HANDLERS
[RH_KEY
]
318 pytest
.skip(f
'{RH_KEY} request handler is not available')
320 return functools
.partial(handler
, logger
=FakeLogger
)
323 class TestHTTPRequestHandler(TestRequestHandlerBase
):
324 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
325 def test_verify_cert(self
, handler
):
326 with handler() as rh
:
327 with pytest
.raises(CertificateVerifyError
):
328 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
330 with handler(verify
=False) as rh
:
331 r
= validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
332 assert r
.status
== 200
335 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
336 def test_ssl_error(self
, handler
):
337 # HTTPS server with too old TLS version
338 # XXX: is there a better way to test this than to create a new server?
339 https_httpd
= http
.server
.ThreadingHTTPServer(
340 ('127.0.0.1', 0), HTTPTestRequestHandler
)
341 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
342 https_httpd
.socket
= sslctx
.wrap_socket(https_httpd
.socket
, server_side
=True)
343 https_port
= http_server_port(https_httpd
)
344 https_server_thread
= threading
.Thread(target
=https_httpd
.serve_forever
)
345 https_server_thread
.daemon
= True
346 https_server_thread
.start()
348 with handler(verify
=False) as rh
:
349 with pytest
.raises(SSLError
, match
='sslv3 alert handshake failure') as exc_info
:
350 validate_and_send(rh
, Request(f
'https://127.0.0.1:{https_port}/headers'))
351 assert not issubclass(exc_info
.type, CertificateVerifyError
)
353 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
354 def test_percent_encode(self
, handler
):
355 with handler() as rh
:
356 # Unicode characters should be encoded with uppercase percent-encoding
357 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/中文.html'))
358 assert res
.status
== 200
360 # don't normalize existing percent encodings
361 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/%c7%9f'))
362 assert res
.status
== 200
365 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
366 def test_remove_dot_segments(self
, handler
):
367 with handler() as rh
:
368 # This isn't a comprehensive test,
369 # but it should be enough to check whether the handler is removing dot segments
370 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/a/b/./../../headers'))
371 assert res
.status
== 200
372 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
375 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_dotsegments'))
376 assert res
.status
== 200
377 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/headers'
380 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
381 def test_unicode_path_redirection(self
, handler
):
382 with handler() as rh
:
383 r
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
384 assert r
.url
== f
'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
387 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
388 def test_raise_http_error(self
, handler
):
389 with handler() as rh
:
390 for bad_status
in (400, 500, 599, 302):
391 with pytest
.raises(HTTPError
):
392 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_%d' % (self
.http_port
, bad_status
)))
394 # Should not raise an error
395 validate_and_send(rh
, Request('http://127.0.0.1:%d/gen_200' % self
.http_port
)).close()
397 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
398 def test_response_url(self
, handler
):
399 with handler() as rh
:
400 # Response url should be that of the last url in redirect chain
401 res
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_301'))
402 assert res
.url
== f
'http://127.0.0.1:{self.http_port}/method'
404 res2
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_200'))
405 assert res2
.url
== f
'http://127.0.0.1:{self.http_port}/gen_200'
408 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
409 def test_redirect(self
, handler
):
410 with handler() as rh
:
411 def do_req(redirect_status
, method
, assert_no_content
=False):
412 data
= b
'testdata' if method
in ('POST', 'PUT') else None
413 res
= validate_and_send(
414 rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method
=method
, data
=data
))
419 data_sent
+= res
.read(len(data
))
420 if data_sent
!= data
:
424 headers
+= res
.read()
426 if assert_no_content
or data
is None:
427 assert b
'Content-Type' not in headers
428 assert b
'Content-Length' not in headers
430 assert b
'Content-Type' in headers
431 assert b
'Content-Length' in headers
433 return data_sent
.decode(), res
.headers
.get('method', '')
435 # A 303 must either use GET or HEAD for subsequent request
436 assert do_req(303, 'POST', True) == ('', 'GET')
437 assert do_req(303, 'HEAD') == ('', 'HEAD')
439 assert do_req(303, 'PUT', True) == ('', 'GET')
441 # 301 and 302 turn POST only into a GET
442 assert do_req(301, 'POST', True) == ('', 'GET')
443 assert do_req(301, 'HEAD') == ('', 'HEAD')
444 assert do_req(302, 'POST', True) == ('', 'GET')
445 assert do_req(302, 'HEAD') == ('', 'HEAD')
447 assert do_req(301, 'PUT') == ('testdata', 'PUT')
448 assert do_req(302, 'PUT') == ('testdata', 'PUT')
450 # 307 and 308 should not change method
451 for m
in ('POST', 'PUT'):
452 assert do_req(307, m
) == ('testdata', m
)
453 assert do_req(308, m
) == ('testdata', m
)
455 assert do_req(307, 'HEAD') == ('', 'HEAD')
456 assert do_req(308, 'HEAD') == ('', 'HEAD')
458 # These should not redirect and instead raise an HTTPError
459 for code
in (300, 304, 305, 306):
460 with pytest
.raises(HTTPError
):
463 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
464 def test_request_cookie_header(self
, handler
):
465 # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
466 with handler() as rh
:
467 # Specified Cookie header should be used
468 res
= validate_and_send(
470 f
'http://127.0.0.1:{self.http_port}/headers',
471 headers
={'Cookie': 'test=test'}
)).read().decode()
472 assert 'Cookie: test=test' in res
474 # Specified Cookie header should be removed on any redirect
475 res
= validate_and_send(
477 f
'http://127.0.0.1:{self.http_port}/308-to-headers',
478 headers
={'Cookie': 'test=test'}
)).read().decode()
479 assert 'Cookie: test=test' not in res
481 # Specified Cookie header should override global cookiejar for that request
482 cookiejar
= YoutubeDLCookieJar()
483 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
484 version
=0, name
='test', value
='ytdlp', port
=None, port_specified
=False,
485 domain
='127.0.0.1', domain_specified
=True, domain_initial_dot
=False, path
='/',
486 path_specified
=True, secure
=False, expires
=None, discard
=False, comment
=None,
487 comment_url
=None, rest
={}))
489 with handler(cookiejar
=cookiejar
) as rh
:
490 data
= validate_and_send(
491 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', headers
={'cookie': 'test=test'}
)).read()
492 assert b
'Cookie: test=ytdlp' not in data
493 assert b
'Cookie: test=test' in data
495 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
496 def test_redirect_loop(self
, handler
):
497 with handler() as rh
:
498 with pytest
.raises(HTTPError
, match
='redirect loop'):
499 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/redirect_loop'))
501 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
502 def test_incompleteread(self
, handler
):
503 with handler(timeout
=2) as rh
:
504 with pytest
.raises(IncompleteRead
):
505 validate_and_send(rh
, Request('http://127.0.0.1:%d/incompleteread' % self
.http_port
)).read()
507 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
508 def test_cookies(self
, handler
):
509 cookiejar
= YoutubeDLCookieJar()
510 cookiejar
.set_cookie(http
.cookiejar
.Cookie(
511 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
512 False, '/headers', True, False, None, False, None, None, {}))
514 with handler(cookiejar
=cookiejar
) as rh
:
515 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
516 assert b
'Cookie: test=ytdlp' in data
519 with handler() as rh
:
520 data
= validate_and_send(
521 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers', extensions
={'cookiejar': cookiejar}
)).read()
522 assert b
'Cookie: test=ytdlp' in data
524 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
525 def test_headers(self
, handler
):
527 with handler(headers
=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'}
)) as rh
:
529 data
= validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers')).read()
530 assert b
'Test1: test' in data
532 # Per request headers, merged with global
533 data
= validate_and_send(rh
, Request(
534 f
'http://127.0.0.1:{self.http_port}/headers', headers
={'test2': 'changed', 'test3': 'test3'}
)).read()
535 assert b
'Test1: test' in data
536 assert b
'Test2: changed' in data
537 assert b
'Test2: test2' not in data
538 assert b
'Test3: test3' in data
540 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
541 def test_timeout(self
, handler
):
542 with handler() as rh
:
543 # Default timeout is 20 seconds, so this should go through
545 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_3'))
547 with handler(timeout
=0.5) as rh
:
548 with pytest
.raises(TransportError
):
550 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1'))
552 # Per request timeout, should override handler timeout
554 rh
, Request(f
'http://127.0.0.1:{self.http_port}/timeout_1', extensions
={'timeout': 4}
))
556 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
557 def test_source_address(self
, handler
):
558 source_address
= f
'127.0.0.{random.randint(5, 255)}'
559 with handler(source_address
=source_address
) as rh
:
560 data
= validate_and_send(
561 rh
, Request(f
'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
562 assert source_address
== data
564 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
565 def test_gzip_trailing_garbage(self
, handler
):
566 with handler() as rh
:
567 data
= validate_and_send(rh
, Request(f
'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
568 assert data
== '<html><video src="/vid.mp4" /></html>'
570 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
571 @pytest.mark.skipif(not brotli
, reason
='brotli support is not installed')
572 def test_brotli(self
, handler
):
573 with handler() as rh
:
574 res
= validate_and_send(
576 f
'http://127.0.0.1:{self.http_port}/content-encoding',
577 headers
={'ytdl-encoding': 'br'}
))
578 assert res
.headers
.get('Content-Encoding') == 'br'
579 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
581 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
582 def test_deflate(self
, handler
):
583 with handler() as rh
:
584 res
= validate_and_send(
586 f
'http://127.0.0.1:{self.http_port}/content-encoding',
587 headers
={'ytdl-encoding': 'deflate'}
))
588 assert res
.headers
.get('Content-Encoding') == 'deflate'
589 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
591 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
592 def test_gzip(self
, handler
):
593 with handler() as rh
:
594 res
= validate_and_send(
596 f
'http://127.0.0.1:{self.http_port}/content-encoding',
597 headers
={'ytdl-encoding': 'gzip'}
))
598 assert res
.headers
.get('Content-Encoding') == 'gzip'
599 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
601 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
602 def test_multiple_encodings(self
, handler
):
603 with handler() as rh
:
604 for pair
in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
605 res
= validate_and_send(
607 f
'http://127.0.0.1:{self.http_port}/content-encoding',
608 headers
={'ytdl-encoding': pair}
))
609 assert res
.headers
.get('Content-Encoding') == pair
610 assert res
.read() == b
'<html><video src="/vid.mp4" /></html>'
612 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
613 def test_unsupported_encoding(self
, handler
):
614 with handler() as rh
:
615 res
= validate_and_send(
617 f
'http://127.0.0.1:{self.http_port}/content-encoding',
618 headers
={'ytdl-encoding': 'unsupported'}
))
619 assert res
.headers
.get('Content-Encoding') == 'unsupported'
620 assert res
.read() == b
'raw'
622 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
623 def test_read(self
, handler
):
624 with handler() as rh
:
625 res
= validate_and_send(
626 rh
, Request(f
'http://127.0.0.1:{self.http_port}/headers'))
627 assert res
.readable()
628 assert res
.read(1) == b
'H'
629 assert res
.read(3) == b
'ost'
632 class TestHTTPProxy(TestRequestHandlerBase
):
634 def setup_class(cls
):
635 super().setup_class()
637 cls
.proxy
= http
.server
.ThreadingHTTPServer(
638 ('127.0.0.1', 0), _build_proxy_handler('normal'))
639 cls
.proxy_port
= http_server_port(cls
.proxy
)
640 cls
.proxy_thread
= threading
.Thread(target
=cls
.proxy
.serve_forever
)
641 cls
.proxy_thread
.daemon
= True
642 cls
.proxy_thread
.start()
645 cls
.geo_proxy
= http
.server
.ThreadingHTTPServer(
646 ('127.0.0.1', 0), _build_proxy_handler('geo'))
647 cls
.geo_port
= http_server_port(cls
.geo_proxy
)
648 cls
.geo_proxy_thread
= threading
.Thread(target
=cls
.geo_proxy
.serve_forever
)
649 cls
.geo_proxy_thread
.daemon
= True
650 cls
.geo_proxy_thread
.start()
652 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
653 def test_http_proxy(self
, handler
):
654 http_proxy
= f
'http://127.0.0.1:{self.proxy_port}'
655 geo_proxy
= f
'http://127.0.0.1:{self.geo_port}'
657 # Test global http proxy
658 # Test per request http proxy
659 # Test per request http proxy disables proxy
660 url
= 'http://foo.com/bar'
663 with handler(proxies
={'http': http_proxy}
) as rh
:
664 res
= validate_and_send(rh
, Request(url
)).read().decode()
665 assert res
== f
'normal: {url}'
667 # Per request proxy overrides global
668 res
= validate_and_send(rh
, Request(url
, proxies
={'http': geo_proxy}
)).read().decode()
669 assert res
== f
'geo: {url}'
671 # and setting to None disables all proxies for that request
672 real_url
= f
'http://127.0.0.1:{self.http_port}/headers'
673 res
= validate_and_send(
674 rh
, Request(real_url
, proxies
={'http': None}
)).read().decode()
675 assert res
!= f
'normal: {real_url}'
676 assert 'Accept' in res
678 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
679 def test_noproxy(self
, handler
):
680 with handler(proxies
={'proxy': f'http://127.0.0.1:{self.proxy_port}
'}) as rh:
682 for no_proxy in (f'127.0.0.1:{self.http_port}
', '127.0.0.1', 'localhost
'):
683 nop_response = validate_and_send(
684 rh, Request(f'http
://127.0.0.1:{self.http_port}
/headers
', proxies={'no': no_proxy})).read().decode(
686 assert 'Accept
' in nop_response
688 @pytest.mark.parametrize('handler
', ['Urllib
'], indirect=True)
689 def test_allproxy(self, handler):
690 url = 'http
://foo
.com
/bar
'
691 with handler() as rh:
692 response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
694 assert response
== f
'normal: {url}'
696 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
697 def test_http_proxy_with_idn(self
, handler
):
698 with handler(proxies
={
699 'http': f
'http://127.0.0.1:{self.proxy_port}',
701 url
= 'http://中文.tw/'
702 response
= rh
.send(Request(url
)).read().decode()
703 # b'xn--fiq228c' is '中文'.encode('idna')
704 assert response
== 'normal: http://xn--fiq228c.tw/'
707 class TestClientCertificate
:
710 def setup_class(cls
):
711 certfn
= os
.path
.join(TEST_DIR
, 'testcert.pem')
712 cls
.certdir
= os
.path
.join(TEST_DIR
, 'testdata', 'certificate')
713 cacertfn
= os
.path
.join(cls
.certdir
, 'ca.crt')
714 cls
.httpd
= http
.server
.ThreadingHTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler
)
715 sslctx
= ssl
.SSLContext(ssl
.PROTOCOL_TLS_SERVER
)
716 sslctx
.verify_mode
= ssl
.CERT_REQUIRED
717 sslctx
.load_verify_locations(cafile
=cacertfn
)
718 sslctx
.load_cert_chain(certfn
, None)
719 cls
.httpd
.socket
= sslctx
.wrap_socket(cls
.httpd
.socket
, server_side
=True)
720 cls
.port
= http_server_port(cls
.httpd
)
721 cls
.server_thread
= threading
.Thread(target
=cls
.httpd
.serve_forever
)
722 cls
.server_thread
.daemon
= True
723 cls
.server_thread
.start()
725 def _run_test(self
, handler
, **handler_kwargs
):
727 # Disable client-side validation of unacceptable self-signed testcert.pem
728 # The test is of a check on the server side, so unaffected
732 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.port}/video.html')).read().decode()
734 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
735 def test_certificate_combined_nopass(self
, handler
):
736 self
._run
_test
(handler
, client_cert
={
737 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithkey.crt'),
740 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
741 def test_certificate_nocombined_nopass(self
, handler
):
742 self
._run
_test
(handler
, client_cert
={
743 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
744 'client_certificate_key': os
.path
.join(self
.certdir
, 'client.key'),
747 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
748 def test_certificate_combined_pass(self
, handler
):
749 self
._run
_test
(handler
, client_cert
={
750 'client_certificate': os
.path
.join(self
.certdir
, 'clientwithencryptedkey.crt'),
751 'client_certificate_password': 'foobar',
754 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
755 def test_certificate_nocombined_pass(self
, handler
):
756 self
._run
_test
(handler
, client_cert
={
757 'client_certificate': os
.path
.join(self
.certdir
, 'client.crt'),
758 'client_certificate_key': os
.path
.join(self
.certdir
, 'clientencrypted.key'),
759 'client_certificate_password': 'foobar',
763 class TestUrllibRequestHandler(TestRequestHandlerBase
):
764 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
765 def test_file_urls(self
, handler
):
766 # See https://github.com/ytdl-org/youtube-dl/issues/8227
767 tf
= tempfile
.NamedTemporaryFile(delete
=False)
770 req
= Request(pathlib
.Path(tf
.name
).as_uri())
771 with handler() as rh
:
772 with pytest
.raises(UnsupportedRequest
):
775 # Test that urllib never loaded FileHandler
776 with pytest
.raises(TransportError
):
779 with handler(enable_file_urls
=True) as rh
:
780 res
= validate_and_send(rh
, req
)
781 assert res
.read() == b
'foobar'
786 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
787 def test_http_error_returns_content(self
, handler
):
788 # urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
790 with handler() as rh
:
793 validate_and_send(rh
, Request(f
'http://127.0.0.1:{self.http_port}/gen_404'))
794 except HTTPError
as e
:
797 assert get_response().read() == b
'<html></html>'
799 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
800 def test_verify_cert_error_text(self
, handler
):
801 # Check the output of the error message
802 with handler() as rh
:
804 CertificateVerifyError
,
805 match
=r
'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate'
807 validate_and_send(rh
, Request(f
'https://127.0.0.1:{self.https_port}/headers'))
809 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
810 @pytest.mark.parametrize('req,match,version_check', [
811 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
812 # bpo-39603: Check implemented in 3.7.9+, 3.8.5+
814 Request('http://127.0.0.1', method
='GET\n'),
815 'method can\'t contain control characters',
816 lambda v
: v
< (3, 7, 9) or (3, 8, 0) <= v
< (3, 8, 5)
818 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
819 # bpo-38576: Check implemented in 3.7.8+, 3.8.3+
821 Request('http://127.0.0. 1', method
='GET'),
822 'URL can\'t contain control characters',
823 lambda v
: v
< (3, 7, 8) or (3, 8, 0) <= v
< (3, 8, 3)
825 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
826 (Request('http://127.0.0.1', headers
={'foo\n': 'bar'}
), 'Invalid header name', None),
828 def test_httplib_validation_errors(self
, handler
, req
, match
, version_check
):
829 if version_check
and version_check(sys
.version_info
):
830 pytest
.skip(f
'Python {sys.version} version does not have the required validation for this test.')
832 with handler() as rh
:
833 with pytest
.raises(RequestError
, match
=match
) as exc_info
:
834 validate_and_send(rh
, req
)
835 assert not isinstance(exc_info
.value
, TransportError
)
838 def run_validation(handler
, error
, req
, **handler_kwargs
):
839 with handler(**handler_kwargs
) as rh
:
841 with pytest
.raises(error
):
847 class TestRequestHandlerValidation
:
849 class ValidationRH(RequestHandler
):
850 def _send(self
, request
):
851 raise RequestError('test')
853 class NoCheckRH(ValidationRH
):
854 _SUPPORTED_FEATURES
= None
855 _SUPPORTED_PROXY_SCHEMES
= None
856 _SUPPORTED_URL_SCHEMES
= None
858 def _check_extensions(self
, extensions
):
861 class HTTPSupportedRH(ValidationRH
):
862 _SUPPORTED_URL_SCHEMES
= ('http',)
865 # scheme, expected to fail, handler kwargs
868 ('https', False, {}),
871 ('file', UnsupportedRequest
, {}),
872 ('file', False, {'enable_file_urls': True}
),
874 (NoCheckRH
, [('http', False, {})]),
875 (ValidationRH
, [('http', UnsupportedRequest
, {})])
878 PROXY_SCHEME_TESTS
= [
879 # scheme, expected to fail
882 ('https', UnsupportedRequest
),
887 ('socks', UnsupportedRequest
),
889 (NoCheckRH
, [('http', False)]),
890 (HTTPSupportedRH
, [('http', UnsupportedRequest
)]),
894 # key, expected to fail
897 ('unrelated', False),
899 (NoCheckRH
, [('all', False)]),
900 (HTTPSupportedRH
, [('all', UnsupportedRequest
)]),
901 (HTTPSupportedRH
, [('no', UnsupportedRequest
)]),
906 ({'cookiejar': 'notacookiejar'}
, AssertionError),
907 ({'cookiejar': YoutubeDLCookieJar()}
, False),
908 ({'cookiejar': CookieJar()}
, AssertionError),
909 ({'timeout': 1}
, False),
910 ({'timeout': 'notatimeout'}
, AssertionError),
911 ({'unsupported': 'value'}
, UnsupportedRequest
),
914 ({'cookiejar': 'notacookiejar'}
, False),
915 ({'somerandom': 'test'}
, False), # but any extension is allowed through
919 @pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
920 (handler_tests
[0], scheme
, fail
, handler_kwargs
)
921 for handler_tests
in URL_SCHEME_TESTS
922 for scheme
, fail
, handler_kwargs
in handler_tests
[1]
924 ], indirect
=['handler'])
925 def test_url_scheme(self
, handler
, scheme
, fail
, handler_kwargs
):
926 run_validation(handler
, fail
, Request(f
'{scheme}://'), **(handler_kwargs
or {}))
928 @pytest.mark.parametrize('handler,fail', [('Urllib', False)], indirect
=['handler'])
929 def test_no_proxy(self
, handler
, fail
):
930 run_validation(handler
, fail
, Request('http://', proxies
={'no': '127.0.0.1,github.com'}
))
931 run_validation(handler
, fail
, Request('http://'), proxies
={'no': '127.0.0.1,github.com'}
)
933 @pytest.mark.parametrize('handler,proxy_key,fail', [
934 (handler_tests
[0], proxy_key
, fail
)
935 for handler_tests
in PROXY_KEY_TESTS
936 for proxy_key
, fail
in handler_tests
[1]
937 ], indirect
=['handler'])
938 def test_proxy_key(self
, handler
, proxy_key
, fail
):
939 run_validation(handler
, fail
, Request('http://', proxies
={proxy_key: 'http://example.com'}
))
940 run_validation(handler
, fail
, Request('http://'), proxies
={proxy_key: 'http://example.com'}
)
942 @pytest.mark.parametrize('handler,scheme,fail', [
943 (handler_tests
[0], scheme
, fail
)
944 for handler_tests
in PROXY_SCHEME_TESTS
945 for scheme
, fail
in handler_tests
[1]
946 ], indirect
=['handler'])
947 def test_proxy_scheme(self
, handler
, scheme
, fail
):
948 run_validation(handler
, fail
, Request('http://', proxies
={'http': f'{scheme}
://example
.com
'}))
949 run_validation(handler, fail, Request('http
://'), proxies={'http': f'{scheme}://example.com'})
951 @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH
], indirect
=True)
952 def test_empty_proxy(self
, handler
):
953 run_validation(handler
, False, Request('http://', proxies
={'http': None}
))
954 run_validation(handler
, False, Request('http://'), proxies
={'http': None}
)
956 @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
957 @pytest.mark.parametrize('handler', ['Urllib'], indirect
=True)
958 def test_invalid_proxy_url(self
, handler
, proxy_url
):
959 run_validation(handler
, UnsupportedRequest
, Request('http://', proxies
={'http': proxy_url}
))
961 @pytest.mark.parametrize('handler,extensions,fail', [
962 (handler_tests
[0], extensions
, fail
)
963 for handler_tests
in EXTENSION_TESTS
964 for extensions
, fail
in handler_tests
[1]
965 ], indirect
=['handler'])
966 def test_extension(self
, handler
, extensions
, fail
):
968 handler
, fail
, Request('http://', extensions
=extensions
))
970 def test_invalid_request_type(self
):
971 rh
= self
.ValidationRH(logger
=FakeLogger())
972 for method
in (rh
.validate
, rh
.send
):
973 with pytest
.raises(TypeError, match
='Expected an instance of Request'):
974 method('not a request')
977 class FakeResponse(Response
):
978 def __init__(self
, request
):
979 # XXX: we could make request part of standard response interface
980 self
.request
= request
981 super().__init
__(fp
=io
.BytesIO(b
''), headers
={}, url
=request
.url
)
984 class FakeRH(RequestHandler
):
986 def _validate(self
, request
):
989 def _send(self
, request
: Request
):
990 if request
.url
.startswith('ssl://'):
991 raise SSLError(request
.url
[len('ssl://'):])
992 return FakeResponse(request
)
995 class FakeRHYDL(FakeYDL
):
996 def __init__(self
, *args
, **kwargs
):
997 super().__init
__(*args
, **kwargs
)
998 self
._request
_director
= self
.build_request_director([FakeRH
])
1001 class TestRequestDirector
:
1003 def test_handler_operations(self
):
1004 director
= RequestDirector(logger
=FakeLogger())
1005 handler
= FakeRH(logger
=FakeLogger())
1006 director
.add_handler(handler
)
1007 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler
1009 # Handler should overwrite
1010 handler2
= FakeRH(logger
=FakeLogger())
1011 director
.add_handler(handler2
)
1012 assert director
.handlers
.get(FakeRH
.RH_KEY
) is not handler
1013 assert director
.handlers
.get(FakeRH
.RH_KEY
) is handler2
1014 assert len(director
.handlers
) == 1
1016 class AnotherFakeRH(FakeRH
):
1018 director
.add_handler(AnotherFakeRH(logger
=FakeLogger()))
1019 assert len(director
.handlers
) == 2
1020 assert director
.handlers
.get(AnotherFakeRH
.RH_KEY
).RH_KEY
== AnotherFakeRH
.RH_KEY
1022 director
.handlers
.pop(FakeRH
.RH_KEY
, None)
1023 assert director
.handlers
.get(FakeRH
.RH_KEY
) is None
1024 assert len(director
.handlers
) == 1
1026 # RequestErrors should passthrough
1027 with pytest
.raises(SSLError
):
1028 director
.send(Request('ssl://something'))
1030 def test_send(self
):
1031 director
= RequestDirector(logger
=FakeLogger())
1032 with pytest
.raises(RequestError
):
1033 director
.send(Request('any://'))
1034 director
.add_handler(FakeRH(logger
=FakeLogger()))
1035 assert isinstance(director
.send(Request('http://')), FakeResponse
)
1037 def test_unsupported_handlers(self
):
1038 director
= RequestDirector(logger
=FakeLogger())
1039 director
.add_handler(FakeRH(logger
=FakeLogger()))
1041 class SupportedRH(RequestHandler
):
1042 _SUPPORTED_URL_SCHEMES
= ['http']
1044 def _send(self
, request
: Request
):
1045 return Response(fp
=io
.BytesIO(b
'supported'), headers
={}, url
=request
.url
)
1047 # This handler should by default take preference over FakeRH
1048 director
.add_handler(SupportedRH(logger
=FakeLogger()))
1049 assert director
.send(Request('http://')).read() == b
'supported'
1050 assert director
.send(Request('any://')).read() == b
''
1052 director
.handlers
.pop(FakeRH
.RH_KEY
)
1053 with pytest
.raises(NoSupportingHandlers
):
1054 director
.send(Request('any://'))
1056 def test_unexpected_error(self
):
1057 director
= RequestDirector(logger
=FakeLogger())
1059 class UnexpectedRH(FakeRH
):
1060 def _send(self
, request
: Request
):
1061 raise TypeError('something')
1063 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1064 with pytest
.raises(NoSupportingHandlers
, match
=r
'1 unexpected error'):
1065 director
.send(Request('any://'))
1067 director
.handlers
.clear()
1068 assert len(director
.handlers
) == 0
1070 # Should not be fatal
1071 director
.add_handler(FakeRH(logger
=FakeLogger()))
1072 director
.add_handler(UnexpectedRH(logger
=FakeLogger
))
1073 assert director
.send(Request('any://'))
1076 # XXX: do we want to move this to test_YoutubeDL.py?
1077 class TestYoutubeDLNetworking
:
1080 def build_handler(ydl
, handler
: RequestHandler
= FakeRH
):
1081 return ydl
.build_request_director([handler
]).handlers
.get(handler
.RH_KEY
)
1083 def test_compat_opener(self
):
1084 with FakeYDL() as ydl
:
1085 with warnings
.catch_warnings():
1086 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1087 assert isinstance(ydl
._opener
, urllib
.request
.OpenerDirector
)
1089 @pytest.mark.parametrize('proxy,expected', [
1090 ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}
),
1091 ('', {'all': '__noproxy__'}
),
1092 (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}
) # env, set https
1094 def test_proxy(self
, proxy
, expected
):
1095 old_http_proxy
= os
.environ
.get('HTTP_PROXY')
1097 os
.environ
['HTTP_PROXY'] = 'http://127.0.0.1:8081' # ensure that provided proxies override env
1098 with FakeYDL({'proxy': proxy}
) as ydl
:
1099 assert ydl
.proxies
== expected
1102 os
.environ
['HTTP_PROXY'] = old_http_proxy
1104 def test_compat_request(self
):
1105 with FakeRHYDL() as ydl
:
1106 assert ydl
.urlopen('test://')
1107 urllib_req
= urllib
.request
.Request('http://foo.bar', data
=b
'test', method
='PUT', headers
={'X-Test': '1'}
)
1108 urllib_req
.add_unredirected_header('Cookie', 'bob=bob')
1109 urllib_req
.timeout
= 2
1110 with warnings
.catch_warnings():
1111 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1112 req
= ydl
.urlopen(urllib_req
).request
1113 assert req
.url
== urllib_req
.get_full_url()
1114 assert req
.data
== urllib_req
.data
1115 assert req
.method
== urllib_req
.get_method()
1116 assert 'X-Test' in req
.headers
1117 assert 'Cookie' in req
.headers
1118 assert req
.extensions
.get('timeout') == 2
1120 with pytest
.raises(AssertionError):
1123 def test_extract_basic_auth(self
):
1124 with FakeRHYDL() as ydl
:
1125 res
= ydl
.urlopen(Request('http://user:pass@foo.bar'))
1126 assert res
.request
.headers
['Authorization'] == 'Basic dXNlcjpwYXNz'
1128 def test_sanitize_url(self
):
1129 with FakeRHYDL() as ydl
:
1130 res
= ydl
.urlopen(Request('httpss://foo.bar'))
1131 assert res
.request
.url
== 'https://foo.bar'
1133 def test_file_urls_error(self
):
1134 # use urllib handler
1135 with FakeYDL() as ydl
:
1136 with pytest
.raises(RequestError
, match
=r
'file:// URLs are disabled by default'):
1137 ydl
.urlopen('file://')
1139 def test_legacy_server_connect_error(self
):
1140 with FakeRHYDL() as ydl
:
1141 for error
in ('UNSAFE_LEGACY_RENEGOTIATION_DISABLED', 'SSLV3_ALERT_HANDSHAKE_FAILURE'):
1142 with pytest
.raises(RequestError
, match
=r
'Try using --legacy-server-connect'):
1143 ydl
.urlopen(f
'ssl://{error}')
1145 with pytest
.raises(SSLError
, match
='testerror'):
1146 ydl
.urlopen('ssl://testerror')
1148 @pytest.mark.parametrize('proxy_key,proxy_url,expected', [
1149 ('http', '__noproxy__', None),
1150 ('no', '127.0.0.1,foo.bar', '127.0.0.1,foo.bar'),
1151 ('https', 'example.com', 'http://example.com'),
1152 ('https', '//example.com', 'http://example.com'),
1153 ('https', 'socks5://example.com', 'socks5h://example.com'),
1154 ('http', 'socks://example.com', 'socks4://example.com'),
1155 ('http', 'socks4://example.com', 'socks4://example.com'),
1156 ('unrelated', '/bad/proxy', '/bad/proxy'), # clean_proxies should ignore bad proxies
1158 def test_clean_proxy(self
, proxy_key
, proxy_url
, expected
):
1159 # proxies should be cleaned in urlopen()
1160 with FakeRHYDL() as ydl
:
1161 req
= ydl
.urlopen(Request('test://', proxies
={proxy_key: proxy_url}
)).request
1162 assert req
.proxies
[proxy_key
] == expected
1164 # and should also be cleaned when building the handler
1165 env_key
= f
'{proxy_key.upper()}_PROXY'
1166 old_env_proxy
= os
.environ
.get(env_key
)
1168 os
.environ
[env_key
] = proxy_url
# ensure that provided proxies override env
1169 with FakeYDL() as ydl
:
1170 rh
= self
.build_handler(ydl
)
1171 assert rh
.proxies
[proxy_key
] == expected
1174 os
.environ
[env_key
] = old_env_proxy
1176 def test_clean_proxy_header(self
):
1177 with FakeRHYDL() as ydl
:
1178 req
= ydl
.urlopen(Request('test://', headers
={'ytdl-request-proxy': '//foo.bar'}
)).request
1179 assert 'ytdl-request-proxy' not in req
.headers
1180 assert req
.proxies
== {'all': 'http://foo.bar'}
1182 with FakeYDL({'http_headers': {'ytdl-request-proxy': '//foo.bar'}
}) as ydl
:
1183 rh
= self
.build_handler(ydl
)
1184 assert 'ytdl-request-proxy' not in rh
.headers
1185 assert rh
.proxies
== {'all': 'http://foo.bar'}
1187 def test_clean_header(self
):
1188 with FakeRHYDL() as ydl
:
1189 res
= ydl
.urlopen(Request('test://', headers
={'Youtubedl-no-compression': True}
))
1190 assert 'Youtubedl-no-compression' not in res
.request
.headers
1191 assert res
.request
.headers
.get('Accept-Encoding') == 'identity'
1193 with FakeYDL({'http_headers': {'Youtubedl-no-compression': True}
}) as ydl
:
1194 rh
= self
.build_handler(ydl
)
1195 assert 'Youtubedl-no-compression' not in rh
.headers
1196 assert rh
.headers
.get('Accept-Encoding') == 'identity'
1198 def test_build_handler_params(self
):
1200 'http_headers': {'test': 'testtest'}
,
1201 'socket_timeout': 2,
1202 'proxy': 'http://127.0.0.1:8080',
1203 'source_address': '127.0.0.45',
1204 'debug_printtraffic': True,
1205 'compat_opts': ['no-certifi'],
1206 'nocheckcertificate': True,
1207 'legacyserverconnect': True,
1209 rh
= self
.build_handler(ydl
)
1210 assert rh
.headers
.get('test') == 'testtest'
1211 assert 'Accept' in rh
.headers
# ensure std_headers are still there
1212 assert rh
.timeout
== 2
1213 assert rh
.proxies
.get('all') == 'http://127.0.0.1:8080'
1214 assert rh
.source_address
== '127.0.0.45'
1215 assert rh
.verbose
is True
1216 assert rh
.prefer_system_certs
is True
1217 assert rh
.verify
is False
1218 assert rh
.legacy_ssl_support
is True
1220 @pytest.mark.parametrize('ydl_params', [
1221 {'client_certificate': 'fakecert.crt'}
,
1222 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key'}
,
1223 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1224 {'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'}
,
1226 def test_client_certificate(self
, ydl_params
):
1227 with FakeYDL(ydl_params
) as ydl
:
1228 rh
= self
.build_handler(ydl
)
1229 assert rh
._client
_cert
== ydl_params
# XXX: Too bound to implementation
1231 def test_urllib_file_urls(self
):
1232 with FakeYDL({'enable_file_urls': False}
) as ydl
:
1233 rh
= self
.build_handler(ydl
, UrllibRH
)
1234 assert rh
.enable_file_urls
is False
1236 with FakeYDL({'enable_file_urls': True}
) as ydl
:
1237 rh
= self
.build_handler(ydl
, UrllibRH
)
1238 assert rh
.enable_file_urls
is True
1243 def test_query(self
):
1244 req
= Request('http://example.com?q=something', query
={'v': 'xyz'}
)
1245 assert req
.url
== 'http://example.com?q=something&v=xyz'
1247 req
.update(query
={'v': '123'}
)
1248 assert req
.url
== 'http://example.com?q=something&v=123'
1249 req
.update(url
='http://example.com', query
={'v': 'xyz'}
)
1250 assert req
.url
== 'http://example.com?v=xyz'
1252 def test_method(self
):
1253 req
= Request('http://example.com')
1254 assert req
.method
== 'GET'
1256 assert req
.method
== 'POST'
1258 assert req
.method
== 'GET'
1261 assert req
.method
== 'PUT'
1263 assert req
.method
== 'PUT'
1264 with pytest
.raises(TypeError):
1267 def test_request_helpers(self
):
1268 assert HEADRequest('http://example.com').method
== 'HEAD'
1269 assert PUTRequest('http://example.com').method
== 'PUT'
1271 def test_headers(self
):
1272 req
= Request('http://example.com', headers
={'tesT': 'test'}
)
1273 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1274 req
.update(headers
={'teSt2': 'test2'}
)
1275 assert req
.headers
== HTTPHeaderDict({'test': 'test', 'test2': 'test2'}
)
1277 req
.headers
= new_headers
= HTTPHeaderDict({'test': 'test'}
)
1278 assert req
.headers
== HTTPHeaderDict({'test': 'test'}
)
1279 assert req
.headers
is new_headers
1281 # test converts dict to case insensitive dict
1282 req
.headers
= new_headers
= {'test2': 'test2'}
1283 assert isinstance(req
.headers
, HTTPHeaderDict
)
1284 assert req
.headers
is not new_headers
1286 with pytest
.raises(TypeError):
1289 def test_data_type(self
):
1290 req
= Request('http://example.com')
1291 assert req
.data
is None
1292 # test bytes is allowed
1294 assert req
.data
== b
'test'
1295 # test iterable of bytes is allowed
1296 i
= [b
'test', b
'test2']
1298 assert req
.data
== i
1300 # test file-like object is allowed
1301 f
= io
.BytesIO(b
'test')
1303 assert req
.data
== f
1305 # common mistake: test str not allowed
1306 with pytest
.raises(TypeError):
1308 assert req
.data
!= 'test'
1310 # common mistake: test dict is not allowed
1311 with pytest
.raises(TypeError):
1312 req
.data
= {'test': 'test'}
1313 assert req
.data
!= {'test': 'test'}
1315 def test_content_length_header(self
):
1316 req
= Request('http://example.com', headers
={'Content-Length': '0'}
, data
=b
'')
1317 assert req
.headers
.get('Content-Length') == '0'
1320 assert 'Content-Length' not in req
.headers
1322 req
= Request('http://example.com', headers
={'Content-Length': '10'}
)
1323 assert 'Content-Length' not in req
.headers
1325 def test_content_type_header(self
):
1326 req
= Request('http://example.com', headers
={'Content-Type': 'test'}
, data
=b
'test')
1327 assert req
.headers
.get('Content-Type') == 'test'
1329 assert req
.headers
.get('Content-Type') == 'test'
1331 assert 'Content-Type' not in req
.headers
1333 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1335 def test_update_req(self
):
1336 req
= Request('http://example.com')
1337 assert req
.data
is None
1338 assert req
.method
== 'GET'
1339 assert 'Content-Type' not in req
.headers
1340 # Test that zero-byte payloads will be sent
1341 req
.update(data
=b
'')
1342 assert req
.data
== b
''
1343 assert req
.method
== 'POST'
1344 assert req
.headers
.get('Content-Type') == 'application/x-www-form-urlencoded'
1346 def test_proxies(self
):
1347 req
= Request(url
='http://example.com', proxies
={'http': 'http://127.0.0.1:8080'}
)
1348 assert req
.proxies
== {'http': 'http://127.0.0.1:8080'}
1350 def test_extensions(self
):
1351 req
= Request(url
='http://example.com', extensions
={'timeout': 2}
)
1352 assert req
.extensions
== {'timeout': 2}
1354 def test_copy(self
):
1356 url
='http://example.com',
1357 extensions
={'cookiejar': CookieJar()}
,
1358 headers
={'Accept-Encoding': 'br'}
,
1359 proxies
={'http': 'http://127.0.0.1'}
,
1362 req_copy
= req
.copy()
1363 assert req_copy
is not req
1364 assert req_copy
.url
== req
.url
1365 assert req_copy
.headers
== req
.headers
1366 assert req_copy
.headers
is not req
.headers
1367 assert req_copy
.proxies
== req
.proxies
1368 assert req_copy
.proxies
is not req
.proxies
1370 # Data is not able to be copied
1371 assert req_copy
.data
== req
.data
1372 assert req_copy
.data
is req
.data
1374 # Shallow copy extensions
1375 assert req_copy
.extensions
is not req
.extensions
1376 assert req_copy
.extensions
['cookiejar'] == req
.extensions
['cookiejar']
1378 # Subclasses are copied by default
1379 class AnotherRequest(Request
):
1382 req
= AnotherRequest(url
='http://127.0.0.1')
1383 assert isinstance(req
.copy(), AnotherRequest
)
1386 req
= Request(url
='https://фtest.example.com/ some spaceв?ä=c',)
1387 assert req
.url
== 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c'
1389 assert Request(url
='//example.com').url
== 'http://example.com'
1391 with pytest
.raises(TypeError):
1392 Request(url
='https://').url
= None
1397 @pytest.mark.parametrize('reason,status,expected', [
1398 ('custom', 200, 'custom'),
1399 (None, 404, 'Not Found'), # fallback status
1400 ('', 403, 'Forbidden'),
1403 def test_reason(self
, reason
, status
, expected
):
1404 res
= Response(io
.BytesIO(b
''), url
='test://', headers
={}, status
=status
, reason
=reason
)
1405 assert res
.reason
== expected
1407 def test_headers(self
):
1409 headers
.add_header('Test', 'test')
1410 headers
.add_header('Test', 'test2')
1411 headers
.add_header('content-encoding', 'br')
1412 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1413 assert res
.headers
.get_all('test') == ['test', 'test2']
1414 assert 'Content-Encoding' in res
.headers
1416 def test_get_header(self
):
1418 headers
.add_header('Set-Cookie', 'cookie1')
1419 headers
.add_header('Set-cookie', 'cookie2')
1420 headers
.add_header('Test', 'test')
1421 headers
.add_header('Test', 'test2')
1422 res
= Response(io
.BytesIO(b
''), headers
=headers
, url
='test://')
1423 assert res
.get_header('test') == 'test, test2'
1424 assert res
.get_header('set-Cookie') == 'cookie1'
1425 assert res
.get_header('notexist', 'default') == 'default'
1427 def test_compat(self
):
1428 res
= Response(io
.BytesIO(b
''), url
='test://', status
=404, headers
={'test': 'test'}
)
1429 with warnings
.catch_warnings():
1430 warnings
.simplefilter('ignore', category
=DeprecationWarning)
1431 assert res
.code
== res
.getcode() == res
.status
1432 assert res
.geturl() == res
.url
1433 assert res
.info() is res
.headers
1434 assert res
.getheader('test') == res
.get_header('test')