test/test_networking.py

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