]> jfr.im git - yt-dlp.git/blame - test/test_networking.py
projects / yt-dlp.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
Bugfix for beaa1a44554d04d9fe63a743a5bb4431ca778f28 (#9235)
[yt-dlp.git] / test / test_networking.py
CommitLineData
cc52de43 1#!/usr/bin/env python3
54007a45 2
83fda3c0
PH		 3# Allow direct execution
4import os
5import sys
227bf1a3 6
7import pytest
f8271158 8
83fda3c0
PH		 9sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
10
08916a49 11import gzip
227bf1a3 12import http.client
08916a49 13import http.cookiejar
54007a45 14import http.server
08916a49 15import io
16import pathlib
227bf1a3 17import random
f8271158 18import ssl
08916a49 19import tempfile
f8271158 20import threading
227bf1a3 21import time
08916a49 22import urllib.error
ac668111 23import urllib.request
227bf1a3 24import warnings
daafbf49 25import zlib
227bf1a3 26from email.message import Message
27from http.cookiejar import CookieJar
f8271158 28
69d31914 29from test.helper import FakeYDL, http_server_port, verify_address_availability
6148833f 30from yt_dlp.cookies import YoutubeDLCookieJar
8a8b5452 31from yt_dlp.dependencies import brotli, requests, urllib3
227bf1a3 32from yt_dlp.networking import (
33 HEADRequest,
34 PUTRequest,
35 Request,
36 RequestDirector,
37 RequestHandler,
38 Response,
39)
40from yt_dlp.networking._urllib import UrllibRH
227bf1a3 41from yt_dlp.networking.exceptions import (
42 CertificateVerifyError,
43 HTTPError,
44 IncompleteRead,
45 NoSupportingHandlers,
8a8b5452 46 ProxyError,
227bf1a3 47 RequestError,
48 SSLError,
49 TransportError,
50 UnsupportedRequest,
51)
52from yt_dlp.utils._utils import _YDLLogger as FakeLogger
53from yt_dlp.utils.networking import HTTPHeaderDict
83fda3c0 54
ccfd70f4 55from test.conftest import validate_and_send
56
83fda3c0
PH		 57TEST_DIR = os.path.dirname(os.path.abspath(__file__))
58
03d8d4df 59
227bf1a3 60def _build_proxy_handler(name):
61 class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
62 proxy_name = name
63
64 def log_message(self, format, *args):
65 pass
66
67 def do_GET(self):
68 self.send_response(200)
69 self.send_header('Content-Type', 'text/plain; charset=utf-8')
70 self.end_headers()
71 self.wfile.write('{self.proxy_name}: {self.path}'.format(self=self).encode())
72 return HTTPTestRequestHandler
73
74
ac668111 75class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
08916a49 76 protocol_version = 'HTTP/1.1'
77
83fda3c0
PH		 78 def log_message(self, format, *args):
79 pass
80
08916a49 81 def _headers(self):
227bf1a3 82 payload = str(self.headers).encode()
08916a49 83 self.send_response(200)
84 self.send_header('Content-Type', 'application/json')
85 self.send_header('Content-Length', str(len(payload)))
86 self.end_headers()
87 self.wfile.write(payload)
88
89 def _redirect(self):
90 self.send_response(int(self.path[len('/redirect_'):]))
91 self.send_header('Location', '/method')
92 self.send_header('Content-Length', '0')
93 self.end_headers()
94
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)
99 self.end_headers()
100 if payload:
101 self.wfile.write(payload)
102
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)))
108 self.end_headers()
109 self.wfile.write(payload)
110
111 def _read_data(self):
112 if 'Content-Length' in self.headers:
113 return self.rfile.read(int(self.headers['Content-Length']))
114
115 def do_POST(self):
227bf1a3 116 data = self._read_data() + str(self.headers).encode()
08916a49 117 if self.path.startswith('/redirect_'):
118 self._redirect()
119 elif self.path.startswith('/method'):
120 self._method('POST', data)
121 elif self.path.startswith('/headers'):
122 self._headers()
123 else:
124 self._status(404)
125
126 def do_HEAD(self):
127 if self.path.startswith('/redirect_'):
128 self._redirect()
129 elif self.path.startswith('/method'):
130 self._method('HEAD')
131 else:
132 self._status(404)
133
134 def do_PUT(self):
227bf1a3 135 data = self._read_data() + str(self.headers).encode()
08916a49 136 if self.path.startswith('/redirect_'):
137 self._redirect()
138 elif self.path.startswith('/method'):
139 self._method('PUT', data)
140 else:
141 self._status(404)
142
83fda3c0
PH		 143 def do_GET(self):
144 if self.path == '/video.html':
08916a49 145 payload = b'<html><video src="/vid.mp4" /></html>'
83fda3c0
PH		 146 self.send_response(200)
147 self.send_header('Content-Type', 'text/html; charset=utf-8')
227bf1a3 148 self.send_header('Content-Length', str(len(payload)))
83fda3c0 149 self.end_headers()
08916a49 150 self.wfile.write(payload)
83fda3c0 151 elif self.path == '/vid.mp4':
08916a49 152 payload = b'\x00\x00\x00\x00\x20\x66\x74[video]'
83fda3c0
PH		 153 self.send_response(200)
154 self.send_header('Content-Type', 'video/mp4')
08916a49 155 self.send_header('Content-Length', str(len(payload)))
83fda3c0 156 self.end_headers()
08916a49 157 self.wfile.write(payload)
8c32e5dc 158 elif self.path == '/%E4%B8%AD%E6%96%87.html':
08916a49 159 payload = b'<html><video src="/vid.mp4" /></html>'
8c32e5dc
YCH		 160 self.send_response(200)
161 self.send_header('Content-Type', 'text/html; charset=utf-8')
08916a49 162 self.send_header('Content-Length', str(len(payload)))
163 self.end_headers()
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)))
170 self.end_headers()
171 self.wfile.write(payload)
227bf1a3 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')
176 self.end_headers()
4bf91228 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')
182 self.end_headers()
35f4f764 183 elif self.path == '/redirect_dotsegments_absolute':
184 self.send_response(301)
185 # redirect to /headers but with dot segments before - absolute url
186 self.send_header('Location', f'http://127.0.0.1:{http_server_port(self.server)}/a/b/./../../headers')
187 self.send_header('Content-Length', '0')
188 self.end_headers()
08916a49 189 elif self.path.startswith('/redirect_'):
190 self._redirect()
191 elif self.path.startswith('/method'):
227bf1a3 192 self._method('GET', str(self.headers).encode())
08916a49 193 elif self.path.startswith('/headers'):
194 self._headers()
f8b4bcc0 195 elif self.path.startswith('/308-to-headers'):
196 self.send_response(308)
197 self.send_header('Location', '/headers')
198 self.send_header('Content-Length', '0')
199 self.end_headers()
08916a49 200 elif self.path == '/trailing_garbage':
201 payload = b'<html><video src="/vid.mp4" /></html>'
202 self.send_response(200)
203 self.send_header('Content-Type', 'text/html; charset=utf-8')
204 self.send_header('Content-Encoding', 'gzip')
205 buf = io.BytesIO()
206 with gzip.GzipFile(fileobj=buf, mode='wb') as f:
207 f.write(payload)
208 compressed = buf.getvalue() + b'trailing garbage'
209 self.send_header('Content-Length', str(len(compressed)))
210 self.end_headers()
211 self.wfile.write(compressed)
212 elif self.path == '/302-non-ascii-redirect':
213 new_url = f'http://127.0.0.1:{http_server_port(self.server)}/中文.html'
214 self.send_response(301)
215 self.send_header('Location', new_url)
216 self.send_header('Content-Length', '0')
8c32e5dc 217 self.end_headers()
daafbf49 218 elif self.path == '/content-encoding':
219 encodings = self.headers.get('ytdl-encoding', '')
220 payload = b'<html><video src="/vid.mp4" /></html>'
221 for encoding in filter(None, (e.strip() for e in encodings.split(','))):
222 if encoding == 'br' and brotli:
223 payload = brotli.compress(payload)
224 elif encoding == 'gzip':
225 buf = io.BytesIO()
226 with gzip.GzipFile(fileobj=buf, mode='wb') as f:
227 f.write(payload)
228 payload = buf.getvalue()
229 elif encoding == 'deflate':
230 payload = zlib.compress(payload)
231 elif encoding == 'unsupported':
232 payload = b'raw'
233 break
234 else:
235 self._status(415)
236 return
237 self.send_response(200)
238 self.send_header('Content-Encoding', encodings)
239 self.send_header('Content-Length', str(len(payload)))
240 self.end_headers()
241 self.wfile.write(payload)
227bf1a3 242 elif self.path.startswith('/gen_'):
243 payload = b'<html></html>'
244 self.send_response(int(self.path[len('/gen_'):]))
245 self.send_header('Content-Type', 'text/html; charset=utf-8')
246 self.send_header('Content-Length', str(len(payload)))
247 self.end_headers()
248 self.wfile.write(payload)
249 elif self.path.startswith('/incompleteread'):
250 payload = b'<html></html>'
251 self.send_response(200)
252 self.send_header('Content-Type', 'text/html; charset=utf-8')
253 self.send_header('Content-Length', '234234')
254 self.end_headers()
255 self.wfile.write(payload)
256 self.finish()
257 elif self.path.startswith('/timeout_'):
258 time.sleep(int(self.path[len('/timeout_'):]))
259 self._headers()
260 elif self.path == '/source_address':
261 payload = str(self.client_address[0]).encode()
262 self.send_response(200)
263 self.send_header('Content-Type', 'text/html; charset=utf-8')
264 self.send_header('Content-Length', str(len(payload)))
265 self.end_headers()
266 self.wfile.write(payload)
267 self.finish()
83fda3c0 268 else:
08916a49 269 self._status(404)
270
271 def send_header(self, keyword, value):
272 """
273 Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
274 This is against what is defined in RFC 3986, however we need to test we support this
275 since some sites incorrectly do this.
276 """
277 if keyword.lower() == 'connection':
278 return super().send_header(keyword, value)
279
280 if not hasattr(self, '_headers_buffer'):
281 self._headers_buffer = []
282
283 self._headers_buffer.append(f'{keyword}: {value}\r\n'.encode())
83fda3c0
PH		 284
285
227bf1a3 286class TestRequestHandlerBase:
287 @classmethod
288 def setup_class(cls):
289 cls.http_httpd = http.server.ThreadingHTTPServer(
f19eae42 290 ('127.0.0.1', 0), HTTPTestRequestHandler)
227bf1a3 291 cls.http_port = http_server_port(cls.http_httpd)
292 cls.http_server_thread = threading.Thread(target=cls.http_httpd.serve_forever)
08916a49 293 # FIXME: we should probably stop the http server thread after each test
294 # See: https://github.com/yt-dlp/yt-dlp/pull/7094#discussion_r1199746041
227bf1a3 295 cls.http_server_thread.daemon = True
296 cls.http_server_thread.start()
08916a49 297
298 # HTTPS server
83fda3c0 299 certfn = os.path.join(TEST_DIR, 'testcert.pem')
227bf1a3 300 cls.https_httpd = http.server.ThreadingHTTPServer(
f19eae42 301 ('127.0.0.1', 0), HTTPTestRequestHandler)
b6dc37fe 302 sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
303 sslctx.load_cert_chain(certfn, None)
227bf1a3 304 cls.https_httpd.socket = sslctx.wrap_socket(cls.https_httpd.socket, server_side=True)
305 cls.https_port = http_server_port(cls.https_httpd)
306 cls.https_server_thread = threading.Thread(target=cls.https_httpd.serve_forever)
307 cls.https_server_thread.daemon = True
308 cls.https_server_thread.start()
309
310
227bf1a3 311class TestHTTPRequestHandler(TestRequestHandlerBase):
8a8b5452 312 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 313 def test_verify_cert(self, handler):
314 with handler() as rh:
315 with pytest.raises(CertificateVerifyError):
316 validate_and_send(rh, Request(f'https://127.0.0.1:{self.https_port}/headers'))
317
318 with handler(verify=False) as rh:
319 r = validate_and_send(rh, Request(f'https://127.0.0.1:{self.https_port}/headers'))
320 assert r.status == 200
08916a49 321 r.close()
322
8a8b5452 323 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 324 def test_ssl_error(self, handler):
325 # HTTPS server with too old TLS version
326 # XXX: is there a better way to test this than to create a new server?
327 https_httpd = http.server.ThreadingHTTPServer(
328 ('127.0.0.1', 0), HTTPTestRequestHandler)
329 sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
330 https_httpd.socket = sslctx.wrap_socket(https_httpd.socket, server_side=True)
331 https_port = http_server_port(https_httpd)
332 https_server_thread = threading.Thread(target=https_httpd.serve_forever)
333 https_server_thread.daemon = True
334 https_server_thread.start()
335
336 with handler(verify=False) as rh:
37755a03 337 with pytest.raises(SSLError, match=r'ssl(?:v3|/tls) alert handshake failure') as exc_info:
227bf1a3 338 validate_and_send(rh, Request(f'https://127.0.0.1:{https_port}/headers'))
339 assert not issubclass(exc_info.type, CertificateVerifyError)
340
8a8b5452 341 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 342 def test_percent_encode(self, handler):
343 with handler() as rh:
08916a49 344 # Unicode characters should be encoded with uppercase percent-encoding
227bf1a3 345 res = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/中文.html'))
346 assert res.status == 200
08916a49 347 res.close()
348 # don't normalize existing percent encodings
227bf1a3 349 res = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/%c7%9f'))
350 assert res.status == 200
08916a49 351 res.close()
352
8a8b5452 353 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
35f4f764 354 @pytest.mark.parametrize('path', [
355 '/a/b/./../../headers',
356 '/redirect_dotsegments',
357 # https://github.com/yt-dlp/yt-dlp/issues/9020
358 '/redirect_dotsegments_absolute',
359 ])
360 def test_remove_dot_segments(self, handler, path):
361 with handler(verbose=True) as rh:
4bf91228 362 # This isn't a comprehensive test,
35f4f764 363 # but it should be enough to check whether the handler is removing dot segments in required scenarios
364 res = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}{path}'))
4bf91228 365 assert res.status == 200
366 assert res.url == f'http://127.0.0.1:{self.http_port}/headers'
367 res.close()
368
8a8b5452 369 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 370 def test_unicode_path_redirection(self, handler):
371 with handler() as rh:
372 r = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
373 assert r.url == f'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
08916a49 374 r.close()
375
8a8b5452 376 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 377 def test_raise_http_error(self, handler):
378 with handler() as rh:
379 for bad_status in (400, 500, 599, 302):
380 with pytest.raises(HTTPError):
381 validate_and_send(rh, Request('http://127.0.0.1:%d/gen_%d' % (self.http_port, bad_status)))
382
383 # Should not raise an error
384 validate_and_send(rh, Request('http://127.0.0.1:%d/gen_200' % self.http_port)).close()
385
8a8b5452 386 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 387 def test_response_url(self, handler):
388 with handler() as rh:
389 # Response url should be that of the last url in redirect chain
390 res = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/redirect_301'))
391 assert res.url == f'http://127.0.0.1:{self.http_port}/method'
392 res.close()
393 res2 = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/gen_200'))
394 assert res2.url == f'http://127.0.0.1:{self.http_port}/gen_200'
395 res2.close()
396
8a8b5452 397 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 398 def test_redirect(self, handler):
399 with handler() as rh:
400 def do_req(redirect_status, method, assert_no_content=False):
08916a49 401 data = b'testdata' if method in ('POST', 'PUT') else None
227bf1a3 402 res = validate_and_send(
403 rh, Request(f'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method=method, data=data))
404
405 headers = b''
406 data_sent = b''
407 if data is not None:
408 data_sent += res.read(len(data))
409 if data_sent != data:
410 headers += data_sent
411 data_sent = b''
412
413 headers += res.read()
414
415 if assert_no_content or data is None:
416 assert b'Content-Type' not in headers
417 assert b'Content-Length' not in headers
418 else:
419 assert b'Content-Type' in headers
420 assert b'Content-Length' in headers
421
422 return data_sent.decode(), res.headers.get('method', '')
08916a49 423
424 # A 303 must either use GET or HEAD for subsequent request
227bf1a3 425 assert do_req(303, 'POST', True) == ('', 'GET')
426 assert do_req(303, 'HEAD') == ('', 'HEAD')
08916a49 427
227bf1a3 428 assert do_req(303, 'PUT', True) == ('', 'GET')
08916a49 429
430 # 301 and 302 turn POST only into a GET
227bf1a3 431 assert do_req(301, 'POST', True) == ('', 'GET')
432 assert do_req(301, 'HEAD') == ('', 'HEAD')
433 assert do_req(302, 'POST', True) == ('', 'GET')
434 assert do_req(302, 'HEAD') == ('', 'HEAD')
08916a49 435
227bf1a3 436 assert do_req(301, 'PUT') == ('testdata', 'PUT')
437 assert do_req(302, 'PUT') == ('testdata', 'PUT')
08916a49 438
439 # 307 and 308 should not change method
440 for m in ('POST', 'PUT'):
227bf1a3 441 assert do_req(307, m) == ('testdata', m)
442 assert do_req(308, m) == ('testdata', m)
08916a49 443
227bf1a3 444 assert do_req(307, 'HEAD') == ('', 'HEAD')
445 assert do_req(308, 'HEAD') == ('', 'HEAD')
08916a49 446
447 # These should not redirect and instead raise an HTTPError
448 for code in (300, 304, 305, 306):
227bf1a3 449 with pytest.raises(HTTPError):
08916a49 450 do_req(code, 'GET')
451
8a8b5452 452 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 453 def test_request_cookie_header(self, handler):
f8b4bcc0 454 # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
227bf1a3 455 with handler() as rh:
f8b4bcc0 456 # Specified Cookie header should be used
227bf1a3 457 res = validate_and_send(
458 rh, Request(
08916a49 459 f'http://127.0.0.1:{self.http_port}/headers',
227bf1a3 460 headers={'Cookie': 'test=test'})).read().decode()
461 assert 'Cookie: test=test' in res
08916a49 462
227bf1a3 463 # Specified Cookie header should be removed on any redirect
464 res = validate_and_send(
465 rh, Request(
466 f'http://127.0.0.1:{self.http_port}/308-to-headers',
467 headers={'Cookie': 'test=test'})).read().decode()
468 assert 'Cookie: test=test' not in res
469
470 # Specified Cookie header should override global cookiejar for that request
6148833f 471 cookiejar = YoutubeDLCookieJar()
227bf1a3 472 cookiejar.set_cookie(http.cookiejar.Cookie(
473 version=0, name='test', value='ytdlp', port=None, port_specified=False,
474 domain='127.0.0.1', domain_specified=True, domain_initial_dot=False, path='/',
475 path_specified=True, secure=False, expires=None, discard=False, comment=None,
476 comment_url=None, rest={}))
477
478 with handler(cookiejar=cookiejar) as rh:
479 data = validate_and_send(
480 rh, Request(f'http://127.0.0.1:{self.http_port}/headers', headers={'cookie': 'test=test'})).read()
481 assert b'Cookie: test=ytdlp' not in data
482 assert b'Cookie: test=test' in data
483
8a8b5452 484 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 485 def test_redirect_loop(self, handler):
486 with handler() as rh:
487 with pytest.raises(HTTPError, match='redirect loop'):
488 validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/redirect_loop'))
489
8a8b5452 490 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 491 def test_incompleteread(self, handler):
492 with handler(timeout=2) as rh:
493 with pytest.raises(IncompleteRead):
494 validate_and_send(rh, Request('http://127.0.0.1:%d/incompleteread' % self.http_port)).read()
495
8a8b5452 496 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 497 def test_cookies(self, handler):
6148833f 498 cookiejar = YoutubeDLCookieJar()
227bf1a3 499 cookiejar.set_cookie(http.cookiejar.Cookie(
500 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
501 False, '/headers', True, False, None, False, None, None, {}))
502
503 with handler(cookiejar=cookiejar) as rh:
504 data = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/headers')).read()
505 assert b'Cookie: test=ytdlp' in data
506
507 # Per request
508 with handler() as rh:
509 data = validate_and_send(
510 rh, Request(f'http://127.0.0.1:{self.http_port}/headers', extensions={'cookiejar': cookiejar})).read()
511 assert b'Cookie: test=ytdlp' in data
512
8a8b5452 513 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 514 def test_headers(self, handler):
515
516 with handler(headers=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'})) as rh:
517 # Global Headers
518 data = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/headers')).read()
519 assert b'Test1: test' in data
520
521 # Per request headers, merged with global
522 data = validate_and_send(rh, Request(
523 f'http://127.0.0.1:{self.http_port}/headers', headers={'test2': 'changed', 'test3': 'test3'})).read()
524 assert b'Test1: test' in data
525 assert b'Test2: changed' in data
526 assert b'Test2: test2' not in data
527 assert b'Test3: test3' in data
528
8a8b5452 529 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 530 def test_timeout(self, handler):
531 with handler() as rh:
532 # Default timeout is 20 seconds, so this should go through
533 validate_and_send(
534 rh, Request(f'http://127.0.0.1:{self.http_port}/timeout_3'))
535
536 with handler(timeout=0.5) as rh:
537 with pytest.raises(TransportError):
538 validate_and_send(
539 rh, Request(f'http://127.0.0.1:{self.http_port}/timeout_1'))
540
541 # Per request timeout, should override handler timeout
542 validate_and_send(
543 rh, Request(f'http://127.0.0.1:{self.http_port}/timeout_1', extensions={'timeout': 4}))
544
8a8b5452 545 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 546 def test_source_address(self, handler):
547 source_address = f'127.0.0.{random.randint(5, 255)}'
69d31914 548 # on some systems these loopback addresses we need for testing may not be available
549 # see: https://github.com/yt-dlp/yt-dlp/issues/8890
550 verify_address_availability(source_address)
227bf1a3 551 with handler(source_address=source_address) as rh:
552 data = validate_and_send(
553 rh, Request(f'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
554 assert source_address == data
555
8a8b5452 556 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 557 def test_gzip_trailing_garbage(self, handler):
558 with handler() as rh:
559 data = validate_and_send(rh, Request(f'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
560 assert data == '<html><video src="/vid.mp4" /></html>'
561
8a8b5452 562 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 563 @pytest.mark.skipif(not brotli, reason='brotli support is not installed')
564 def test_brotli(self, handler):
565 with handler() as rh:
566 res = validate_and_send(
567 rh, Request(
daafbf49 568 f'http://127.0.0.1:{self.http_port}/content-encoding',
569 headers={'ytdl-encoding': 'br'}))
227bf1a3 570 assert res.headers.get('Content-Encoding') == 'br'
571 assert res.read() == b'<html><video src="/vid.mp4" /></html>'
572
8a8b5452 573 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 574 def test_deflate(self, handler):
575 with handler() as rh:
576 res = validate_and_send(
577 rh, Request(
daafbf49 578 f'http://127.0.0.1:{self.http_port}/content-encoding',
579 headers={'ytdl-encoding': 'deflate'}))
227bf1a3 580 assert res.headers.get('Content-Encoding') == 'deflate'
581 assert res.read() == b'<html><video src="/vid.mp4" /></html>'
582
8a8b5452 583 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 584 def test_gzip(self, handler):
585 with handler() as rh:
586 res = validate_and_send(
587 rh, Request(
daafbf49 588 f'http://127.0.0.1:{self.http_port}/content-encoding',
589 headers={'ytdl-encoding': 'gzip'}))
227bf1a3 590 assert res.headers.get('Content-Encoding') == 'gzip'
591 assert res.read() == b'<html><video src="/vid.mp4" /></html>'
daafbf49 592
8a8b5452 593 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 594 def test_multiple_encodings(self, handler):
595 with handler() as rh:
daafbf49 596 for pair in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
227bf1a3 597 res = validate_and_send(
598 rh, Request(
daafbf49 599 f'http://127.0.0.1:{self.http_port}/content-encoding',
600 headers={'ytdl-encoding': pair}))
227bf1a3 601 assert res.headers.get('Content-Encoding') == pair
602 assert res.read() == b'<html><video src="/vid.mp4" /></html>'
603
8a8b5452 604 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 605 def test_unsupported_encoding(self, handler):
606 with handler() as rh:
607 res = validate_and_send(
608 rh, Request(
daafbf49 609 f'http://127.0.0.1:{self.http_port}/content-encoding',
610 headers={'ytdl-encoding': 'unsupported'}))
227bf1a3 611 assert res.headers.get('Content-Encoding') == 'unsupported'
612 assert res.read() == b'raw'
613
8a8b5452 614 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 615 def test_read(self, handler):
616 with handler() as rh:
617 res = validate_and_send(
618 rh, Request(f'http://127.0.0.1:{self.http_port}/headers'))
619 assert res.readable()
620 assert res.read(1) == b'H'
621 assert res.read(3) == b'ost'
622
623
624class TestHTTPProxy(TestRequestHandlerBase):
625 @classmethod
626 def setup_class(cls):
627 super().setup_class()
628 # HTTP Proxy server
629 cls.proxy = http.server.ThreadingHTTPServer(
630 ('127.0.0.1', 0), _build_proxy_handler('normal'))
631 cls.proxy_port = http_server_port(cls.proxy)
632 cls.proxy_thread = threading.Thread(target=cls.proxy.serve_forever)
633 cls.proxy_thread.daemon = True
634 cls.proxy_thread.start()
daafbf49 635
227bf1a3 636 # Geo proxy server
637 cls.geo_proxy = http.server.ThreadingHTTPServer(
638 ('127.0.0.1', 0), _build_proxy_handler('geo'))
639 cls.geo_port = http_server_port(cls.geo_proxy)
640 cls.geo_proxy_thread = threading.Thread(target=cls.geo_proxy.serve_forever)
641 cls.geo_proxy_thread.daemon = True
642 cls.geo_proxy_thread.start()
643
8a8b5452 644 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 645 def test_http_proxy(self, handler):
646 http_proxy = f'http://127.0.0.1:{self.proxy_port}'
647 geo_proxy = f'http://127.0.0.1:{self.geo_port}'
648
649 # Test global http proxy
650 # Test per request http proxy
651 # Test per request http proxy disables proxy
652 url = 'http://foo.com/bar'
01218f91 653
227bf1a3 654 # Global HTTP proxy
655 with handler(proxies={'http': http_proxy}) as rh:
656 res = validate_and_send(rh, Request(url)).read().decode()
657 assert res == f'normal: {url}'
658
659 # Per request proxy overrides global
660 res = validate_and_send(rh, Request(url, proxies={'http': geo_proxy})).read().decode()
661 assert res == f'geo: {url}'
662
663 # and setting to None disables all proxies for that request
664 real_url = f'http://127.0.0.1:{self.http_port}/headers'
665 res = validate_and_send(
666 rh, Request(real_url, proxies={'http': None})).read().decode()
667 assert res != f'normal: {real_url}'
668 assert 'Accept' in res
669
8a8b5452 670 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 671 def test_noproxy(self, handler):
672 with handler(proxies={'proxy': f'http://127.0.0.1:{self.proxy_port}'}) as rh:
673 # NO_PROXY
674 for no_proxy in (f'127.0.0.1:{self.http_port}', '127.0.0.1', 'localhost'):
675 nop_response = validate_and_send(
676 rh, Request(f'http://127.0.0.1:{self.http_port}/headers', proxies={'no': no_proxy})).read().decode(
677 'utf-8')
678 assert 'Accept' in nop_response
679
8a8b5452 680 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 681 def test_allproxy(self, handler):
682 url = 'http://foo.com/bar'
683 with handler() as rh:
684 response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
685 'utf-8')
686 assert response == f'normal: {url}'
687
8a8b5452 688 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 689 def test_http_proxy_with_idn(self, handler):
690 with handler(proxies={
691 'http': f'http://127.0.0.1:{self.proxy_port}',
692 }) as rh:
693 url = 'http://中文.tw/'
694 response = rh.send(Request(url)).read().decode()
695 # b'xn--fiq228c' is '中文'.encode('idna')
696 assert response == 'normal: http://xn--fiq228c.tw/'
697
698
699class TestClientCertificate:
700
701 @classmethod
702 def setup_class(cls):
bb58c9ed 703 certfn = os.path.join(TEST_DIR, 'testcert.pem')
227bf1a3 704 cls.certdir = os.path.join(TEST_DIR, 'testdata', 'certificate')
705 cacertfn = os.path.join(cls.certdir, 'ca.crt')
706 cls.httpd = http.server.ThreadingHTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler)
bb58c9ed 707 sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
708 sslctx.verify_mode = ssl.CERT_REQUIRED
709 sslctx.load_verify_locations(cafile=cacertfn)
710 sslctx.load_cert_chain(certfn, None)
227bf1a3 711 cls.httpd.socket = sslctx.wrap_socket(cls.httpd.socket, server_side=True)
712 cls.port = http_server_port(cls.httpd)
713 cls.server_thread = threading.Thread(target=cls.httpd.serve_forever)
714 cls.server_thread.daemon = True
715 cls.server_thread.start()
716
717 def _run_test(self, handler, **handler_kwargs):
718 with handler(
bb58c9ed 719 # Disable client-side validation of unacceptable self-signed testcert.pem
720 # The test is of a check on the server side, so unaffected
227bf1a3 721 verify=False,
722 **handler_kwargs,
723 ) as rh:
724 validate_and_send(rh, Request(f'https://127.0.0.1:{self.port}/video.html')).read().decode()
725
8a8b5452 726 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 727 def test_certificate_combined_nopass(self, handler):
728 self._run_test(handler, client_cert={
729 'client_certificate': os.path.join(self.certdir, 'clientwithkey.crt'),
bb58c9ed 730 })
bb58c9ed 731
8a8b5452 732 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 733 def test_certificate_nocombined_nopass(self, handler):
734 self._run_test(handler, client_cert={
735 'client_certificate': os.path.join(self.certdir, 'client.crt'),
736 'client_certificate_key': os.path.join(self.certdir, 'client.key'),
737 })
bb58c9ed 738
8a8b5452 739 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 740 def test_certificate_combined_pass(self, handler):
741 self._run_test(handler, client_cert={
742 'client_certificate': os.path.join(self.certdir, 'clientwithencryptedkey.crt'),
743 'client_certificate_password': 'foobar',
744 })
bb58c9ed 745
8a8b5452 746 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
227bf1a3 747 def test_certificate_nocombined_pass(self, handler):
748 self._run_test(handler, client_cert={
749 'client_certificate': os.path.join(self.certdir, 'client.crt'),
750 'client_certificate_key': os.path.join(self.certdir, 'clientencrypted.key'),
751 'client_certificate_password': 'foobar',
752 })
bb58c9ed 753
bb58c9ed 754
227bf1a3 755class TestUrllibRequestHandler(TestRequestHandlerBase):
756 @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
757 def test_file_urls(self, handler):
758 # See https://github.com/ytdl-org/youtube-dl/issues/8227
759 tf = tempfile.NamedTemporaryFile(delete=False)
760 tf.write(b'foobar')
761 tf.close()
762 req = Request(pathlib.Path(tf.name).as_uri())
763 with handler() as rh:
764 with pytest.raises(UnsupportedRequest):
765 rh.validate(req)
766
767 # Test that urllib never loaded FileHandler
768 with pytest.raises(TransportError):
769 rh.send(req)
770
771 with handler(enable_file_urls=True) as rh:
772 res = validate_and_send(rh, req)
773 assert res.read() == b'foobar'
774 res.close()
bb58c9ed 775
227bf1a3 776 os.unlink(tf.name)
01218f91 777
227bf1a3 778 @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
779 def test_http_error_returns_content(self, handler):
780 # urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
781 def get_response():
782 with handler() as rh:
783 # headers url
784 try:
785 validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/gen_404'))
786 except HTTPError as e:
787 return e.response
788
789 assert get_response().read() == b'<html></html>'
790
791 @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
792 def test_verify_cert_error_text(self, handler):
793 # Check the output of the error message
794 with handler() as rh:
795 with pytest.raises(
796 CertificateVerifyError,
797 match=r'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate'
798 ):
799 validate_and_send(rh, Request(f'https://127.0.0.1:{self.https_port}/headers'))
800
801 @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
95abea9a 802 @pytest.mark.parametrize('req,match,version_check', [
803 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
804 # bpo-39603: Check implemented in 3.7.9+, 3.8.5+
805 (
806 Request('http://127.0.0.1', method='GET\n'),
807 'method can\'t contain control characters',
808 lambda v: v < (3, 7, 9) or (3, 8, 0) <= v < (3, 8, 5)
809 ),
810 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
811 # bpo-38576: Check implemented in 3.7.8+, 3.8.3+
812 (
813 Request('http://127.0.0. 1', method='GET'),
814 'URL can\'t contain control characters',
815 lambda v: v < (3, 7, 8) or (3, 8, 0) <= v < (3, 8, 3)
816 ),
817 # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
818 (Request('http://127.0.0.1', headers={'foo\n': 'bar'}), 'Invalid header name', None),
819 ])
820 def test_httplib_validation_errors(self, handler, req, match, version_check):
821 if version_check and version_check(sys.version_info):
822 pytest.skip(f'Python {sys.version} version does not have the required validation for this test.')
227bf1a3 823
95abea9a 824 with handler() as rh:
825 with pytest.raises(RequestError, match=match) as exc_info:
826 validate_and_send(rh, req)
227bf1a3 827 assert not isinstance(exc_info.value, TransportError)
828
829
8a8b5452 830class TestRequestsRequestHandler(TestRequestHandlerBase):
831 @pytest.mark.parametrize('raised,expected', [
832 (lambda: requests.exceptions.ConnectTimeout(), TransportError),
833 (lambda: requests.exceptions.ReadTimeout(), TransportError),
834 (lambda: requests.exceptions.Timeout(), TransportError),
835 (lambda: requests.exceptions.ConnectionError(), TransportError),
836 (lambda: requests.exceptions.ProxyError(), ProxyError),
837 (lambda: requests.exceptions.SSLError('12[CERTIFICATE_VERIFY_FAILED]34'), CertificateVerifyError),
838 (lambda: requests.exceptions.SSLError(), SSLError),
839 (lambda: requests.exceptions.InvalidURL(), RequestError),
840 (lambda: requests.exceptions.InvalidHeader(), RequestError),
841 # catch-all: https://github.com/psf/requests/blob/main/src/requests/adapters.py#L535
842 (lambda: urllib3.exceptions.HTTPError(), TransportError),
843 (lambda: requests.exceptions.RequestException(), RequestError)
844 # (lambda: requests.exceptions.TooManyRedirects(), HTTPError) - Needs a response object
845 ])
846 @pytest.mark.parametrize('handler', ['Requests'], indirect=True)
847 def test_request_error_mapping(self, handler, monkeypatch, raised, expected):
848 with handler() as rh:
849 def mock_get_instance(*args, **kwargs):
850 class MockSession:
851 def request(self, *args, **kwargs):
852 raise raised()
853 return MockSession()
854
855 monkeypatch.setattr(rh, '_get_instance', mock_get_instance)
856
857 with pytest.raises(expected) as exc_info:
858 rh.send(Request('http://fake'))
859
860 assert exc_info.type is expected
861
862 @pytest.mark.parametrize('raised,expected,match', [
863 (lambda: urllib3.exceptions.SSLError(), SSLError, None),
864 (lambda: urllib3.exceptions.TimeoutError(), TransportError, None),
865 (lambda: urllib3.exceptions.ReadTimeoutError(None, None, None), TransportError, None),
866 (lambda: urllib3.exceptions.ProtocolError(), TransportError, None),
867 (lambda: urllib3.exceptions.DecodeError(), TransportError, None),
868 (lambda: urllib3.exceptions.HTTPError(), TransportError, None), # catch-all
869 (
870 lambda: urllib3.exceptions.ProtocolError('error', http.client.IncompleteRead(partial=b'abc', expected=4)),
871 IncompleteRead,
872 '3 bytes read, 4 more expected'
873 ),
874 (
4e38e2ae 875 lambda: urllib3.exceptions.ProtocolError('error', urllib3.exceptions.IncompleteRead(partial=3, expected=5)),
8a8b5452 876 IncompleteRead,
877 '3 bytes read, 5 more expected'
878 ),
879 ])
880 @pytest.mark.parametrize('handler', ['Requests'], indirect=True)
881 def test_response_error_mapping(self, handler, monkeypatch, raised, expected, match):
8a8b5452 882 from requests.models import Response as RequestsResponse
ccfd70f4 883 from urllib3.response import HTTPResponse as Urllib3Response
884
8a8b5452 885 from yt_dlp.networking._requests import RequestsResponseAdapter
886 requests_res = RequestsResponse()
887 requests_res.raw = Urllib3Response(body=b'', status=200)
888 res = RequestsResponseAdapter(requests_res)
889
890 def mock_read(*args, **kwargs):
891 raise raised()
892 monkeypatch.setattr(res.fp, 'read', mock_read)
893
894 with pytest.raises(expected, match=match) as exc_info:
895 res.read()
896
897 assert exc_info.type is expected
898
899
86aea0d3 900def run_validation(handler, error, req, **handler_kwargs):
227bf1a3 901 with handler(**handler_kwargs) as rh:
86aea0d3 902 if error:
903 with pytest.raises(error):
227bf1a3 904 rh.validate(req)
905 else:
906 rh.validate(req)
907
908
909class TestRequestHandlerValidation:
910
911 class ValidationRH(RequestHandler):
912 def _send(self, request):
913 raise RequestError('test')
914
915 class NoCheckRH(ValidationRH):
916 _SUPPORTED_FEATURES = None
917 _SUPPORTED_PROXY_SCHEMES = None
918 _SUPPORTED_URL_SCHEMES = None
919
86aea0d3 920 def _check_extensions(self, extensions):
921 extensions.clear()
922
227bf1a3 923 class HTTPSupportedRH(ValidationRH):
924 _SUPPORTED_URL_SCHEMES = ('http',)
925
926 URL_SCHEME_TESTS = [
927 # scheme, expected to fail, handler kwargs
928 ('Urllib', [
929 ('http', False, {}),
930 ('https', False, {}),
931 ('data', False, {}),
932 ('ftp', False, {}),
86aea0d3 933 ('file', UnsupportedRequest, {}),
227bf1a3 934 ('file', False, {'enable_file_urls': True}),
935 ]),
8a8b5452 936 ('Requests', [
937 ('http', False, {}),
938 ('https', False, {}),
939 ]),
ccfd70f4 940 ('Websockets', [
941 ('ws', False, {}),
942 ('wss', False, {}),
943 ]),
227bf1a3 944 (NoCheckRH, [('http', False, {})]),
86aea0d3 945 (ValidationRH, [('http', UnsupportedRequest, {})])
227bf1a3 946 ]
947
948 PROXY_SCHEME_TESTS = [
949 # scheme, expected to fail
ccfd70f4 950 ('Urllib', 'http', [
227bf1a3 951 ('http', False),
86aea0d3 952 ('https', UnsupportedRequest),
227bf1a3 953 ('socks4', False),
954 ('socks4a', False),
955 ('socks5', False),
956 ('socks5h', False),
86aea0d3 957 ('socks', UnsupportedRequest),
227bf1a3 958 ]),
ccfd70f4 959 ('Requests', 'http', [
8a8b5452 960 ('http', False),
961 ('https', False),
962 ('socks4', False),
963 ('socks4a', False),
964 ('socks5', False),
965 ('socks5h', False),
966 ]),
ccfd70f4 967 (NoCheckRH, 'http', [('http', False)]),
968 (HTTPSupportedRH, 'http', [('http', UnsupportedRequest)]),
969 ('Websockets', 'ws', [('http', UnsupportedRequest)]),
970 (NoCheckRH, 'http', [('http', False)]),
971 (HTTPSupportedRH, 'http', [('http', UnsupportedRequest)]),
227bf1a3 972 ]
973
974 PROXY_KEY_TESTS = [
975 # key, expected to fail
976 ('Urllib', [
977 ('all', False),
978 ('unrelated', False),
979 ]),
8a8b5452 980 ('Requests', [
981 ('all', False),
982 ('unrelated', False),
983 ]),
227bf1a3 984 (NoCheckRH, [('all', False)]),
86aea0d3 985 (HTTPSupportedRH, [('all', UnsupportedRequest)]),
986 (HTTPSupportedRH, [('no', UnsupportedRequest)]),
987 ]
988
989 EXTENSION_TESTS = [
ccfd70f4 990 ('Urllib', 'http', [
86aea0d3 991 ({'cookiejar': 'notacookiejar'}, AssertionError),
6148833f 992 ({'cookiejar': YoutubeDLCookieJar()}, False),
993 ({'cookiejar': CookieJar()}, AssertionError),
86aea0d3 994 ({'timeout': 1}, False),
995 ({'timeout': 'notatimeout'}, AssertionError),
996 ({'unsupported': 'value'}, UnsupportedRequest),
997 ]),
ccfd70f4 998 ('Requests', 'http', [
8a8b5452 999 ({'cookiejar': 'notacookiejar'}, AssertionError),
1000 ({'cookiejar': YoutubeDLCookieJar()}, False),
1001 ({'timeout': 1}, False),
1002 ({'timeout': 'notatimeout'}, AssertionError),
1003 ({'unsupported': 'value'}, UnsupportedRequest),
1004 ]),
ccfd70f4 1005 (NoCheckRH, 'http', [
86aea0d3 1006 ({'cookiejar': 'notacookiejar'}, False),
1007 ({'somerandom': 'test'}, False), # but any extension is allowed through
1008 ]),
ccfd70f4 1009 ('Websockets', 'ws', [
1010 ({'cookiejar': YoutubeDLCookieJar()}, False),
1011 ({'timeout': 2}, False),
1012 ]),
227bf1a3 1013 ]
1014
1015 @pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
1016 (handler_tests[0], scheme, fail, handler_kwargs)
1017 for handler_tests in URL_SCHEME_TESTS
1018 for scheme, fail, handler_kwargs in handler_tests[1]
1019
1020 ], indirect=['handler'])
1021 def test_url_scheme(self, handler, scheme, fail, handler_kwargs):
1022 run_validation(handler, fail, Request(f'{scheme}://'), **(handler_kwargs or {}))
1023
8a8b5452 1024 @pytest.mark.parametrize('handler,fail', [('Urllib', False), ('Requests', False)], indirect=['handler'])
227bf1a3 1025 def test_no_proxy(self, handler, fail):
1026 run_validation(handler, fail, Request('http://', proxies={'no': '127.0.0.1,github.com'}))
1027 run_validation(handler, fail, Request('http://'), proxies={'no': '127.0.0.1,github.com'})
1028
1029 @pytest.mark.parametrize('handler,proxy_key,fail', [
1030 (handler_tests[0], proxy_key, fail)
1031 for handler_tests in PROXY_KEY_TESTS
1032 for proxy_key, fail in handler_tests[1]
1033 ], indirect=['handler'])
1034 def test_proxy_key(self, handler, proxy_key, fail):
1035 run_validation(handler, fail, Request('http://', proxies={proxy_key: 'http://example.com'}))
1036 run_validation(handler, fail, Request('http://'), proxies={proxy_key: 'http://example.com'})
1037
ccfd70f4 1038 @pytest.mark.parametrize('handler,req_scheme,scheme,fail', [
1039 (handler_tests[0], handler_tests[1], scheme, fail)
227bf1a3 1040 for handler_tests in PROXY_SCHEME_TESTS
ccfd70f4 1041 for scheme, fail in handler_tests[2]
227bf1a3 1042 ], indirect=['handler'])
ccfd70f4 1043 def test_proxy_scheme(self, handler, req_scheme, scheme, fail):
1044 run_validation(handler, fail, Request(f'{req_scheme}://', proxies={req_scheme: f'{scheme}://example.com'}))
1045 run_validation(handler, fail, Request(f'{req_scheme}://'), proxies={req_scheme: f'{scheme}://example.com'})
227bf1a3 1046
8a8b5452 1047 @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH, 'Requests'], indirect=True)
227bf1a3 1048 def test_empty_proxy(self, handler):
1049 run_validation(handler, False, Request('http://', proxies={'http': None}))
1050 run_validation(handler, False, Request('http://'), proxies={'http': None})
1051
bbeacff7 1052 @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
8a8b5452 1053 @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
bbeacff7 1054 def test_invalid_proxy_url(self, handler, proxy_url):
1055 run_validation(handler, UnsupportedRequest, Request('http://', proxies={'http': proxy_url}))
227bf1a3 1056
ccfd70f4 1057 @pytest.mark.parametrize('handler,scheme,extensions,fail', [
1058 (handler_tests[0], handler_tests[1], extensions, fail)
86aea0d3 1059 for handler_tests in EXTENSION_TESTS
ccfd70f4 1060 for extensions, fail in handler_tests[2]
86aea0d3 1061 ], indirect=['handler'])
ccfd70f4 1062 def test_extension(self, handler, scheme, extensions, fail):
86aea0d3 1063 run_validation(
ccfd70f4 1064 handler, fail, Request(f'{scheme}://', extensions=extensions))
227bf1a3 1065
1066 def test_invalid_request_type(self):
1067 rh = self.ValidationRH(logger=FakeLogger())
1068 for method in (rh.validate, rh.send):
1069 with pytest.raises(TypeError, match='Expected an instance of Request'):
1070 method('not a request')
1071
1072
1073class FakeResponse(Response):
1074 def __init__(self, request):
1075 # XXX: we could make request part of standard response interface
1076 self.request = request
1077 super().__init__(fp=io.BytesIO(b''), headers={}, url=request.url)
1078
1079
1080class FakeRH(RequestHandler):
1081
1082 def _validate(self, request):
1083 return
1084
1085 def _send(self, request: Request):
1086 if request.url.startswith('ssl://'):
1087 raise SSLError(request.url[len('ssl://'):])
1088 return FakeResponse(request)
1089
1090
1091class FakeRHYDL(FakeYDL):
1092 def __init__(self, *args, **kwargs):
1093 super().__init__(*args, **kwargs)
1094 self._request_director = self.build_request_director([FakeRH])
1095
1096
ccfd70f4 1097class AllUnsupportedRHYDL(FakeYDL):
1098
1099 def __init__(self, *args, **kwargs):
1100
1101 class UnsupportedRH(RequestHandler):
1102 def _send(self, request: Request):
1103 pass
1104
1105 _SUPPORTED_FEATURES = ()
1106 _SUPPORTED_PROXY_SCHEMES = ()
1107 _SUPPORTED_URL_SCHEMES = ()
1108
1109 super().__init__(*args, **kwargs)
1110 self._request_director = self.build_request_director([UnsupportedRH])
1111
1112
227bf1a3 1113class TestRequestDirector:
1114
1115 def test_handler_operations(self):
1116 director = RequestDirector(logger=FakeLogger())
1117 handler = FakeRH(logger=FakeLogger())
1118 director.add_handler(handler)
1119 assert director.handlers.get(FakeRH.RH_KEY) is handler
1120
1121 # Handler should overwrite
1122 handler2 = FakeRH(logger=FakeLogger())
1123 director.add_handler(handler2)
1124 assert director.handlers.get(FakeRH.RH_KEY) is not handler
1125 assert director.handlers.get(FakeRH.RH_KEY) is handler2
1126 assert len(director.handlers) == 1
1127
1128 class AnotherFakeRH(FakeRH):
01218f91 1129 pass
227bf1a3 1130 director.add_handler(AnotherFakeRH(logger=FakeLogger()))
1131 assert len(director.handlers) == 2
1132 assert director.handlers.get(AnotherFakeRH.RH_KEY).RH_KEY == AnotherFakeRH.RH_KEY
01218f91 1133
227bf1a3 1134 director.handlers.pop(FakeRH.RH_KEY, None)
1135 assert director.handlers.get(FakeRH.RH_KEY) is None
1136 assert len(director.handlers) == 1
01218f91 1137
227bf1a3 1138 # RequestErrors should passthrough
1139 with pytest.raises(SSLError):
1140 director.send(Request('ssl://something'))
01218f91 1141
227bf1a3 1142 def test_send(self):
1143 director = RequestDirector(logger=FakeLogger())
1144 with pytest.raises(RequestError):
1145 director.send(Request('any://'))
1146 director.add_handler(FakeRH(logger=FakeLogger()))
1147 assert isinstance(director.send(Request('http://')), FakeResponse)
01218f91 1148
227bf1a3 1149 def test_unsupported_handlers(self):
227bf1a3 1150 class SupportedRH(RequestHandler):
1151 _SUPPORTED_URL_SCHEMES = ['http']
01218f91 1152
227bf1a3 1153 def _send(self, request: Request):
1154 return Response(fp=io.BytesIO(b'supported'), headers={}, url=request.url)
efbed08d 1155
db7b054a 1156 director = RequestDirector(logger=FakeLogger())
227bf1a3 1157 director.add_handler(SupportedRH(logger=FakeLogger()))
db7b054a 1158 director.add_handler(FakeRH(logger=FakeLogger()))
1159
1160 # First should take preference
227bf1a3 1161 assert director.send(Request('http://')).read() == b'supported'
1162 assert director.send(Request('any://')).read() == b''
582be358 1163
227bf1a3 1164 director.handlers.pop(FakeRH.RH_KEY)
1165 with pytest.raises(NoSupportingHandlers):
1166 director.send(Request('any://'))
1167
1168 def test_unexpected_error(self):
1169 director = RequestDirector(logger=FakeLogger())
1170
1171 class UnexpectedRH(FakeRH):
1172 def _send(self, request: Request):
1173 raise TypeError('something')
1174
1175 director.add_handler(UnexpectedRH(logger=FakeLogger))
1176 with pytest.raises(NoSupportingHandlers, match=r'1 unexpected error'):
1177 director.send(Request('any://'))
1178
1179 director.handlers.clear()
1180 assert len(director.handlers) == 0
1181
1182 # Should not be fatal
1183 director.add_handler(FakeRH(logger=FakeLogger()))
1184 director.add_handler(UnexpectedRH(logger=FakeLogger))
1185 assert director.send(Request('any://'))
1186
db7b054a 1187 def test_preference(self):
1188 director = RequestDirector(logger=FakeLogger())
1189 director.add_handler(FakeRH(logger=FakeLogger()))
1190
1191 class SomeRH(RequestHandler):
1192 _SUPPORTED_URL_SCHEMES = ['http']
1193
1194 def _send(self, request: Request):
1195 return Response(fp=io.BytesIO(b'supported'), headers={}, url=request.url)
1196
1197 def some_preference(rh, request):
1198 return (0 if not isinstance(rh, SomeRH)
1199 else 100 if 'prefer' in request.headers
1200 else -1)
1201
1202 director.add_handler(SomeRH(logger=FakeLogger()))
1203 director.preferences.add(some_preference)
1204
1205 assert director.send(Request('http://')).read() == b''
1206 assert director.send(Request('http://', headers={'prefer': '1'})).read() == b'supported'
1207
227bf1a3 1208
1209# XXX: do we want to move this to test_YoutubeDL.py?
1210class TestYoutubeDLNetworking:
1211
1212 @staticmethod
1213 def build_handler(ydl, handler: RequestHandler = FakeRH):
1214 return ydl.build_request_director([handler]).handlers.get(handler.RH_KEY)
1215
1216 def test_compat_opener(self):
08916a49 1217 with FakeYDL() as ydl:
227bf1a3 1218 with warnings.catch_warnings():
1219 warnings.simplefilter('ignore', category=DeprecationWarning)
1220 assert isinstance(ydl._opener, urllib.request.OpenerDirector)
1221
1222 @pytest.mark.parametrize('proxy,expected', [
1223 ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}),
1224 ('', {'all': '__noproxy__'}),
1225 (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}) # env, set https
1226 ])
1227 def test_proxy(self, proxy, expected):
1228 old_http_proxy = os.environ.get('HTTP_PROXY')
1229 try:
1230 os.environ['HTTP_PROXY'] = 'http://127.0.0.1:8081' # ensure that provided proxies override env
1231 with FakeYDL({'proxy': proxy}) as ydl:
1232 assert ydl.proxies == expected
1233 finally:
1234 if old_http_proxy:
1235 os.environ['HTTP_PROXY'] = old_http_proxy
1236
1237 def test_compat_request(self):
1238 with FakeRHYDL() as ydl:
1239 assert ydl.urlopen('test://')
1240 urllib_req = urllib.request.Request('http://foo.bar', data=b'test', method='PUT', headers={'X-Test': '1'})
1241 urllib_req.add_unredirected_header('Cookie', 'bob=bob')
1242 urllib_req.timeout = 2
3d2623a8 1243 with warnings.catch_warnings():
1244 warnings.simplefilter('ignore', category=DeprecationWarning)
1245 req = ydl.urlopen(urllib_req).request
1246 assert req.url == urllib_req.get_full_url()
1247 assert req.data == urllib_req.data
1248 assert req.method == urllib_req.get_method()
1249 assert 'X-Test' in req.headers
1250 assert 'Cookie' in req.headers
1251 assert req.extensions.get('timeout') == 2
227bf1a3 1252
1253 with pytest.raises(AssertionError):
1254 ydl.urlopen(None)
1255
1256 def test_extract_basic_auth(self):
1257 with FakeRHYDL() as ydl:
1258 res = ydl.urlopen(Request('http://user:pass@foo.bar'))
1259 assert res.request.headers['Authorization'] == 'Basic dXNlcjpwYXNz'
1260
1261 def test_sanitize_url(self):
1262 with FakeRHYDL() as ydl:
1263 res = ydl.urlopen(Request('httpss://foo.bar'))
1264 assert res.request.url == 'https://foo.bar'
1265
1266 def test_file_urls_error(self):
1267 # use urllib handler
1268 with FakeYDL() as ydl:
1269 with pytest.raises(RequestError, match=r'file:// URLs are disabled by default'):
1270 ydl.urlopen('file://')
1271
ccfd70f4 1272 @pytest.mark.parametrize('scheme', (['ws', 'wss']))
1273 def test_websocket_unavailable_error(self, scheme):
1274 with AllUnsupportedRHYDL() as ydl:
1275 with pytest.raises(RequestError, match=r'This request requires WebSocket support'):
1276 ydl.urlopen(f'{scheme}://')
1277
227bf1a3 1278 def test_legacy_server_connect_error(self):
1279 with FakeRHYDL() as ydl:
1280 for error in ('UNSAFE_LEGACY_RENEGOTIATION_DISABLED', 'SSLV3_ALERT_HANDSHAKE_FAILURE'):
1281 with pytest.raises(RequestError, match=r'Try using --legacy-server-connect'):
1282 ydl.urlopen(f'ssl://{error}')
1283
1284 with pytest.raises(SSLError, match='testerror'):
1285 ydl.urlopen('ssl://testerror')
1286
1287 @pytest.mark.parametrize('proxy_key,proxy_url,expected', [
1288 ('http', '__noproxy__', None),
1289 ('no', '127.0.0.1,foo.bar', '127.0.0.1,foo.bar'),
1290 ('https', 'example.com', 'http://example.com'),
bbeacff7 1291 ('https', '//example.com', 'http://example.com'),
227bf1a3 1292 ('https', 'socks5://example.com', 'socks5h://example.com'),
1293 ('http', 'socks://example.com', 'socks4://example.com'),
1294 ('http', 'socks4://example.com', 'socks4://example.com'),
bbeacff7 1295 ('unrelated', '/bad/proxy', '/bad/proxy'), # clean_proxies should ignore bad proxies
227bf1a3 1296 ])
1297 def test_clean_proxy(self, proxy_key, proxy_url, expected):
1298 # proxies should be cleaned in urlopen()
1299 with FakeRHYDL() as ydl:
1300 req = ydl.urlopen(Request('test://', proxies={proxy_key: proxy_url})).request
1301 assert req.proxies[proxy_key] == expected
1302
1303 # and should also be cleaned when building the handler
1304 env_key = f'{proxy_key.upper()}_PROXY'
1305 old_env_proxy = os.environ.get(env_key)
1306 try:
1307 os.environ[env_key] = proxy_url # ensure that provided proxies override env
1308 with FakeYDL() as ydl:
1309 rh = self.build_handler(ydl)
1310 assert rh.proxies[proxy_key] == expected
1311 finally:
1312 if old_env_proxy:
1313 os.environ[env_key] = old_env_proxy
1314
1315 def test_clean_proxy_header(self):
1316 with FakeRHYDL() as ydl:
1317 req = ydl.urlopen(Request('test://', headers={'ytdl-request-proxy': '//foo.bar'})).request
1318 assert 'ytdl-request-proxy' not in req.headers
1319 assert req.proxies == {'all': 'http://foo.bar'}
1320
1321 with FakeYDL({'http_headers': {'ytdl-request-proxy': '//foo.bar'}}) as ydl:
1322 rh = self.build_handler(ydl)
1323 assert 'ytdl-request-proxy' not in rh.headers
1324 assert rh.proxies == {'all': 'http://foo.bar'}
1325
1326 def test_clean_header(self):
1327 with FakeRHYDL() as ydl:
1328 res = ydl.urlopen(Request('test://', headers={'Youtubedl-no-compression': True}))
1329 assert 'Youtubedl-no-compression' not in res.request.headers
1330 assert res.request.headers.get('Accept-Encoding') == 'identity'
1331
1332 with FakeYDL({'http_headers': {'Youtubedl-no-compression': True}}) as ydl:
1333 rh = self.build_handler(ydl)
1334 assert 'Youtubedl-no-compression' not in rh.headers
1335 assert rh.headers.get('Accept-Encoding') == 'identity'
1336
f04b5bed 1337 with FakeYDL({'http_headers': {'Ytdl-socks-proxy': 'socks://localhost:1080'}}) as ydl:
1338 rh = self.build_handler(ydl)
1339 assert 'Ytdl-socks-proxy' not in rh.headers
1340
227bf1a3 1341 def test_build_handler_params(self):
1342 with FakeYDL({
1343 'http_headers': {'test': 'testtest'},
1344 'socket_timeout': 2,
1345 'proxy': 'http://127.0.0.1:8080',
1346 'source_address': '127.0.0.45',
1347 'debug_printtraffic': True,
1348 'compat_opts': ['no-certifi'],
1349 'nocheckcertificate': True,
75dc8e67 1350 'legacyserverconnect': True,
227bf1a3 1351 }) as ydl:
1352 rh = self.build_handler(ydl)
1353 assert rh.headers.get('test') == 'testtest'
1354 assert 'Accept' in rh.headers # ensure std_headers are still there
1355 assert rh.timeout == 2
1356 assert rh.proxies.get('all') == 'http://127.0.0.1:8080'
1357 assert rh.source_address == '127.0.0.45'
1358 assert rh.verbose is True
1359 assert rh.prefer_system_certs is True
1360 assert rh.verify is False
1361 assert rh.legacy_ssl_support is True
1362
1363 @pytest.mark.parametrize('ydl_params', [
1364 {'client_certificate': 'fakecert.crt'},
1365 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key'},
1366 {'client_certificate': 'fakecert.crt', 'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'},
1367 {'client_certificate_key': 'fakekey.key', 'client_certificate_password': 'foobar'},
1368 ])
1369 def test_client_certificate(self, ydl_params):
1370 with FakeYDL(ydl_params) as ydl:
1371 rh = self.build_handler(ydl)
1372 assert rh._client_cert == ydl_params # XXX: Too bound to implementation
1373
1374 def test_urllib_file_urls(self):
1375 with FakeYDL({'enable_file_urls': False}) as ydl:
1376 rh = self.build_handler(ydl, UrllibRH)
1377 assert rh.enable_file_urls is False
08916a49 1378
227bf1a3 1379 with FakeYDL({'enable_file_urls': True}) as ydl:
1380 rh = self.build_handler(ydl, UrllibRH)
1381 assert rh.enable_file_urls is True
1382
8a8b5452 1383 def test_compat_opt_prefer_urllib(self):
1384 # This assumes urllib only has a preference when this compat opt is given
1385 with FakeYDL({'compat_opts': ['prefer-legacy-http-handler']}) as ydl:
1386 director = ydl.build_request_director([UrllibRH])
1387 assert len(director.preferences) == 1
1388 assert director.preferences.pop()(UrllibRH, None)
1389
227bf1a3 1390
1391class TestRequest:
1392
1393 def test_query(self):
1394 req = Request('http://example.com?q=something', query={'v': 'xyz'})
1395 assert req.url == 'http://example.com?q=something&v=xyz'
1396
1397 req.update(query={'v': '123'})
1398 assert req.url == 'http://example.com?q=something&v=123'
1399 req.update(url='http://example.com', query={'v': 'xyz'})
1400 assert req.url == 'http://example.com?v=xyz'
1401
1402 def test_method(self):
1403 req = Request('http://example.com')
1404 assert req.method == 'GET'
1405 req.data = b'test'
1406 assert req.method == 'POST'
1407 req.data = None
1408 assert req.method == 'GET'
1409 req.data = b'test2'
1410 req.method = 'PUT'
1411 assert req.method == 'PUT'
1412 req.data = None
1413 assert req.method == 'PUT'
1414 with pytest.raises(TypeError):
1415 req.method = 1
1416
1417 def test_request_helpers(self):
1418 assert HEADRequest('http://example.com').method == 'HEAD'
1419 assert PUTRequest('http://example.com').method == 'PUT'
1420
1421 def test_headers(self):
1422 req = Request('http://example.com', headers={'tesT': 'test'})
1423 assert req.headers == HTTPHeaderDict({'test': 'test'})
1424 req.update(headers={'teSt2': 'test2'})
1425 assert req.headers == HTTPHeaderDict({'test': 'test', 'test2': 'test2'})
1426
1427 req.headers = new_headers = HTTPHeaderDict({'test': 'test'})
1428 assert req.headers == HTTPHeaderDict({'test': 'test'})
1429 assert req.headers is new_headers
1430
1431 # test converts dict to case insensitive dict
1432 req.headers = new_headers = {'test2': 'test2'}
1433 assert isinstance(req.headers, HTTPHeaderDict)
1434 assert req.headers is not new_headers
1435
1436 with pytest.raises(TypeError):
1437 req.headers = None
1438
1439 def test_data_type(self):
1440 req = Request('http://example.com')
1441 assert req.data is None
1442 # test bytes is allowed
1443 req.data = b'test'
1444 assert req.data == b'test'
1445 # test iterable of bytes is allowed
1446 i = [b'test', b'test2']
1447 req.data = i
1448 assert req.data == i
1449
1450 # test file-like object is allowed
1451 f = io.BytesIO(b'test')
1452 req.data = f
1453 assert req.data == f
1454
1455 # common mistake: test str not allowed
1456 with pytest.raises(TypeError):
1457 req.data = 'test'
1458 assert req.data != 'test'
1459
1460 # common mistake: test dict is not allowed
1461 with pytest.raises(TypeError):
1462 req.data = {'test': 'test'}
1463 assert req.data != {'test': 'test'}
1464
1465 def test_content_length_header(self):
1466 req = Request('http://example.com', headers={'Content-Length': '0'}, data=b'')
1467 assert req.headers.get('Content-Length') == '0'
1468
1469 req.data = b'test'
1470 assert 'Content-Length' not in req.headers
1471
1472 req = Request('http://example.com', headers={'Content-Length': '10'})
1473 assert 'Content-Length' not in req.headers
1474
1475 def test_content_type_header(self):
1476 req = Request('http://example.com', headers={'Content-Type': 'test'}, data=b'test')
1477 assert req.headers.get('Content-Type') == 'test'
1478 req.data = b'test2'
1479 assert req.headers.get('Content-Type') == 'test'
1480 req.data = None
1481 assert 'Content-Type' not in req.headers
1482 req.data = b'test3'
1483 assert req.headers.get('Content-Type') == 'application/x-www-form-urlencoded'
1484
71baa490 1485 def test_update_req(self):
1486 req = Request('http://example.com')
1487 assert req.data is None
1488 assert req.method == 'GET'
1489 assert 'Content-Type' not in req.headers
1490 # Test that zero-byte payloads will be sent
1491 req.update(data=b'')
1492 assert req.data == b''
1493 assert req.method == 'POST'
1494 assert req.headers.get('Content-Type') == 'application/x-www-form-urlencoded'
1495
227bf1a3 1496 def test_proxies(self):
1497 req = Request(url='http://example.com', proxies={'http': 'http://127.0.0.1:8080'})
1498 assert req.proxies == {'http': 'http://127.0.0.1:8080'}
1499
1500 def test_extensions(self):
1501 req = Request(url='http://example.com', extensions={'timeout': 2})
1502 assert req.extensions == {'timeout': 2}
1503
1504 def test_copy(self):
1505 req = Request(
1506 url='http://example.com',
1507 extensions={'cookiejar': CookieJar()},
1508 headers={'Accept-Encoding': 'br'},
1509 proxies={'http': 'http://127.0.0.1'},
1510 data=[b'123']
1511 )
1512 req_copy = req.copy()
1513 assert req_copy is not req
1514 assert req_copy.url == req.url
1515 assert req_copy.headers == req.headers
1516 assert req_copy.headers is not req.headers
1517 assert req_copy.proxies == req.proxies
1518 assert req_copy.proxies is not req.proxies
1519
1520 # Data is not able to be copied
1521 assert req_copy.data == req.data
1522 assert req_copy.data is req.data
1523
1524 # Shallow copy extensions
1525 assert req_copy.extensions is not req.extensions
1526 assert req_copy.extensions['cookiejar'] == req.extensions['cookiejar']
1527
1528 # Subclasses are copied by default
1529 class AnotherRequest(Request):
1530 pass
08916a49 1531
227bf1a3 1532 req = AnotherRequest(url='http://127.0.0.1')
1533 assert isinstance(req.copy(), AnotherRequest)
1534
1535 def test_url(self):
1536 req = Request(url='https://фtest.example.com/ some spaceв?ä=c',)
1537 assert req.url == 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c'
1538
1539 assert Request(url='//example.com').url == 'http://example.com'
1540
1541 with pytest.raises(TypeError):
1542 Request(url='https://').url = None
1543
1544
1545class TestResponse:
1546
1547 @pytest.mark.parametrize('reason,status,expected', [
1548 ('custom', 200, 'custom'),
1549 (None, 404, 'Not Found'), # fallback status
1550 ('', 403, 'Forbidden'),
1551 (None, 999, None)
1552 ])
1553 def test_reason(self, reason, status, expected):
1554 res = Response(io.BytesIO(b''), url='test://', headers={}, status=status, reason=reason)
1555 assert res.reason == expected
1556
1557 def test_headers(self):
1558 headers = Message()
1559 headers.add_header('Test', 'test')
1560 headers.add_header('Test', 'test2')
1561 headers.add_header('content-encoding', 'br')
1562 res = Response(io.BytesIO(b''), headers=headers, url='test://')
1563 assert res.headers.get_all('test') == ['test', 'test2']
1564 assert 'Content-Encoding' in res.headers
1565
1566 def test_get_header(self):
1567 headers = Message()
1568 headers.add_header('Set-Cookie', 'cookie1')
1569 headers.add_header('Set-cookie', 'cookie2')
1570 headers.add_header('Test', 'test')
1571 headers.add_header('Test', 'test2')
1572 res = Response(io.BytesIO(b''), headers=headers, url='test://')
1573 assert res.get_header('test') == 'test, test2'
1574 assert res.get_header('set-Cookie') == 'cookie1'
1575 assert res.get_header('notexist', 'default') == 'default'
1576
1577 def test_compat(self):
1578 res = Response(io.BytesIO(b''), url='test://', status=404, headers={'test': 'test'})
3d2623a8 1579 with warnings.catch_warnings():
1580 warnings.simplefilter('ignore', category=DeprecationWarning)
1581 assert res.code == res.getcode() == res.status
1582 assert res.geturl() == res.url
1583 assert res.info() is res.headers
1584 assert res.getheader('test') == res.get_header('test')
fork of https://github.com/yt-dlp/yt-dlp/