🐛 relax hard assert for response type in the synchronous code

partially resolve #320

Author: Ousret

mypy-requirements.txt

@@ -1,4 +1,4 @@
-mypy==1.13.0; python_version >= '3.8'
+mypy==1.19.1; python_version >= '3.8'
 mypy==1.4.1; python_version < '3.8'
 idna>=2.0.0
 cryptography<47
@@ -15,3 +15,4 @@ wsproto>=1.2,<2
 aiofile>=3.9.0,<4
 zstandard>=0.18.0
 wsproto
+types-PySocks

src/urllib3/_async/connection.py

@@ -813,7 +813,7 @@ async def connect(self) -> None:
                 self.assert_hostname,
                 self.cert_reqs,
             )
-            is NotImplemented
+            is False
         ):
             server_hostname: str = self.host
             tls_in_tls = False

src/urllib3/_collections.py

@@ -357,7 +357,6 @@ def extend(self, *args: ValidHTTPHeaderSource, **kwargs: str) -> None:
             for key, val in other.items():
                 self.add(key, val)
         elif isinstance(other, typing.Iterable):
-            other = typing.cast(typing.Iterable[typing.Tuple[str, str]], other)
             for key, value in other:
                 self.add(key, value)
         elif hasattr(other, "keys") and hasattr(other, "__getitem__"):

src/urllib3/backend/_async/hface.py

@@ -14,9 +14,9 @@
 
 
 try:  # We shouldn't do this, it is private. Only for chain extraction check. We should find another way.
-    from _ssl import Certificate  # type: ignore[import-not-found]
+    from _ssl import Certificate
 except (ImportError, AttributeError):
-    Certificate = None
+    Certificate = None  # type: ignore[misc,assignment]
 
 from ..._collections import HTTPHeaderDict
 from ..._constant import (
@@ -311,8 +311,8 @@ def _custom_tls(
         cert_reqs: int | str | None = None,
     ) -> bool:
         """Meant to support TLS over QUIC meanwhile cpython does not ship with its native implementation."""
-        if self._svn != HttpVersion.h3:
-            return NotImplemented
+        if self._svn is not HttpVersion.h3:
+            return False
 
         cert_use_common_name = False
 
@@ -532,7 +532,7 @@ async def _post_conn(self) -> None:  # type: ignore[override]
                         self.conn_info.issuer_certificate_der = (
                             ssl.PEM_cert_to_DER_cert(chain[1].public_bytes())
                         )
-                        self.conn_info.issuer_certificate_dict = chain[1].get_info()
+                        self.conn_info.issuer_certificate_dict = chain[1].get_info()  # type: ignore[assignment]
 
             if cipher_tuple:
                 self.conn_info.cipher = cipher_tuple[0]

src/urllib3/backend/hface.py

@@ -19,9 +19,9 @@
 
 
 try:  # We shouldn't do this, it is private. Only for chain extraction check. We should find another way.
-    from _ssl import Certificate  # type: ignore[import-not-found]
+    from _ssl import Certificate
 except (ImportError, AttributeError):
-    Certificate = None
+    Certificate = None  # type: ignore[misc,assignment]
 
 from .._collections import HTTPHeaderDict
 from .._constant import (
@@ -336,8 +336,8 @@ def _custom_tls(
         cert_reqs: int | str | None = None,
     ) -> bool:
         """Meant to support TLS over QUIC meanwhile cpython does not ship with its native implementation."""
-        if self._svn != HttpVersion.h3:
-            return NotImplemented
+        if self._svn is not HttpVersion.h3:
+            return False
 
         cert_use_common_name = False
 
@@ -574,7 +574,7 @@ def _post_conn(self) -> None:
                         self.conn_info.issuer_certificate_der = (
                             ssl.PEM_cert_to_DER_cert(chain[1].public_bytes())
                         )
-                        self.conn_info.issuer_certificate_dict = chain[1].get_info()
+                        self.conn_info.issuer_certificate_dict = chain[1].get_info()  # type: ignore[assignment]
 
             elif hasattr(self.sock, "getpeercert"):
                 self.conn_info.certificate_der = self.sock.getpeercert(binary_form=True)
@@ -608,7 +608,7 @@ def _post_conn(self) -> None:
                         self.conn_info.issuer_certificate_der = (
                             ssl.PEM_cert_to_DER_cert(chain[1].public_bytes())
                         )
-                        self.conn_info.issuer_certificate_dict = chain[1].get_info()
+                        self.conn_info.issuer_certificate_dict = chain[1].get_info()  # type: ignore[assignment]
 
             if cipher_tuple:
                 self.conn_info.cipher = cipher_tuple[0]

src/urllib3/connection.py

@@ -791,7 +791,7 @@ def connect(self) -> None:
                 self.assert_hostname,
                 self.cert_reqs,
             )
-            is NotImplemented
+            is False
         ):
             server_hostname: str = self.host
             tls_in_tls = False

src/urllib3/connectionpool.py

@@ -1894,7 +1894,12 @@ def urlopen(
             assert isinstance(response, ResponsePromise)
             return response  # actually a response promise!
 
-        assert isinstance(response, HTTPResponse)
+        # we are relaxing that constraint to behave properly
+        # with 3rd party mocking tool such as vcrpy.
+        # see https://github.com/jawah/urllib3.future/issues/320
+        # for more details.
+        # assert isinstance(response, HTTPResponse)
+        response = typing.cast(HTTPResponse, response)
 
         if redirect and response.get_redirect_location():
             # Handle redirect?

src/urllib3/contrib/_socks_legacy.py

@@ -41,7 +41,7 @@
 from __future__ import annotations
 
 try:
-    import socks  # type: ignore[import-not-found]
+    import socks
 except ImportError:
     import warnings
 
@@ -113,13 +113,13 @@ def _new_conn(self) -> socks.socksocket:
         try:
             conn = socks.create_connection(
                 (self.host, self.port),
-                proxy_type=self._socks_options["socks_version"],
+                proxy_type=self._socks_options["socks_version"],  # type: ignore[arg-type]
                 proxy_addr=self._socks_options["proxy_host"],
-                proxy_port=self._socks_options["proxy_port"],
+                proxy_port=self._socks_options["proxy_port"],  # type: ignore[arg-type]
                 proxy_username=self._socks_options["username"],
                 proxy_password=self._socks_options["password"],
                 proxy_rdns=self._socks_options["rdns"],
-                timeout=self.timeout,
+                timeout=self.timeout,  # type: ignore[arg-type]
                 **extra_kw,
             )
 

src/urllib3/contrib/hface/protocols/_protocols.py

@@ -329,7 +329,7 @@ def multiplexed(self) -> bool:
     def should_wait_remote_flow_control(
         self, stream_id: int, amt: int | None = None
     ) -> bool | None:
-        return NotImplemented
+        return NotImplemented  # type: ignore[no-any-return]
 
 
 class HTTP2Protocol(HTTPOverTCPProtocol, metaclass=ABCMeta):

src/urllib3/contrib/imcc/_ctypes.py

@@ -220,7 +220,7 @@ def pull_error(self) -> typing.NoReturn:
 #
 class PySSLContextStruct(ctypes.Structure):
     _fields_ = (
-        _head_extra_fields
+        _head_extra_fields  # type: ignore[assignment]
         + [
             ("ob_refcnt", ctypes.c_ssize_t),  # Py_ssize_t ob_refcnt;
             ("ob_type", ctypes.c_void_p),  # PyTypeObject *ob_type;