Add coverage for all raise branches
MODIFY
.coveragerc
.coveragerc
@@ -21,7 +21,7 @@ exclude_lines =
except ImportError:
pass
import
- raise
+ raise NotImplementedError
.* # Platform-specific.*
.*:.* # Python \d.*
.* # Abstract

MODIFY
CONTRIBUTORS.txt
CONTRIBUTORS.txt
@@ -318,5 +318,8 @@ In chronological order:
* Bastian Venthur <https://venthur.de>
* Added body-parameter to top-level request() and RequestMethods.request
+* Bastian Venthur <https://venthur.de>
+ * fixed all remaining coverage checks for raise statements
+
* [Your name or handle] <[email or website]>
* [Brief summary of your changes]

MODIFY
test/contrib/test_socks.py
test/contrib/test_socks.py
import socket
import threading
from socket import getaddrinfo as real_getaddrinfo
+from socket import timeout as SocketTimeout
from test import SHORT_TIMEOUT
+from unittest.mock import patch
import pytest
import socks as py_socks
@@ -341,6 +343,14 @@ class TestSocks5Proxy(IPV4SocketDummyServerTestCase):
)
event.set()
+ @patch("socks.create_connection")
+ def test_socket_timeout(self, create_connection):
+ create_connection.side_effect = SocketTimeout()
+ proxy_url = f"socks5h://{self.host}:{self.port}"
+ with socks.SOCKSProxyManager(proxy_url) as pm:
+ with pytest.raises(ConnectTimeoutError, match="timed out"):
+ pm.request("GET", "http://example.com", retries=False)
+
def test_connection_failure(self):
event = threading.Event()

MODIFY
test/test_connection.py
test/test_connection.py
@@ -13,7 +13,7 @@ from urllib3.connection import (
from urllib3.util.ssl_match_hostname import (
CertificateError as ImplementationCertificateError,
)
-from urllib3.util.ssl_match_hostname import match_hostname
+from urllib3.util.ssl_match_hostname import _dnsname_match, match_hostname
class TestConnection:
@@ -83,6 +83,16 @@ class TestConnection:
asserted_hostname = "foobar"
_match_hostname(cert, asserted_hostname)
+ def test_match_hostname_more_than_one_dnsname_error(self):
+ cert = {"subjectAltName": [("DNS", "foo*"), ("DNS", "fo*")]}
+ asserted_hostname = "bar"
+ with pytest.raises(CertificateError, match="doesn't match either of"):
+ _match_hostname(cert, asserted_hostname)
+
+ def test_dnsname_match_include_more_than_one_wildcard_error(self):
+ with pytest.raises(CertificateError, match="too many wildcards in certificate"):
+ _dnsname_match("foo**", "foobar")
+
def test_match_hostname_ignore_common_name(self):
cert = {"subject": [("commonName", "foo")]}
asserted_hostname = "foo"

MODIFY
test/test_connectionpool.py
test/test_connectionpool.py
@@ -5,7 +5,7 @@ from queue import Empty
from socket import error as SocketError
from ssl import SSLError as BaseSSLError
from test import SHORT_TIMEOUT
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
import pytest
@@ -25,6 +25,7 @@ from urllib3.exceptions import (
LocationValueError,
MaxRetryError,
ProtocolError,
+ ReadTimeoutError,
SSLError,
TimeoutError,
)
@@ -533,6 +534,14 @@ class TestConnectionPool:
_test(SocketError)
_test(ProtocolError)
+ def test_read_timeout_0_does_not_raise_bad_status_line_error(self):
+ with HTTPConnectionPool(host="localhost", maxsize=1) as pool:
+ conn = Mock()
+ with patch.object(Timeout, "read_timeout", 0):
+ timeout = Timeout(1, 1, 1)
+ with pytest.raises(ReadTimeoutError):
+ pool._make_request(conn, "", "", timeout)
+
def test_custom_http_response_class(self):
class CustomHTTPResponse(HTTPResponse):
pass

MODIFY
test/test_response.py
test/test_response.py
@@ -11,6 +11,7 @@ from unittest import mock
import pytest
from urllib3.exceptions import (
+ BodyNotHttplibCompatible,
DecodeError,
IncompleteRead,
InvalidChunkLength,
@@ -745,6 +746,15 @@ class TestResponse:
with pytest.raises(ResponseNotChunked):
next(r)
+ def test_read_chunked_not_supported(self):
+ fp = BytesIO(b"foo")
+ resp = HTTPResponse(
+ fp, preload_content=False, headers={"transfer-encoding": "chunked"}
+ )
+ r = resp.read_chunked()
+ with pytest.raises(BodyNotHttplibCompatible):
+ next(r)
+
def test_buggy_incomplete_read(self):
# Simulate buggy versions of Python (<2.7.4)
# See http://bugs.python.org/issue16298

MODIFY
test/test_ssl.py
test/test_ssl.py
@@ -2,7 +2,7 @@ from unittest import mock
import pytest
-from urllib3.exceptions import SNIMissingWarning, SSLError
+from urllib3.exceptions import ProxySchemeUnsupported, SNIMissingWarning, SSLError
from urllib3.util import ssl_
@@ -95,6 +95,11 @@ class TestSSL:
assert context.set_ciphers.call_count == 1
assert context.set_ciphers.call_args == mock.call(expected_ciphers)
+ def test_create_urllib3_no_context(self):
+ with mock.patch("urllib3.util.ssl_.SSLContext", None):
+ with pytest.raises(TypeError):
+ ssl_.create_urllib3_context()
+
def test_wrap_socket_given_context_no_load_default_certs(self):
context = mock.create_autospec(ssl_.SSLContext)
context.load_default_certs = mock.Mock()
@@ -129,6 +134,12 @@ class TestSSL:
context.load_default_certs.assert_called_with()
+ def test_wrap_socket_no_ssltransport(self):
+ with mock.patch("urllib3.util.ssl_.SSLTransport", None):
+ with pytest.raises(ProxySchemeUnsupported):
+ sock = mock.Mock()
+ ssl_.ssl_wrap_socket(sock, tls_in_tls=True)
+
@pytest.mark.parametrize(
["pha", "expected_pha"], [(None, None), (False, True), (True, True)]
)

MODIFY
test/test_ssltransport.py
test/test_ssltransport.py
@@ -503,3 +503,44 @@ class TestSSLTransportWithMock:
mock.ANY, mock.ANY, server_hostname=server_hostname
)
assert not ssl_transport.suppress_ragged_eofs
+
+ def test_various_flags_errors(self):
+ server_hostname = "example-domain.com"
+ sock = mock.Mock()
+ context = mock.create_autospec(ssl_.SSLContext)
+ ssl_transport = SSLTransport(
+ sock, context, server_hostname=server_hostname, suppress_ragged_eofs=False
+ )
+ with pytest.raises(ValueError):
+ ssl_transport.recv(flags=1)
+
+ with pytest.raises(ValueError):
+ ssl_transport.recv_into(None, flags=1)
+
+ with pytest.raises(ValueError):
+ ssl_transport.sendall(None, flags=1)
+
+ with pytest.raises(ValueError):
+ ssl_transport.send(None, flags=1)
+
+ def test_makefile_wrong_mode_error(self):
+ server_hostname = "example-domain.com"
+ sock = mock.Mock()
+ context = mock.create_autospec(ssl_.SSLContext)
+ ssl_transport = SSLTransport(
+ sock, context, server_hostname=server_hostname, suppress_ragged_eofs=False
+ )
+ with pytest.raises(ValueError):
+ ssl_transport.makefile(mode="x")
+
+ def test_wrap_ssl_read_error(self):
+ server_hostname = "example-domain.com"
+ sock = mock.Mock()
+ context = mock.create_autospec(ssl_.SSLContext)
+ ssl_transport = SSLTransport(
+ sock, context, server_hostname=server_hostname, suppress_ragged_eofs=False
+ )
+ with mock.patch.object(ssl_transport, "_ssl_io_loop") as _ssl_io_loop:
+ _ssl_io_loop.side_effect = ssl.SSLError()
+ with pytest.raises(ssl.SSLError):
+ ssl_transport._wrap_ssl_read(1)

MODIFY
test/test_util.py
test/test_util.py
@@ -145,6 +145,7 @@ class TestUtil:
"http://::1/",
"http://::1:80/",
"http://google.com:-80",
+ "http://google.com:65536",
"http://google.com:\xb2\xb2", # \xb2 = ^2
# Invalid IDNA labels
"http://\uD7FF.com",
@@ -546,6 +547,7 @@ class TestUtil:
({"read": True}, "cannot be a boolean"),
({"connect": 0}, "less than or equal"),
({"read": "foo"}, "int, float or None"),
+ ({"read": "1.0"}, "int, float or None"),
],
)
def test_invalid_timeouts(self, kwargs, message):
@@ -744,6 +746,12 @@ class TestUtil:
socket.return_value = Mock()
create_connection((host, 80))
+ @patch("socket.getaddrinfo")
+ def test_create_connection_error(self, getaddrinfo):
+ getaddrinfo.return_value = []
+ with pytest.raises(OSError, match="getaddrinfo returns an empty list"):
+ create_connection(("example.com", 80))
+
@pytest.mark.parametrize(
"input,params,expected",
(

MODIFY
test/with_dummyserver/test_socketlevel.py
test/with_dummyserver/test_socketlevel.py
@@ -34,6 +34,7 @@ except ImportError:
pass
+import errno
import os
import os.path
import select
@@ -478,6 +479,7 @@ class TestSocketClosing(SocketDummyServerTestCase):
timed_out.wait()
sock.close()
+ # first ReadTimeoutError due to SocketTimeout
self._start_server(socket_handler)
with HTTPSConnectionPool(
self.host, self.port, timeout=LONG_TIMEOUT, retries=False
@@ -488,6 +490,13 @@ class TestSocketClosing(SocketDummyServerTestCase):
finally:
timed_out.set()
+ # second ReadTimeoutError due to errno
+ with HTTPSConnectionPool(host=self.host):
+ err = mock.Mock()
+ err.errno = errno.EAGAIN
+ with pytest.raises(ReadTimeoutError):
+ pool._raise_timeout(err, "", 0)
+
def test_timeout_errors_cause_retries(self):
def socket_handler(listener):
sock_timeout = listener.accept()[0]

Add coverage for all raise branches