Fix issue with custom adapters (#643)

Fix issue with custom adapters when positional arguments are sent instead of kwargs. See #642

CHANGES

@@ -2,6 +2,7 @@
 ------
 
 * Updated dependency to urllib3>=2 and requests>=2.30.0. See #635
+* Fixed issue when custom adapters were sending only positional args. See #642
 
 
 0.23.1

responses/__init__.py

@@ -1099,9 +1099,27 @@ def start(self) -> None:
             return
 
         def unbound_on_send(
-            adapter: "HTTPAdapter", request: "PreparedRequest", *a: Any, **kwargs: Any
+            adapter: "HTTPAdapter",
+            request: "PreparedRequest",
+            *args: Any,
+            **kwargs: Any,
         ) -> "models.Response":
-            return self._on_request(adapter, request, *a, **kwargs)
+            if args:
+                # that probably means that the request was sent from the custom adapter
+                # It is fully legit to send positional args from adapter, although,
+                # `requests` implementation does it always with kwargs
+                # See for more info: https://github.com/getsentry/responses/issues/642
+                try:
+                    kwargs["stream"] = args[0]
+                    kwargs["timeout"] = args[1]
+                    kwargs["verify"] = args[2]
+                    kwargs["cert"] = args[3]
+                    kwargs["proxies"] = args[4]
+                except IndexError:
+                    # not all kwargs are required
+                    pass
+
+            return self._on_request(adapter, request, **kwargs)
 
         self._patcher = std_mock.patch(target=self.target, new=unbound_on_send)
         self._patcher.start()

responses/tests/test_responses.py

@@ -801,34 +801,70 @@ def test_base_response_get_response():
         resp.get_response(requests.PreparedRequest())
 
 
-def test_custom_adapter():
-    @responses.activate
-    def run():
-        url = "http://example.com"
-        responses.add(responses.GET, url, body=b"test")
+class TestAdapters:
+    class CustomAdapter(requests.adapters.HTTPAdapter):
+        """Classic custom adapter."""
 
-        calls = [0]
+        def send(self, *a, **k):
+            return super().send(*a, **k)
 
-        class DummyAdapter(requests.adapters.HTTPAdapter):
-            def send(self, *a, **k):
-                calls[0] += 1
-                return super().send(*a, **k)
+    class PositionalArgsAdapter(requests.adapters.HTTPAdapter):
+        """Custom adapter that sends only positional args.
+        See https://github.com/getsentry/responses/issues/642 for more into.
+        """
 
-        # Test that the adapter is actually used
-        session = requests.Session()
-        session.mount("http://", DummyAdapter())
+        def send(
+            self,
+            request,
+            stream=False,
+            timeout=None,
+            verify=True,
+            cert=None,
+            proxies=None,
+        ):
+            return super().send(request, stream, timeout, verify, cert, proxies)
+
+    class PositionalArgsIncompleteAdapter(requests.adapters.HTTPAdapter):
+        """Custom adapter that sends only positional args.
+        Not all arguments are forwarded to the send method.
+                    See https://github.com/getsentry/responses/issues/642 for more into.
+        """
 
-        session.get(url, allow_redirects=False)
-        assert calls[0] == 1
+        def send(
+            self,
+            request,
+            stream=False,
+            timeout=None,
+            verify=True,
+            # following args are intentionally not forwarded
+            cert=None,
+            proxies=None,
+        ):
+            return super().send(request, stream, timeout, verify)
+
+    @pytest.mark.parametrize(
+        "adapter_class",
+        (CustomAdapter, PositionalArgsAdapter, PositionalArgsIncompleteAdapter),
+    )
+    def test_custom_adapter(self, adapter_class):
+        """Test basic adapter implementation and that responses can patch them properly."""
 
-        # Test that the response is still correctly emulated
-        session = requests.Session()
-        session.mount("http://", DummyAdapter())
+        @responses.activate
+        def run():
+            url = "http://example.com"
+            responses.add(responses.GET, url, body=b"test adapter")
 
-        resp = session.get(url)
-        assert_response(resp, "test")
+            # Test that the adapter is actually used
+            session = requests.Session()
+            adapter = adapter_class()
+            session.mount("http://", adapter)
+            with patch.object(adapter, "send", side_effect=adapter.send) as mock_send:
+                resp = session.get(url, allow_redirects=False)
 
-    run()
+            assert mock_send.call_count == 1
+            assert_response(resp, "test adapter")
+
+        run()
 
 
 def test_responses_as_context_manager():