Use `yt_dlp.networking.exceptions.HTTPError`.
`_CompatHTTPError` was to help with transition to the networking framework.
Authored by: coletdjnz
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
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(4, 3, cause='test')
assert isinstance(error, IncompleteRead)
NoSupportingHandlers,
RequestError,
SSLError,
- _CompatHTTPError,
network_exceptions,
)
from .plugins import directories as plugin_directories
'SSLV3_ALERT_HANDSHAKE_FAILURE: The server may not support the current cipher list. '
'Try using --legacy-server-connect', cause=e) from e
raise
- except HTTPError as e: # TODO: Remove in a future release
- raise _CompatHTTPError(e) from e
def build_request_director(self, handlers, preferences=None):
logger = _YDLLogger(self)
from ..dependencies import brotli as compat_brotli # noqa: F401
from ..dependencies import websockets as compat_websockets # noqa: F401
from ..dependencies.Cryptodome import AES as compat_pycrypto_AES # noqa: F401
+from ..networking.exceptions import HTTPError as compat_HTTPError # noqa: F401
passthrough_module(__name__, '...utils', ('WINDOWS_VT_MODE', 'windows_enable_vt_mode'))
compat_HTMLParser = compat_html_parser_HTMLParser = html.parser.HTMLParser
compat_http_client = http.client
compat_http_server = http.server
-compat_HTTPError = urllib.error.HTTPError
compat_input = input
compat_integer_types = (int, )
compat_itertools_count = itertools.count
compat_subprocess_get_DEVNULL = lambda: subprocess.DEVNULL
compat_tokenize_tokenize = tokenize.tokenize
compat_urllib_error = urllib.error
-compat_urllib_HTTPError = urllib.error.HTTPError
+compat_urllib_HTTPError = compat_HTTPError
compat_urllib_parse = urllib.parse
compat_urllib_parse_parse_qs = urllib.parse.parse_qs
compat_urllib_parse_quote = urllib.parse.quote
from __future__ import annotations
import typing
-import urllib.error
-from ..utils import YoutubeDLError, deprecation_warning
+from ..utils import YoutubeDLError
if typing.TYPE_CHECKING:
from .common import RequestHandler, Response
pass
-class _CompatHTTPError(urllib.error.HTTPError, HTTPError):
- """
- Provides backwards compatibility with urllib.error.HTTPError.
- Do not use this class directly, use HTTPError instead.
- """
-
- def __init__(self, http_error: HTTPError):
- super().__init__(
- url=http_error.response.url,
- code=http_error.status,
- msg=http_error.msg,
- hdrs=http_error.response.headers,
- fp=http_error.response
- )
- self._closer.close_called = True # Disable auto close
- self._http_error = http_error
- HTTPError.__init__(self, http_error.response, redirect_loop=http_error.redirect_loop)
-
- @property
- def status(self):
- return self._http_error.status
-
- @status.setter
- def status(self, value):
- return
-
- @property
- def reason(self):
- return self._http_error.reason
-
- @reason.setter
- def reason(self, value):
- return
-
- @property
- def headers(self):
- deprecation_warning('HTTPError.headers is deprecated, use HTTPError.response.headers instead')
- return self._http_error.response.headers
-
- @headers.setter
- def headers(self, value):
- return
-
- def info(self):
- deprecation_warning('HTTPError.info() is deprecated, use HTTPError.response.headers instead')
- return self.response.headers
-
- def getcode(self):
- deprecation_warning('HTTPError.getcode is deprecated, use HTTPError.status instead')
- return self.status
-
- def geturl(self):
- deprecation_warning('HTTPError.geturl is deprecated, use HTTPError.response.url instead')
- return self.response.url
-
- @property
- def code(self):
- deprecation_warning('HTTPError.code is deprecated, use HTTPError.status instead')
- return self.status
-
- @code.setter
- def code(self, value):
- return
-
- @property
- def url(self):
- deprecation_warning('HTTPError.url is deprecated, use HTTPError.response.url instead')
- return self.response.url
-
- @url.setter
- def url(self, value):
- return
-
- @property
- def hdrs(self):
- deprecation_warning('HTTPError.hdrs is deprecated, use HTTPError.response.headers instead')
- return self.response.headers
-
- @hdrs.setter
- def hdrs(self, value):
- return
-
- @property
- def filename(self):
- deprecation_warning('HTTPError.filename is deprecated, use HTTPError.response.url instead')
- return self.response.url
-
- @filename.setter
- def filename(self, value):
- return
-
- def __getattr__(self, name):
- # File operations are passed through the response.
- # Warn for some commonly used ones
- passthrough_warnings = {
- 'read': 'response.read()',
- # technically possibly due to passthrough, but we should discourage this
- 'get_header': 'response.get_header()',
- 'readable': 'response.readable()',
- 'closed': 'response.closed',
- 'tell': 'response.tell()',
- }
- if name in passthrough_warnings:
- deprecation_warning(f'HTTPError.{name} is deprecated, use HTTPError.{passthrough_warnings[name]} instead')
- return super().__getattr__(name)
-
- def __str__(self):
- return str(self._http_error)
-
- def __repr__(self):
- return repr(self._http_error)
-
-
network_exceptions = (HTTPError, TransportError)