fix: apply the session-level proxy when no per-request proxy is given

The per-request proxy resolution never fell back to the session default, so FetcherSession(proxy=...) was silently ignored, and requests went direct. Same fix in the sync and async paths, with regression tests asserting on the proxy that reaches curl_cffi.

Closes #295
This commit is contained in:
Karim shoair
2026-06-07 18:35:18 +03:00
parent 5fdb46fea0
commit 74c5848060
3 changed files with 99 additions and 10 deletions
+45 -6
View File
@@ -1,7 +1,9 @@
import pytest
from unittest.mock import patch, MagicMock
from curl_cffi.curl import CurlError
from scrapling.engines.static import _SyncSessionLogic as FetcherSession, FetcherClient
from scrapling.engines.toolbelt import ProxyRotator
class TestFetcherSession:
@@ -9,11 +11,7 @@ class TestFetcherSession:
def test_fetcher_session_creation(self):
"""Test FetcherSession creation"""
session = FetcherSession(
timeout=30,
retries=3,
stealthy_headers=True
)
session = FetcherSession(timeout=30, retries=3, stealthy_headers=True)
assert session._default_timeout == 30
assert session._default_retries == 3
@@ -43,3 +41,44 @@ class TestFetcherSession:
# Should not have context manager methods
assert client.__enter__ is None
assert client.__exit__ is None
def test_session_level_proxy_is_applied(self):
"""Session-level proxy must reach the request, not be silently dropped (#295)"""
proxy = "http://10.255.255.1:9999"
with FetcherSession(proxy=proxy) as session:
with (
patch.object(session._curl_session, "request") as mocked_request,
patch("scrapling.engines.static.ResponseFactory.from_http_request", return_value=MagicMock()),
):
session.get("http://example.com")
assert mocked_request.call_args.kwargs["proxy"] == proxy
def test_per_request_proxy_overrides_session_proxy(self):
"""A per-request proxy must take precedence over the session-level proxy"""
request_proxy = "http://10.255.255.2:9999"
with FetcherSession(proxy="http://10.255.255.1:9999") as session:
with (
patch.object(session._curl_session, "request") as mocked_request,
patch("scrapling.engines.static.ResponseFactory.from_http_request", return_value=MagicMock()),
):
session.get("http://example.com", proxy=request_proxy)
assert mocked_request.call_args.kwargs["proxy"] == request_proxy
def test_proxy_rotates_per_retry_attempt(self):
"""With a rotator, every retry attempt must pull a fresh proxy"""
rotator = ProxyRotator(["http://p1:8080", "http://p2:8080"])
with FetcherSession(proxy_rotator=rotator, retries=2, retry_delay=0) as session:
with (
patch.object(session._curl_session, "request") as mocked_request,
patch("scrapling.engines.static.ResponseFactory.from_http_request", return_value=MagicMock()),
):
mocked_request.side_effect = [CurlError("transient"), MagicMock()]
session.get("http://example.com")
proxies_used = [call.kwargs["proxy"] for call in mocked_request.call_args_list]
assert proxies_used == ["http://p1:8080", "http://p2:8080"]