# TODO: Break this module up into pieces. Maybe group by functionality tested # rather than the socket level-ness of it. import errno import os import os.path import select import shutil import socket import ssl import tempfile from collections import OrderedDict from test import ( LONG_TIMEOUT, SHORT_TIMEOUT, notSecureTransport, notWindows, requires_ssl_context_keyfile_password, resolvesLocalhostFQDN, ) from threading import Event from unittest import mock import pytest import trustme from dummyserver.server import ( DEFAULT_CA, DEFAULT_CERTS, encrypt_key_pem, get_unreachable_address, ) from dummyserver.testcase import SocketDummyServerTestCase, consume_socket from urllib3 import HTTPConnectionPool, HTTPSConnectionPool, util from urllib3._collections import HTTPHeaderDict from urllib3.connection import HTTPConnection, _get_default_user_agent from urllib3.exceptions import ( MaxRetryError, ProtocolError, ProxyError, ReadTimeoutError, SSLError, ) from urllib3.poolmanager import proxy_from_url from urllib3.util import ssl_, ssl_wrap_socket from urllib3.util.retry import Retry from urllib3.util.timeout import Timeout from .. import LogRecorder, has_alpn # Retry failed tests pytestmark = pytest.mark.flaky class TestCookies(SocketDummyServerTestCase): def test_multi_setcookie(self): def multicookie_response_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( b"HTTP/1.1 200 OK\r\n" b"Set-Cookie: foo=1\r\n" b"Set-Cookie: bar=1\r\n" b"\r\n" ) sock.close() self._start_server(multicookie_response_handler) with HTTPConnectionPool(self.host, self.port) as pool: r = pool.request("GET", "/", retries=0) assert r.headers == {"set-cookie": "foo=1, bar=1"} assert r.headers.getlist("set-cookie") == ["foo=1", "bar=1"] class TestSNI(SocketDummyServerTestCase): def test_hostname_in_first_request_packet(self): if not util.HAS_SNI: pytest.skip("SNI-support not available") done_receiving = Event() self.buf = b"" def socket_handler(listener): sock = listener.accept()[0] self.buf = sock.recv(65536) # We only accept one packet done_receiving.set() # let the test know it can proceed sock.close() self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port) as pool: try: pool.request("GET", "/", retries=0) except MaxRetryError: # We are violating the protocol pass successful = done_receiving.wait(LONG_TIMEOUT) assert successful, "Timed out waiting for connection accept" assert ( self.host.encode("ascii") in self.buf ), "missing hostname in SSL handshake" class TestALPN(SocketDummyServerTestCase): def test_alpn_protocol_in_first_request_packet(self): if not has_alpn(): pytest.skip("ALPN-support not available") done_receiving = Event() self.buf = b"" def socket_handler(listener): sock = listener.accept()[0] self.buf = sock.recv(65536) # We only accept one packet done_receiving.set() # let the test know it can proceed sock.close() self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port) as pool: try: pool.request("GET", "/", retries=0) except MaxRetryError: # We are violating the protocol pass successful = done_receiving.wait(LONG_TIMEOUT) assert successful, "Timed out waiting for connection accept" for protocol in util.ALPN_PROTOCOLS: assert ( protocol.encode("ascii") in self.buf ), "missing ALPN protocol in SSL handshake" class TestClientCerts(SocketDummyServerTestCase): """ Tests for client certificate support. """ @classmethod def setup_class(cls): cls.tmpdir = tempfile.mkdtemp() ca = trustme.CA() cert = ca.issue_cert("localhost") encrypted_key = encrypt_key_pem(cert.private_key_pem, b"letmein") cls.ca_path = os.path.join(cls.tmpdir, "ca.pem") cls.cert_combined_path = os.path.join(cls.tmpdir, "server.combined.pem") cls.cert_path = os.path.join(cls.tmpdir, "server.pem") cls.key_path = os.path.join(cls.tmpdir, "key.pem") cls.password_key_path = os.path.join(cls.tmpdir, "password_key.pem") ca.cert_pem.write_to_path(cls.ca_path) cert.private_key_and_cert_chain_pem.write_to_path(cls.cert_combined_path) cert.cert_chain_pems[0].write_to_path(cls.cert_path) cert.private_key_pem.write_to_path(cls.key_path) encrypted_key.write_to_path(cls.password_key_path) def teardown_class(cls): shutil.rmtree(cls.tmpdir) def _wrap_in_ssl(self, sock): """ Given a single socket, wraps it in TLS. """ return ssl.wrap_socket( sock, ssl_version=ssl.PROTOCOL_SSLv23, cert_reqs=ssl.CERT_REQUIRED, ca_certs=self.ca_path, certfile=self.cert_path, keyfile=self.key_path, server_side=True, ) def test_client_certs_two_files(self): """ Having a client cert in a separate file to its associated key works properly. """ done_receiving = Event() client_certs = [] def socket_handler(listener): sock = listener.accept()[0] sock = self._wrap_in_ssl(sock) client_certs.append(sock.getpeercert()) data = b"" while not data.endswith(b"\r\n\r\n"): data += sock.recv(8192) sock.sendall( b"HTTP/1.1 200 OK\r\n" b"Server: testsocket\r\n" b"Connection: close\r\n" b"Content-Length: 6\r\n" b"\r\n" b"Valid!" ) done_receiving.wait(5) sock.close() self._start_server(socket_handler) with HTTPSConnectionPool( self.host, self.port, cert_file=self.cert_path, key_file=self.key_path, cert_reqs="REQUIRED", ca_certs=self.ca_path, ) as pool: pool.request("GET", "/", retries=0) done_receiving.set() assert len(client_certs) == 1 def test_client_certs_one_file(self): """ Having a client cert and its associated private key in just one file works properly. """ done_receiving = Event() client_certs = [] def socket_handler(listener): sock = listener.accept()[0] sock = self._wrap_in_ssl(sock) client_certs.append(sock.getpeercert()) data = b"" while not data.endswith(b"\r\n\r\n"): data += sock.recv(8192) sock.sendall( b"HTTP/1.1 200 OK\r\n" b"Server: testsocket\r\n" b"Connection: close\r\n" b"Content-Length: 6\r\n" b"\r\n" b"Valid!" ) done_receiving.wait(5) sock.close() self._start_server(socket_handler) with HTTPSConnectionPool( self.host, self.port, cert_file=self.cert_combined_path, cert_reqs="REQUIRED", ca_certs=self.ca_path, ) as pool: pool.request("GET", "/", retries=0) done_receiving.set() assert len(client_certs) == 1 def test_missing_client_certs_raises_error(self): """ Having client certs not be present causes an error. """ done_receiving = Event() def socket_handler(listener): sock = listener.accept()[0] try: self._wrap_in_ssl(sock) except ssl.SSLError: pass done_receiving.wait(5) sock.close() self._start_server(socket_handler) with HTTPSConnectionPool( self.host, self.port, cert_reqs="REQUIRED", ca_certs=self.ca_path ) as pool: with pytest.raises(MaxRetryError): pool.request("GET", "/", retries=0) done_receiving.set() done_receiving.set() @requires_ssl_context_keyfile_password def test_client_cert_with_string_password(self): self.run_client_cert_with_password_test("letmein") @requires_ssl_context_keyfile_password def test_client_cert_with_bytes_password(self): self.run_client_cert_with_password_test(b"letmein") def run_client_cert_with_password_test(self, password): """ Tests client certificate password functionality """ done_receiving = Event() client_certs = [] def socket_handler(listener): sock = listener.accept()[0] sock = self._wrap_in_ssl(sock) client_certs.append(sock.getpeercert()) data = b"" while not data.endswith(b"\r\n\r\n"): data += sock.recv(8192) sock.sendall( b"HTTP/1.1 200 OK\r\n" b"Server: testsocket\r\n" b"Connection: close\r\n" b"Content-Length: 6\r\n" b"\r\n" b"Valid!" ) done_receiving.wait(5) sock.close() self._start_server(socket_handler) ssl_context = ssl_.SSLContext(ssl_.PROTOCOL_SSLv23) ssl_context.load_cert_chain( certfile=self.cert_path, keyfile=self.password_key_path, password=password ) with HTTPSConnectionPool( self.host, self.port, ssl_context=ssl_context, cert_reqs="REQUIRED", ca_certs=self.ca_path, ) as pool: pool.request("GET", "/", retries=0) done_receiving.set() assert len(client_certs) == 1 @requires_ssl_context_keyfile_password def test_load_keyfile_with_invalid_password(self): context = ssl_.SSLContext(ssl_.PROTOCOL_SSLv23) # Different error is raised depending on context. if ssl_.IS_PYOPENSSL: from OpenSSL.SSL import Error expected_error = Error else: expected_error = ssl.SSLError with pytest.raises(expected_error): context.load_cert_chain( certfile=self.cert_path, keyfile=self.password_key_path, password=b"letmei", ) class TestSocketClosing(SocketDummyServerTestCase): def test_recovery_when_server_closes_connection(self): # Does the pool work seamlessly if an open connection in the # connection pool gets hung up on by the server, then reaches # the front of the queue again? done_closing = Event() def socket_handler(listener): for i in 0, 1: sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65536) body = f"Response {int(i)}" sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(body), body) ).encode("utf-8") ) sock.close() # simulate a server timing out, closing socket done_closing.set() # let the test know it can proceed self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: response = pool.request("GET", "/", retries=0) assert response.status == 200 assert response.data == b"Response 0" done_closing.wait() # wait until the socket in our pool gets closed response = pool.request("GET", "/", retries=0) assert response.status == 200 assert response.data == b"Response 1" def test_connection_refused(self): # Does the pool retry if there is no listener on the port? host, port = get_unreachable_address() with HTTPConnectionPool(host, port, maxsize=3, block=True) as http: with pytest.raises(MaxRetryError): http.request("GET", "/", retries=0, release_conn=False) assert http.pool.qsize() == http.pool.maxsize def test_connection_read_timeout(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] while not sock.recv(65536).endswith(b"\r\n\r\n"): pass timed_out.wait() sock.close() self._start_server(socket_handler) with HTTPConnectionPool( self.host, self.port, timeout=SHORT_TIMEOUT, retries=False, maxsize=3, block=True, ) as http: try: with pytest.raises(ReadTimeoutError): http.request("GET", "/", release_conn=False) finally: timed_out.set() assert http.pool.qsize() == http.pool.maxsize def test_read_timeout_dont_retry_method_not_in_allowlist(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] sock.recv(65536) timed_out.wait() sock.close() self._start_server(socket_handler) with HTTPConnectionPool( self.host, self.port, timeout=LONG_TIMEOUT, retries=True ) as pool: try: with pytest.raises(ReadTimeoutError): pool.request("POST", "/") finally: timed_out.set() def test_https_connection_read_timeout(self): """ Handshake timeouts should fail with a Timeout""" timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] while not sock.recv(65536): pass 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 ) as pool: try: with pytest.raises(ReadTimeoutError): pool.request("GET", "/") 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] # Wait for a second request before closing the first socket. sock = listener.accept()[0] sock_timeout.close() # Second request. buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) # Now respond immediately. body = "Response 2" sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(body), body) ).encode("utf-8") ) sock.close() # In situations where the main thread throws an exception, the server # thread can hang on an accept() call. This ensures everything times # out within 1 second. This should be long enough for any socket # operations in the test suite to complete default_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(1) try: self._start_server(socket_handler) t = Timeout(connect=LONG_TIMEOUT, read=LONG_TIMEOUT) with HTTPConnectionPool(self.host, self.port, timeout=t) as pool: response = pool.request("GET", "/", retries=1) assert response.status == 200 assert response.data == b"Response 2" finally: socket.setdefaulttimeout(default_timeout) def test_delayed_body_read_timeout(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" body = "Hi" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65536) sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" % len(body) ).encode("utf-8") ) timed_out.wait() sock.send(body.encode("utf-8")) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: response = pool.urlopen( "GET", "/", retries=0, preload_content=False, timeout=Timeout(connect=1, read=LONG_TIMEOUT), ) try: with pytest.raises(ReadTimeoutError): response.read() finally: timed_out.set() def test_delayed_body_read_timeout_with_preload(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" body = "Hi" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" % len(body) ).encode("utf-8") ) timed_out.wait(5) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: try: with pytest.raises(ReadTimeoutError): timeout = Timeout(connect=LONG_TIMEOUT, read=SHORT_TIMEOUT) pool.urlopen("GET", "/", retries=False, timeout=timeout) finally: timed_out.set() def test_incomplete_response(self): body = "Response" partial_body = body[:2] def socket_handler(listener): sock = listener.accept()[0] # Consume request buf = b"" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65536) # Send partial response and close socket. sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(body), partial_body) ).encode("utf-8") ) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: response = pool.request("GET", "/", retries=0, preload_content=False) with pytest.raises(ProtocolError): response.read() def test_retry_weird_http_version(self): """ Retry class should handle httplib.BadStatusLine errors properly """ def socket_handler(listener): sock = listener.accept()[0] # First request. # Pause before responding so the first request times out. buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) # send unknown http protocol body = "bad http 0.5 response" sock.send( ( "HTTP/0.5 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(body), body) ).encode("utf-8") ) sock.close() # Second request. sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) # Now respond immediately. sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "foo" % (len("foo")) ).encode("utf-8") ) sock.close() # Close the socket. self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: retry = Retry(read=1) response = pool.request("GET", "/", retries=retry) assert response.status == 200 assert response.data == b"foo" def test_connection_cleanup_on_read_timeout(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" body = "Hi" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65536) sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" % len(body) ).encode("utf-8") ) timed_out.wait() sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: poolsize = pool.pool.qsize() response = pool.urlopen( "GET", "/", retries=0, preload_content=False, timeout=LONG_TIMEOUT ) try: with pytest.raises(ReadTimeoutError): response.read() assert poolsize == pool.pool.qsize() finally: timed_out.set() def test_connection_cleanup_on_protocol_error_during_read(self): body = "Response" partial_body = body[:2] def socket_handler(listener): sock = listener.accept()[0] # Consume request buf = b"" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65536) # Send partial response and close socket. sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(body), partial_body) ).encode("utf-8") ) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: poolsize = pool.pool.qsize() response = pool.request("GET", "/", retries=0, preload_content=False) with pytest.raises(ProtocolError): response.read() assert poolsize == pool.pool.qsize() def test_connection_closed_on_read_timeout_preload_false(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] # Consume request buf = b"" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65535) # Send partial chunked response and then hang. sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Transfer-Encoding: chunked\r\n" b"\r\n" b"8\r\n" b"12345678\r\n" ) timed_out.wait(5) # Expect a new request, but keep hold of the old socket to avoid # leaking it. Because we don't want to hang this thread, we # actually use select.select to confirm that a new request is # coming in: this lets us time the thread out. rlist, _, _ = select.select([listener], [], [], 1) assert rlist new_sock = listener.accept()[0] # Consume request buf = b"" while not buf.endswith(b"\r\n\r\n"): buf = new_sock.recv(65535) # Send complete chunked response. new_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Transfer-Encoding: chunked\r\n" b"\r\n" b"8\r\n" b"12345678\r\n" b"0\r\n\r\n" ) new_sock.close() sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: # First request should fail. response = pool.urlopen( "GET", "/", retries=0, preload_content=False, timeout=LONG_TIMEOUT ) try: with pytest.raises(ReadTimeoutError): response.read() finally: timed_out.set() # Second should succeed. response = pool.urlopen( "GET", "/", retries=0, preload_content=False, timeout=LONG_TIMEOUT ) assert len(response.read()) == 8 def test_closing_response_actually_closes_connection(self): done_closing = Event() complete = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf = sock.recv(65536) sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 0\r\n" b"\r\n" ) # Wait for the socket to close. done_closing.wait(timeout=LONG_TIMEOUT) # Look for the empty string to show that the connection got closed. # Don't get stuck in a timeout. sock.settimeout(LONG_TIMEOUT) new_data = sock.recv(65536) assert not new_data sock.close() complete.set() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: response = pool.request("GET", "/", retries=0, preload_content=False) assert response.status == 200 response.close() done_closing.set() # wait until the socket in our pool gets closed successful = complete.wait(timeout=LONG_TIMEOUT) assert successful, "Timed out waiting for connection close" def test_release_conn_param_is_respected_after_timeout_retry(self): """For successful ```urlopen(release_conn=False)```, the connection isn't released, even after a retry. This test allows a retry: one request fails, the next request succeeds. This is a regression test for issue #651 [1], where the connection would be released if the initial request failed, even if a retry succeeded. [1] """ def socket_handler(listener): sock = listener.accept()[0] consume_socket(sock) # Close the connection, without sending any response (not even the # HTTP status line). This will trigger a `Timeout` on the client, # inside `urlopen()`. sock.close() # Expect a new request. Because we don't want to hang this thread, # we actually use select.select to confirm that a new request is # coming in: this lets us time the thread out. rlist, _, _ = select.select([listener], [], [], 5) assert rlist sock = listener.accept()[0] consume_socket(sock) # Send complete chunked response. sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Transfer-Encoding: chunked\r\n" b"\r\n" b"8\r\n" b"12345678\r\n" b"0\r\n\r\n" ) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port, maxsize=1) as pool: # First request should fail, but the timeout and `retries=1` should # save it. response = pool.urlopen( "GET", "/", retries=1, release_conn=False, preload_content=False, timeout=LONG_TIMEOUT, ) # The connection should still be on the response object, and none # should be in the pool. We opened two though. assert pool.num_connections == 2 assert pool.pool.qsize() == 0 assert response.connection is not None # Consume the data. This should put the connection back. response.read() assert pool.pool.qsize() == 1 assert response.connection is None class TestProxyManager(SocketDummyServerTestCase): def test_simple(self): def echo_socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(buf), buf.decode("utf-8")) ).encode("utf-8") ) sock.close() self._start_server(echo_socket_handler) base_url = f"http://{self.host}:{self.port}" with proxy_from_url(base_url) as proxy: r = proxy.request("GET", "http://google.com/") assert r.status == 200 # FIXME: The order of the headers is not predictable right now. We # should fix that someday (maybe when we migrate to # OrderedDict/MultiDict). assert sorted(r.data.split(b"\r\n")) == sorted( [ b"GET http://google.com/ HTTP/1.1", b"Host: google.com", b"Accept-Encoding: identity", b"Accept: */*", b"User-Agent: " + _get_default_user_agent().encode("utf-8"), b"", b"", ] ) def test_headers(self): def echo_socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(buf), buf.decode("utf-8")) ).encode("utf-8") ) sock.close() self._start_server(echo_socket_handler) base_url = f"http://{self.host}:{self.port}" # Define some proxy headers. proxy_headers = HTTPHeaderDict({"For The Proxy": "YEAH!"}) with proxy_from_url(base_url, proxy_headers=proxy_headers) as proxy: conn = proxy.connection_from_url("http://www.google.com/") r = conn.urlopen("GET", "http://www.google.com/", assert_same_host=False) assert r.status == 200 # FIXME: The order of the headers is not predictable right now. We # should fix that someday (maybe when we migrate to # OrderedDict/MultiDict). assert b"For The Proxy: YEAH!\r\n" in r.data def test_retries(self): close_event = Event() def echo_socket_handler(listener): sock = listener.accept()[0] # First request, which should fail sock.close() # Second request sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( ( "HTTP/1.1 200 OK\r\n" "Content-Type: text/plain\r\n" "Content-Length: %d\r\n" "\r\n" "%s" % (len(buf), buf.decode("utf-8")) ).encode("utf-8") ) sock.close() close_event.set() self._start_server(echo_socket_handler) base_url = f"http://{self.host}:{self.port}" with proxy_from_url(base_url) as proxy: conn = proxy.connection_from_url("http://www.google.com") r = conn.urlopen( "GET", "http://www.google.com", assert_same_host=False, retries=1 ) assert r.status == 200 close_event.wait(timeout=LONG_TIMEOUT) with pytest.raises(ProxyError): conn.urlopen( "GET", "http://www.google.com", assert_same_host=False, retries=False, ) def test_connect_reconn(self): def proxy_ssl_one(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) s = buf.decode("utf-8") if not s.startswith("CONNECT "): sock.send(b"HTTP/1.1 405 Method not allowed\r\nAllow: CONNECT\r\n\r\n") sock.close() return if not s.startswith(f"CONNECT {self.host}:443"): sock.send(b"HTTP/1.1 403 Forbidden\r\n\r\n") sock.close() return sock.send(b"HTTP/1.1 200 Connection Established\r\n\r\n") ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ca_certs=DEFAULT_CA, ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 2\r\n" b"Connection: close\r\n" b"\r\n" b"Hi" ) ssl_sock.close() def echo_socket_handler(listener): proxy_ssl_one(listener) proxy_ssl_one(listener) self._start_server(echo_socket_handler) base_url = f"http://{self.host}:{self.port}" with proxy_from_url(base_url, ca_certs=DEFAULT_CA) as proxy: url = f"https://{self.host}" conn = proxy.connection_from_url(url) r = conn.urlopen("GET", url, retries=0) assert r.status == 200 r = conn.urlopen("GET", url, retries=0) assert r.status == 200 def test_connect_ipv6_addr(self): ipv6_addr = "2001:4998:c:a06::2:4008" def echo_socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) s = buf.decode("utf-8") if s.startswith(f"CONNECT [{ipv6_addr}]:443"): sock.send(b"HTTP/1.1 200 Connection Established\r\n\r\n") ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 2\r\n" b"Connection: close\r\n" b"\r\n" b"Hi" ) ssl_sock.close() else: sock.close() self._start_server(echo_socket_handler) base_url = f"http://{self.host}:{self.port}" with proxy_from_url(base_url, cert_reqs="NONE") as proxy: url = f"https://[{ipv6_addr}]" conn = proxy.connection_from_url(url) try: r = conn.urlopen("GET", url, retries=0) assert r.status == 200 except MaxRetryError: self.fail("Invalid IPv6 format in HTTP CONNECT request") class TestSSL(SocketDummyServerTestCase): def test_ssl_failure_midway_through_conn(self): def socket_handler(listener): sock = listener.accept()[0] sock2 = sock.dup() ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ca_certs=DEFAULT_CA, ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) # Deliberately send from the non-SSL socket. sock2.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 2\r\n" b"\r\n" b"Hi" ) sock2.close() ssl_sock.close() self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port) as pool: with pytest.raises(MaxRetryError) as cm: pool.request("GET", "/", retries=0) assert isinstance(cm.value.reason, SSLError) @notSecureTransport def test_ssl_read_timeout(self): timed_out = Event() def socket_handler(listener): sock = listener.accept()[0] ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) # Send incomplete message (note Content-Length) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 10\r\n" b"\r\n" b"Hi-" ) timed_out.wait() sock.close() ssl_sock.close() self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port, ca_certs=DEFAULT_CA) as pool: response = pool.urlopen( "GET", "/", retries=0, preload_content=False, timeout=LONG_TIMEOUT ) try: with pytest.raises(ReadTimeoutError): response.read() finally: timed_out.set() def test_ssl_failed_fingerprint_verification(self): def socket_handler(listener): for i in range(2): sock = listener.accept()[0] ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ca_certs=DEFAULT_CA, ) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 5\r\n\r\n" b"Hello" ) ssl_sock.close() sock.close() self._start_server(socket_handler) # GitHub's fingerprint. Valid, but not matching. fingerprint = "A0:C4:A7:46:00:ED:A7:2D:C0:BE:CB:9A:8C:B6:07:CA:58:EE:74:5E" def request(): pool = HTTPSConnectionPool( self.host, self.port, assert_fingerprint=fingerprint ) try: timeout = Timeout(connect=LONG_TIMEOUT, read=SHORT_TIMEOUT) response = pool.urlopen( "GET", "/", preload_content=False, retries=0, timeout=timeout ) response.read() finally: pool.close() with pytest.raises(MaxRetryError) as cm: request() assert isinstance(cm.value.reason, SSLError) # Should not hang, see https://github.com/urllib3/urllib3/issues/529 with pytest.raises(MaxRetryError): request() def test_retry_ssl_error(self): def socket_handler(listener): # first request, trigger an SSLError sock = listener.accept()[0] sock2 = sock.dup() ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) # Deliberately send from the non-SSL socket to trigger an SSLError sock2.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 4\r\n" b"\r\n" b"Fail" ) sock2.close() ssl_sock.close() # retried request sock = listener.accept()[0] ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 7\r\n\r\n" b"Success" ) ssl_sock.close() self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port, ca_certs=DEFAULT_CA) as pool: response = pool.urlopen("GET", "/", retries=1) assert response.data == b"Success" def test_ssl_load_default_certs_when_empty(self): def socket_handler(listener): sock = listener.accept()[0] ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ca_certs=DEFAULT_CA, ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 5\r\n\r\n" b"Hello" ) ssl_sock.close() sock.close() context = mock.create_autospec(ssl_.SSLContext) context.load_default_certs = mock.Mock() context.options = 0 with mock.patch("urllib3.util.ssl_.SSLContext", lambda *_, **__: context): self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port) as pool: # Without a proper `SSLContext`, this request will fail in some # arbitrary way, but we only want to know if load_default_certs() was # called, which is why we accept any `Exception` here. with pytest.raises(Exception): pool.request("GET", "/", timeout=SHORT_TIMEOUT) context.load_default_certs.assert_called_with() def test_ssl_dont_load_default_certs_when_given(self): def socket_handler(listener): sock = listener.accept()[0] ssl_sock = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ca_certs=DEFAULT_CA, ) buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += ssl_sock.recv(65536) ssl_sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/plain\r\n" b"Content-Length: 5\r\n\r\n" b"Hello" ) ssl_sock.close() sock.close() context = mock.create_autospec(ssl_.SSLContext) context.load_default_certs = mock.Mock() context.options = 0 with mock.patch("urllib3.util.ssl_.SSLContext", lambda *_, **__: context): for kwargs in [ {"ca_certs": "/a"}, {"ca_cert_dir": "/a"}, {"ca_certs": "a", "ca_cert_dir": "a"}, {"ssl_context": context}, ]: self._start_server(socket_handler) with HTTPSConnectionPool(self.host, self.port, **kwargs) as pool: with pytest.raises(Exception): pool.request("GET", "/", timeout=SHORT_TIMEOUT) context.load_default_certs.assert_not_called() def test_load_verify_locations_exception(self): """ Ensure that load_verify_locations raises SSLError for all backends """ with pytest.raises(SSLError): ssl_wrap_socket(None, ca_certs="/tmp/fake-file") def test_ssl_custom_validation_failure_terminates(self, tmpdir): """ Ensure that the underlying socket is terminated if custom validation fails. """ server_closed = Event() def is_closed_socket(sock): try: sock.settimeout(SHORT_TIMEOUT) except OSError: return True return False def socket_handler(listener): sock = listener.accept()[0] try: _ = ssl.wrap_socket( sock, server_side=True, keyfile=DEFAULT_CERTS["keyfile"], certfile=DEFAULT_CERTS["certfile"], ca_certs=DEFAULT_CA, ) except ssl.SSLError as e: assert "alert unknown ca" in str(e) if is_closed_socket(sock): server_closed.set() self._start_server(socket_handler) # client uses a different ca other_ca = trustme.CA() other_ca_path = str(tmpdir / "ca.pem") other_ca.cert_pem.write_to_path(other_ca_path) with HTTPSConnectionPool( self.host, self.port, cert_reqs="REQUIRED", ca_certs=other_ca_path ) as pool: with pytest.raises(SSLError): pool.request("GET", "/", retries=False, timeout=LONG_TIMEOUT) assert server_closed.wait(LONG_TIMEOUT), "The socket was not terminated" class TestErrorWrapping(SocketDummyServerTestCase): def test_bad_statusline(self): self.start_response_handler( b"HTTP/1.1 Omg What Is This?\r\n" b"Content-Length: 0\r\n" b"\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: with pytest.raises(ProtocolError): pool.request("GET", "/") def test_unknown_protocol(self): self.start_response_handler( b"HTTP/1000 200 OK\r\n" b"Content-Length: 0\r\n" b"\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: with pytest.raises(ProtocolError): pool.request("GET", "/") class TestHeaders(SocketDummyServerTestCase): def test_httplib_headers_case_insensitive(self): self.start_response_handler( b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n" b"Content-type: text/plain\r\n" b"\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: HEADERS = {"Content-Length": "0", "Content-type": "text/plain"} r = pool.request("GET", "/") assert HEADERS == dict(r.headers.items()) # to preserve case sensitivity def start_parsing_handler(self): self.parsed_headers = OrderedDict() self.received_headers = [] def socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) self.received_headers = [ header for header in buf.split(b"\r\n")[1:] if header ] for header in self.received_headers: (key, value) = header.split(b": ") self.parsed_headers[key.decode("ascii")] = value.decode("ascii") sock.send(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n") sock.close() self._start_server(socket_handler) def test_headers_are_sent_with_the_original_case(self): headers = {"foo": "bar", "bAz": "quux"} self.start_parsing_handler() expected_headers = { "Accept-Encoding": "identity", "Host": f"{self.host}:{self.port}", "User-Agent": _get_default_user_agent(), } expected_headers.update(headers) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: pool.request("GET", "/", headers=HTTPHeaderDict(headers)) assert expected_headers == self.parsed_headers def test_ua_header_can_be_overridden(self): headers = {"uSeR-AgENt": "Definitely not urllib3!"} self.start_parsing_handler() expected_headers = { "Accept-Encoding": "identity", "Host": f"{self.host}:{self.port}", } expected_headers.update(headers) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: pool.request("GET", "/", headers=HTTPHeaderDict(headers)) assert expected_headers == self.parsed_headers def test_request_headers_are_sent_in_the_original_order(self): # NOTE: Probability this test gives a false negative is 1/(K!) K = 16 # NOTE: Provide headers in non-sorted order (i.e. reversed) # so that if the internal implementation tries to sort them, # a change will be detected. expected_request_headers = [ (f"X-Header-{int(i)}", str(i)) for i in reversed(range(K)) ] def filter_non_x_headers(d): return [(k, v) for (k, v) in d.items() if k.startswith("X-Header-")] request_headers = OrderedDict() self.start_parsing_handler() with HTTPConnectionPool(self.host, self.port, retries=False) as pool: pool.request("GET", "/", headers=OrderedDict(expected_request_headers)) request_headers = filter_non_x_headers(self.parsed_headers) assert expected_request_headers == request_headers @resolvesLocalhostFQDN def test_request_host_header_ignores_fqdn_dot(self): self.start_parsing_handler() with HTTPConnectionPool(self.host + ".", self.port, retries=False) as pool: pool.request("GET", "/") self.assert_header_received( self.received_headers, "Host", f"{self.host}:{self.port}" ) def test_response_headers_are_returned_in_the_original_order(self): # NOTE: Probability this test gives a false negative is 1/(K!) K = 16 # NOTE: Provide headers in non-sorted order (i.e. reversed) # so that if the internal implementation tries to sort them, # a change will be detected. expected_response_headers = [ (f"X-Header-{int(i)}", str(i)) for i in reversed(range(K)) ] def socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( b"HTTP/1.1 200 OK\r\n" + b"\r\n".join( [ (k.encode("utf8") + b": " + v.encode("utf8")) for (k, v) in expected_response_headers ] ) + b"\r\n" ) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: r = pool.request("GET", "/", retries=0) actual_response_headers = [ (k, v) for (k, v) in r.headers.items() if k.startswith("X-Header-") ] assert expected_response_headers == actual_response_headers class TestBrokenHeaders(SocketDummyServerTestCase): def _test_broken_header_parsing(self, headers, unparsed_data_check=None): self.start_response_handler( ( b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n" b"Content-type: text/plain\r\n" ) + b"\r\n".join(headers) + b"\r\n\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: with LogRecorder() as logs: pool.request("GET", "/") for record in logs: if ( "Failed to parse headers" in record.msg and pool._absolute_url("/") == record.args[0] ): if ( unparsed_data_check is None or unparsed_data_check in record.getMessage() ): return self.fail("Missing log about unparsed headers") def test_header_without_name(self): self._test_broken_header_parsing([b": Value", b"Another: Header"]) def test_header_without_name_or_value(self): self._test_broken_header_parsing([b":", b"Another: Header"]) def test_header_without_colon_or_value(self): self._test_broken_header_parsing( [b"Broken Header", b"Another: Header"], "Broken Header" ) class TestHeaderParsingContentType(SocketDummyServerTestCase): def _test_okay_header_parsing(self, header): self.start_response_handler( (b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n") + header + b"\r\n\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: with LogRecorder() as logs: pool.request("GET", "/") for record in logs: assert "Failed to parse headers" not in record.msg def test_header_text_plain(self): self._test_okay_header_parsing(b"Content-type: text/plain") def test_header_message_rfc822(self): self._test_okay_header_parsing(b"Content-type: message/rfc822") class TestHEAD(SocketDummyServerTestCase): def test_chunked_head_response_does_not_hang(self): self.start_response_handler( b"HTTP/1.1 200 OK\r\n" b"Transfer-Encoding: chunked\r\n" b"Content-type: text/plain\r\n" b"\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: r = pool.request("HEAD", "/", timeout=LONG_TIMEOUT, preload_content=False) # stream will use the read_chunked method here. assert [] == list(r.stream()) def test_empty_head_response_does_not_hang(self): self.start_response_handler( b"HTTP/1.1 200 OK\r\n" b"Content-Length: 256\r\n" b"Content-type: text/plain\r\n" b"\r\n" ) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: r = pool.request("HEAD", "/", timeout=LONG_TIMEOUT, preload_content=False) # stream will use the read method here. assert [] == list(r.stream()) class TestStream(SocketDummyServerTestCase): def test_stream_none_unchunked_response_does_not_hang(self): done_event = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Length: 12\r\n" b"Content-type: text/plain\r\n" b"\r\n" b"hello, world" ) done_event.wait(5) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port, retries=False) as pool: r = pool.request("GET", "/", timeout=LONG_TIMEOUT, preload_content=False) # Stream should read to the end. assert [b"hello, world"] == list(r.stream(None)) done_event.set() class TestBadContentLength(SocketDummyServerTestCase): def test_enforce_content_length_get(self): done_event = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Length: 22\r\n" b"Content-type: text/plain\r\n" b"\r\n" b"hello, world" ) done_event.wait(LONG_TIMEOUT) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port, maxsize=1) as conn: # Test stream read when content length less than headers claim get_response = conn.request( "GET", url="/", preload_content=False, enforce_content_length=True ) data = get_response.stream(100) # Read "good" data before we try to read again. # This won't trigger till generator is exhausted. next(data) try: next(data) assert False except ProtocolError as e: assert "12 bytes read, 10 more expected" in str(e) done_event.set() def test_enforce_content_length_no_body(self): done_event = Event() def socket_handler(listener): sock = listener.accept()[0] buf = b"" while not buf.endswith(b"\r\n\r\n"): buf += sock.recv(65536) sock.send( b"HTTP/1.1 200 OK\r\n" b"Content-Length: 22\r\n" b"Content-type: text/plain\r\n" b"\r\n" ) done_event.wait(1) sock.close() self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port, maxsize=1) as conn: # Test stream on 0 length body head_response = conn.request( "HEAD", url="/", preload_content=False, enforce_content_length=True ) data = [chunk for chunk in head_response.stream(1)] assert len(data) == 0 done_event.set() class TestRetryPoolSizeDrainFail(SocketDummyServerTestCase): def test_pool_size_retry_drain_fail(self): def socket_handler(listener): for _ in range(2): sock = listener.accept()[0] while not sock.recv(65536).endswith(b"\r\n\r\n"): pass # send a response with an invalid content length -- this causes # a ProtocolError to raise when trying to drain the connection sock.send( b"HTTP/1.1 404 NOT FOUND\r\n" b"Content-Length: 1000\r\n" b"Content-Type: text/plain\r\n" b"\r\n" ) sock.close() self._start_server(socket_handler) retries = Retry(total=1, raise_on_status=False, status_forcelist=[404]) with HTTPConnectionPool( self.host, self.port, maxsize=10, retries=retries, block=True ) as pool: pool.urlopen("GET", "/not_found", preload_content=False) assert pool.num_connections == 1 class TestBrokenPipe(SocketDummyServerTestCase): @notWindows def test_ignore_broken_pipe_errors(self, monkeypatch): # On Windows an aborted connection raises an error on # attempts to read data out of a socket that's been closed. sock_shut = Event() orig_connect = HTTPConnection.connect # a buffer that will cause two sendall calls buf = "a" * 1024 * 1024 * 4 def connect_and_wait(*args, **kw): ret = orig_connect(*args, **kw) assert sock_shut.wait(5) return ret def socket_handler(listener): for i in range(2): sock = listener.accept()[0] sock.send( b"HTTP/1.1 404 Not Found\r\n" b"Connection: close\r\n" b"Content-Length: 10\r\n" b"\r\n" b"xxxxxxxxxx" ) sock.shutdown(socket.SHUT_RDWR) sock_shut.set() sock.close() monkeypatch.setattr(HTTPConnection, "connect", connect_and_wait) self._start_server(socket_handler) with HTTPConnectionPool(self.host, self.port) as pool: r = pool.request("POST", "/", body=buf) assert r.status == 404 assert r.headers["content-length"] == "10" assert r.data == b"xxxxxxxxxx" r = pool.request("POST", "/admin", chunked=True, body=buf) assert r.status == 404 assert r.headers["content-length"] == "10" assert r.data == b"xxxxxxxxxx" class TestMultipartResponse(SocketDummyServerTestCase): def test_multipart_assert_header_parsing_no_defects(self): def socket_handler(listener): for _ in range(2): sock = listener.accept()[0] while not sock.recv(65536).endswith(b"\r\n\r\n"): pass sock.sendall( b"HTTP/1.1 404 Not Found\r\n" b"Server: example.com\r\n" b"Content-Type: multipart/mixed; boundary=36eeb8c4e26d842a\r\n" b"Content-Length: 73\r\n" b"\r\n" b"--36eeb8c4e26d842a\r\n" b"Content-Type: text/plain\r\n" b"\r\n" b"1\r\n" b"--36eeb8c4e26d842a--\r\n", ) sock.close() self._start_server(socket_handler) from urllib3.connectionpool import log with mock.patch.object(log, "warning") as log_warning: with HTTPConnectionPool(self.host, self.port, timeout=3) as pool: resp = pool.urlopen("GET", "/") assert resp.status == 404 assert ( resp.headers["content-type"] == "multipart/mixed; boundary=36eeb8c4e26d842a" ) assert len(resp.data) == 73 log_warning.assert_not_called()