X-Git-Url: https://jfr.im/git/yt-dlp.git/blobdiff_plain/3d2623a898196640f7cc0fc8b70118ff19e6925d..54a63e80af82791d2f0985bd0176bb182963fd5f:/test/test_networking_utils.py diff --git a/test/test_networking_utils.py b/test/test_networking_utils.py index ef46f79ed..204fe87bd 100644 --- a/test/test_networking_utils.py +++ b/test/test_networking_utils.py @@ -8,13 +8,9 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -import contextlib import io -import platform import random import ssl -import urllib.error -import warnings from yt_dlp.cookies import YoutubeDLCookieJar from yt_dlp.dependencies import certifi @@ -30,7 +26,6 @@ from yt_dlp.networking.exceptions import ( HTTPError, IncompleteRead, - _CompatHTTPError, ) from yt_dlp.socks import ProxyType from yt_dlp.utils.networking import HTTPHeaderDict @@ -44,7 +39,7 @@ def test_select_proxy(self): proxies = { 'all': 'socks5://example.com', 'http': 'http://example.com:1080', - 'no': 'bypass.example.com,yt-dl.org' + 'no': 'bypass.example.com,yt-dl.org', } assert select_proxy('https://example.com', proxies) == proxies['all'] @@ -59,7 +54,7 @@ def test_select_proxy(self): 'port': 1080, 'rdns': True, 'username': None, - 'password': None + 'password': None, }), ('socks5://user:@example.com:5555', { 'proxytype': ProxyType.SOCKS5, @@ -67,7 +62,7 @@ def test_select_proxy(self): 'port': 5555, 'rdns': False, 'username': 'user', - 'password': '' + 'password': '', }), ('socks4://u%40ser:pa%20ss@127.0.0.1:1080', { 'proxytype': ProxyType.SOCKS4, @@ -75,7 +70,7 @@ def test_select_proxy(self): 'port': 1080, 'rdns': False, 'username': 'u@ser', - 'password': 'pa ss' + 'password': 'pa ss', }), ('socks4a://:pa%20ss@127.0.0.1', { 'proxytype': ProxyType.SOCKS4A, @@ -83,8 +78,8 @@ def test_select_proxy(self): 'port': 1080, 'rdns': True, 'username': '', - 'password': 'pa ss' - }) + 'password': 'pa ss', + }), ]) def test_make_socks_proxy_opts(self, socks_proxy, expected): assert make_socks_proxy_opts(socks_proxy) == expected @@ -95,17 +90,20 @@ def test_make_socks_proxy_unknown(self): @pytest.mark.skipif(not certifi, reason='certifi is not installed') def test_load_certifi(self): + context_certifi = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context_certifi.load_verify_locations(cafile=certifi.where()) context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - context2 = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ssl_load_certs(context, use_certifi=True) - context2.load_verify_locations(cafile=certifi.where()) - assert context.get_ca_certs() == context2.get_ca_certs() + assert context.get_ca_certs() == context_certifi.get_ca_certs() + + context_default = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context_default.load_default_certs() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ssl_load_certs(context, use_certifi=False) + assert context.get_ca_certs() == context_default.get_ca_certs() - # Test load normal certs - # XXX: could there be a case where system certs are the same as certifi? - context3 = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ssl_load_certs(context3, use_certifi=False) - assert context3.get_ca_certs() != context.get_ca_certs() + if context_default.get_ca_certs() == context_certifi.get_ca_certs(): + pytest.skip('System uses certifi as default. The test is not valid') @pytest.mark.parametrize('method,status,expected', [ ('GET', 303, 'GET'), @@ -176,11 +174,10 @@ class TestNetworkingExceptions: def create_response(status): return Response(fp=io.BytesIO(b'test'), url='http://example.com', headers={'tesT': 'test'}, status=status) - @pytest.mark.parametrize('http_error_class', [HTTPError, lambda r: _CompatHTTPError(HTTPError(r))]) - def test_http_error(self, http_error_class): + def test_http_error(self): response = self.create_response(403) - error = http_error_class(response) + error = HTTPError(response) assert error.status == 403 assert str(error) == error.msg == 'HTTP Error 403: Forbidden' @@ -191,89 +188,21 @@ def test_http_error(self, http_error_class): assert data == b'test' assert repr(error) == '' - @pytest.mark.parametrize('http_error_class', [HTTPError, lambda *args, **kwargs: _CompatHTTPError(HTTPError(*args, **kwargs))]) - def test_redirect_http_error(self, http_error_class): + def test_redirect_http_error(self): response = self.create_response(301) - error = http_error_class(response, redirect_loop=True) + error = HTTPError(response, redirect_loop=True) assert str(error) == error.msg == 'HTTP Error 301: Moved Permanently (redirect loop detected)' assert error.reason == 'Moved Permanently' - def test_compat_http_error(self): - response = self.create_response(403) - error = _CompatHTTPError(HTTPError(response)) - assert isinstance(error, HTTPError) - assert isinstance(error, urllib.error.HTTPError) - - @contextlib.contextmanager - def raises_deprecation_warning(): - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter('always') - yield - - if len(w) == 0: - pytest.fail('Did not raise DeprecationWarning') - if len(w) > 1: - pytest.fail(f'Raised multiple warnings: {w}') - - if not issubclass(w[-1].category, DeprecationWarning): - pytest.fail(f'Expected DeprecationWarning, got {w[-1].category}') - w.clear() - - with raises_deprecation_warning(): - assert error.code == 403 - - with raises_deprecation_warning(): - assert error.getcode() == 403 - - with raises_deprecation_warning(): - assert error.hdrs is error.response.headers - - with raises_deprecation_warning(): - assert error.info() is error.response.headers - - with raises_deprecation_warning(): - assert error.headers is error.response.headers - - with raises_deprecation_warning(): - assert error.filename == error.response.url - - with raises_deprecation_warning(): - assert error.url == error.response.url - - with raises_deprecation_warning(): - assert error.geturl() == error.response.url - - # Passthrough file operations - with raises_deprecation_warning(): - assert error.read() == b'test' - - with raises_deprecation_warning(): - assert not error.closed - - with raises_deprecation_warning(): - # Technically Response operations are also passed through, which should not be used. - assert error.get_header('test') == 'test' - - # Should not raise a warning - error.close() - - @pytest.mark.skipif( - platform.python_implementation() == 'PyPy', reason='garbage collector works differently in pypy') - def test_compat_http_error_autoclose(self): - # Compat HTTPError should not autoclose response - response = self.create_response(403) - _CompatHTTPError(HTTPError(response)) - assert not response.closed - def test_incomplete_read_error(self): - error = IncompleteRead(b'test', 3, cause='test') + error = IncompleteRead(4, 3, cause='test') assert isinstance(error, IncompleteRead) assert repr(error) == '' assert str(error) == error.msg == '4 bytes read, 3 more expected' - assert error.partial == b'test' + assert error.partial == 4 assert error.expected == 3 assert error.cause == 'test' - error = IncompleteRead(b'aaa') + error = IncompleteRead(3) assert repr(error) == '' assert str(error) == '3 bytes read'