Fix pass_through with HTTP proxy

respx/mocks.py

@@ -166,6 +166,10 @@ def _transport_for_url(self, *args, **kwargs):
 
 
 class AbstractRequestMocker(Mocker):
+    @classmethod
+    def _should_bypass(cls, **kwargs) -> bool:
+        return False
+
     @classmethod
     def mock(cls, spec):
         if spec.__name__ not in cls.target_methods:
@@ -176,6 +180,8 @@ def mock(cls, spec):
 
         def mock(self, *args, **kwargs):
             kwargs = cls._merge_args_and_kwargs(argspec, args, kwargs)
+            if cls._should_bypass(**kwargs):
+                return spec(self, **kwargs)
             request = cls.to_httpx_request(**kwargs)
             request, kwargs = cls.prepare_sync_request(request, **kwargs)
             response = cls._send_sync_request(
@@ -185,6 +191,8 @@ def mock(self, *args, **kwargs):
 
         async def amock(self, *args, **kwargs):
             kwargs = cls._merge_args_and_kwargs(argspec, args, kwargs)
+            if cls._should_bypass(**kwargs):
+                return await spec(self, **kwargs)
             request = cls.to_httpx_request(**kwargs)
             request, kwargs = await cls.prepare_async_request(request, **kwargs)
             response = await cls._send_async_request(
@@ -271,6 +279,17 @@ class HTTPCoreMocker(AbstractRequestMocker):
     ]
     target_methods = ["handle_request", "handle_async_request"]
 
+    @classmethod
+    def _should_bypass(cls, **kwargs) -> bool:
+        # Bypass CONNECT requests because we cannot mock proxy tunnels:
+        # 1. CONNECT URLs lack a scheme, crashing our URL parser.
+        # 2. Tunnels require a live socket, not a static mock response.
+        request = kwargs.get("request")
+        if request is not None:
+            if request.method in (b"CONNECT", "CONNECT"):
+                return True
+        return False
+
     @classmethod
     def prepare_sync_request(cls, httpx_request, **kwargs):
         """

tests/test_api.py

@@ -418,6 +418,33 @@ async def test_pass_through(client, using, route, expected):
         assert request.is_pass_through is expected
 
 
+async def test_pass_through_with_proxy():
+    # Sync
+    with respx.mock:
+        route = respx.get("https://foo.bar/").pass_through()
+        with mock.patch(
+            "socket.create_connection", side_effect=socket.error("test request blocked")
+        ) as connect:
+            with pytest.raises(httpx.NetworkError):
+                with httpx.Client(proxy="https://1.1.1.1:1") as client:
+                    client.get("https://foo.bar/")
+        assert connect.called is True
+        assert route.called is True
+
+    # Async
+    async with respx.mock:
+        route = respx.get("https://foo.bar/").pass_through()
+        with mock.patch(
+            "anyio.connect_tcp",
+            side_effect=ConnectionRefusedError("test request blocked"),
+        ) as open_connection:
+            with pytest.raises(httpx.NetworkError):
+                async with httpx.AsyncClient(proxy="https://1.1.1.1:1") as client:
+                    await client.get("https://foo.bar/")
+        assert open_connection.called is True
+        assert route.called is True
+
+
 @respx.mock
 async def test_parallel_requests(client):
     def content(request, page):