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