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
from yt_dlp.networking.exceptions import (
HTTPError,
IncompleteRead,
- _CompatHTTPError,
)
from yt_dlp.socks import ProxyType
from yt_dlp.utils.networking import HTTPHeaderDict
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']
'port': 1080,
'rdns': True,
'username': None,
- 'password': None
+ 'password': None,
}),
('socks5://user:@example.com:5555', {
'proxytype': ProxyType.SOCKS5,
'port': 5555,
'rdns': False,
'username': 'user',
- 'password': ''
+ 'password': '',
}),
('socks4://u%40ser:pa%20ss@127.0.0.1:1080', {
'proxytype': ProxyType.SOCKS4,
'port': 1080,
'rdns': False,
'username': 'u@ser',
- 'password': 'pa ss'
+ 'password': 'pa ss',
}),
('socks4a://:pa%20ss@127.0.0.1', {
'proxytype': ProxyType.SOCKS4A,
'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
@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'),
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'
assert data == b'test'
assert repr(error) == '<HTTPError 403: Forbidden>'
- @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) == '<IncompleteRead: 4 bytes read, 3 more expected>'
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) == '<IncompleteRead: 3 bytes read>'
assert str(error) == '3 bytes read'
projects / yt-dlp.git / blobdiff