requests 2.34.2: Verify requests 2.34.2 HTTPAdapter connection pooling, retry policy integration with urllib3 Retry, and status-based retry exceptions without real network IO

# Requests 2.34.2 `HTTPAdapter` Connection Pooling and Retry Policy Contract

## Context
`requests.adapters.HTTPAdapter` is the default transport adapter mounted on `http://` and `https://` schemes in `requests.Session`. It manages `urllib3.PoolManager`, controls connection reuse, translates requests and responses across the requests-urllib3 boundary, and configures retry mechanics.

## Key Verified Behaviors
- **Default Configuration**: `HTTPAdapter()` defaults to `DEFAULT_POOLSIZE=10` (`pool_connections` and `pool_maxsize`), non-blocking pools (`DEFAULT_POOLBLOCK=False`), and `DEFAULT_RETRIES=0` wrapped in `urllib3.util.Retry(total=0, connect=None, read=False, redirect=None, status=None)`.
- **Custom Retry Integration**: Providing an explicit `urllib3.util.Retry` instance (with custom backoff factors, status code retry lists like `status_forcelist=[500, 502, 503]`, and total counts) passes through directly to urllib3 connection pools. Passing an integer `max_retries=N` initializes `urllib3.util.Retry.from_int(N)`.
- **Connection Pool Provisioning**: `adapter.get_connection(url)` routes URLs to scheme-specific connection pools (`HTTPConnectionPool` for `http://`, `HTTPSConnectionPool` for `https://`) with matching port and host, reusing existing pool instances for subsequent requests to identical hosts.
- **Response Reconstruction**: `adapter.build_response(req, resp)` converts low-level `urllib3.response.HTTPResponse` into `requests.Response`, preserving status code, headers in a `CaseInsensitiveDict`, reasons, URL, and linking back the `request` and `connection` adapter.
- **URL Resolution**: `adapter.request_url(req, proxies)` extracts relative `path_url` for direct and HTTPS proxy requests, but preserves the absolute URL when routing through an HTTP proxy.
- **Session Mounting Precedence**: Custom `HTTPAdapter` instances mounted to specific prefixes (e.g. `session.mount('https://api.example.org', adapter)`) take precedence over the default `https://` catch-all adapter.
- **TLS Context Dispatch**: In `requests>=2.32`, `HTTPAdapter.send()` retrieves the pool connection via `get_connection_with_tls_context` using request-specific TLS context attributes.
- **Retry Exception Hierarchy Finding**: When urllib3 exhausts status-based retries (`MaxRetryError` with `ResponseError`), `HTTPAdapter.send()` raises `requests.exceptions.RetryError`. Crucially, `RetryError` inherits directly from `requests.exceptions.RequestException` and is **not** a subclass of `requests.exceptions.ConnectionError`. Catch blocks targeting `requests.exceptions.ConnectionError` will not intercept exhausted status retry errors.

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: Verify requests 2.34.2 HTTPAdapter connection pooling, retry policy integration with urllib3 Retry, and status-based retry exceptions without real network IO
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/requests@2.34.2
Demonstrate these symbols/APIs:
  - requests.adapters.HTTPAdapter
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - packageManager: npm@10.9.8
  - runtime: node@22.23.2

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:bcbf7bca15021584e139ca4ac26a1c244c7e15c64b32bee405a0bdb8f2df6911","constraints":{"executionContext":"python"},"contract":["assert HTTPAdapter initializes with default pool size of 10 and a zero-retry urllib3 Retry instance","assert custom urllib3.util.Retry configuration and pool sizes are preserved and configured on the pool manager","assert HTTPAdapter.get_connection provisions HTTPConnectionPool and HTTPSConnectionPool matching target scheme and reuses instances for the same host","assert HTTPAdapter.build_response converts urllib3 HTTPResponse into requests.Response with CaseInsensitiveDict headers and attached request reference","assert HTTPAdapter.request_url returns relative path_url for direct requests and full URL for HTTP proxy requests","assert Session.mount gives custom HTTPAdapter prefix precedence over default adapters","assert HTTPAdapter.send translates urllib3 MaxRetryError caused by ResponseError into requests.exceptions.RetryError","assert RetryError inherits directly from RequestException and is not a subclass of ConnectionError"],"goal":"Verify requests 2.34.2 HTTPAdapter connection pooling, retry policy integration with urllib3 Retry, and status-based retry exceptions without real network IO","kind":"HOW","packages":["pkg:pypi/requests@2.34.2"],"schemaVersion":1,"symbols":["requests.adapters.HTTPAdapter","requests.adapters.HTTPAdapter.init_poolmanager","requests.adapters.HTTPAdapter.get_connection","requests.adapters.HTTPAdapter.get_connection_with_tls_context","requests.adapters.HTTPAdapter.build_response","requests.adapters.HTTPAdapter.request_url","requests.adapters.HTTPAdapter.close","requests.adapters.DEFAULT_POOLSIZE","requests.adapters.DEFAULT_RETRIES","requests.adapters.DEFAULT_POOLBLOCK"]},"contractCommand":["python3","test/contract.py"],"environment":{"arch":"x64","distro":"alpine","ecosystem":"pypi","executionContext":"python","language":"python","libc":"musl","moduleSystem":"standard","os":"linux","osVersionBucket":"24","packageManager":"pip","packageManagerVersion":"24.0","runtime":"python","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/requests@2.34.2"],"schemaVersion":1,"subject":"pkg:pypi/requests@2.34.2","symbols":["requests.adapters.HTTPAdapter","requests.adapters.HTTPAdapter.init_poolmanager","requests.adapters.HTTPAdapter.get_connection","requests.adapters.HTTPAdapter.get_connection_with_tls_context","requests.adapters.HTTPAdapter.build_response","requests.adapters.HTTPAdapter.request_url","requests.adapters.HTTPAdapter.close","requests.adapters.DEFAULT_POOLSIZE","requests.adapters.DEFAULT_RETRIES","requests.adapters.DEFAULT_POOLBLOCK"],"verifierAdapter":"python@1"}

[project]
name = "verify-requests-http-adapter"
version = "0.1.0"
dependencies = [
    "requests==2.34.2",
    "urllib3==2.6.3",
]

certifi==2026.2.25
charset-normalizer==3.4.7
idna==3.18
requests==2.34.2
urllib3==2.6.3

{
  "schemaVersion": 1,
  "goal": "Verify requests 2.34.2 HTTPAdapter connection pooling, retry policy integration with urllib3 Retry, and status-based retry exceptions without real network IO",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/requests@2.34.2"
  ],
  "symbols": [
    "requests.adapters.HTTPAdapter"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "packageManager": "npm@10.9.8",
    "runtime": "node@22.23.2"
  }
}

import io
from importlib.metadata import version
from unittest.mock import MagicMock

import requests
from requests import PreparedRequest, Request, Response, Session
from requests.adapters import (
    DEFAULT_POOLBLOCK,
    DEFAULT_POOLSIZE,
    DEFAULT_RETRIES,
    HTTPAdapter,
)
from requests.exceptions import (
    ConnectionError as RequestsConnectionError,
    ConnectTimeout,
    RequestException,
    RetryError,
    SSLError as RequestsSSLError,
)
from requests.structures import CaseInsensitiveDict
from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
from urllib3.exceptions import (
    ConnectTimeoutError,
    MaxRetryError,
    ResponseError,
    SSLError as Urllib3SSLError,
)
from urllib3.poolmanager import PoolManager
from urllib3.response import HTTPResponse
from urllib3.util import Retry

EXPECTED_VERSION = "2.34.2"


def test_default_initialization() -> None:
    adapter = HTTPAdapter()
    assert adapter._pool_connections == DEFAULT_POOLSIZE
    assert adapter._pool_maxsize == DEFAULT_POOLSIZE
    assert adapter._pool_block == DEFAULT_POOLBLOCK
    assert isinstance(adapter.max_retries, Retry)
    assert adapter.max_retries.total == DEFAULT_RETRIES
    assert adapter.max_retries.read is False
    assert isinstance(adapter.poolmanager, PoolManager)
    adapter.close()


def test_custom_retry_configuration() -> None:
    custom_retry = Retry(
        total=5,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503],
    )
    adapter = HTTPAdapter(
        pool_connections=20,
        pool_maxsize=40,
        max_retries=custom_retry,
        pool_block=True,
    )
    assert adapter._pool_connections == 20
    assert adapter._pool_maxsize == 40
    assert adapter._pool_block is True
    assert adapter.max_retries is custom_retry
    assert adapter.max_retries.total == 5
    assert set(adapter.max_retries.status_forcelist) == {500, 502, 503}
    adapter.close()

    # Passing an integer converts to urllib3 Retry
    adapter_int = HTTPAdapter(max_retries=4)
    assert isinstance(adapter_int.max_retries, Retry)
    assert adapter_int.max_retries.total == 4
    adapter_int.close()


def test_connection_pool_routing_and_reuse() -> None:
    adapter = HTTPAdapter(pool_connections=5, pool_maxsize=10, pool_block=True)

    # HTTP connection pool
    conn_http = adapter.get_connection("http://api.example.com/v1/data")
    assert isinstance(conn_http, HTTPConnectionPool)
    assert conn_http.host == "api.example.com"
    assert conn_http.port == 80
    assert conn_http.scheme == "http"
    assert conn_http.pool.maxsize == 10
    assert conn_http.block is True

    # HTTPS connection pool
    conn_https = adapter.get_connection("https://secure.example.com:8443/auth")
    assert isinstance(conn_https, HTTPSConnectionPool)
    assert conn_https.host == "secure.example.com"
    assert conn_https.port == 8443
    assert conn_https.scheme == "https"

    # Same target host reuses identical connection pool instance
    conn_http_again = adapter.get_connection("http://api.example.com/v1/other")
    assert conn_http is conn_http_again

    adapter.close()


def test_build_response() -> None:
    adapter = HTTPAdapter()
    req = Request("GET", "https://api.example.com/resource?key=val").prepare()
    assert isinstance(req, PreparedRequest)

    body_bytes = b'{"status": "success", "items": [1, 2, 3]}'
    mock_raw = HTTPResponse(
        body=io.BytesIO(body_bytes),
        headers={"Content-Type": "application/json", "X-Server-Id": "node-1"},
        status=200,
        reason="OK",
        preload_content=False,
    )

    response = adapter.build_response(req, mock_raw)
    assert isinstance(response, Response)
    assert response.status_code == 200
    assert response.reason == "OK"
    assert response.url == "https://api.example.com/resource?key=val"
    assert response.request is req
    assert response.connection is adapter
    assert isinstance(response.headers, CaseInsensitiveDict)
    assert response.headers["Content-Type"] == "application/json"
    assert response.headers["x-server-id"] == "node-1"
    assert response.raw is mock_raw
    assert response.content == body_bytes
    assert response.json() == {"status": "success", "items": [1, 2, 3]}
    adapter.close()


def test_request_url_formatting() -> None:
    adapter = HTTPAdapter()
    req_direct = Request("GET", "https://example.com/api/test?filter=true").prepare()

    # Direct request or HTTPS proxy: uses path_url
    assert adapter.request_url(req_direct, proxies=None) == "/api/test?filter=true"
    assert adapter.request_url(req_direct, proxies={}) == "/api/test?filter=true"
    assert (
        adapter.request_url(req_direct, proxies={"https": "https://proxy.example.com:8080"})
        == "/api/test?filter=true"
    )

    # HTTP request through HTTP proxy: uses full absolute URL
    req_http = Request("GET", "http://example.com/api/test?filter=true").prepare()
    assert (
        adapter.request_url(req_http, proxies={"http": "http://proxy.example.com:8080"})
        == "http://example.com/api/test?filter=true"
    )

    adapter.close()


def test_session_adapter_mounting() -> None:
    session = Session()
    default_https_adapter = session.adapters["https://"]
    assert isinstance(default_https_adapter, HTTPAdapter)

    custom_adapter = HTTPAdapter(pool_connections=30, pool_maxsize=60)
    session.mount("https://api.example.org", custom_adapter)

    # Prefix match dispatches to mounted custom adapter
    matched = session.get_adapter("https://api.example.org/v1/endpoint")
    assert matched is custom_adapter
    assert matched._pool_connections == 30
    assert matched._pool_maxsize == 60

    # Unmatched host falls back to default https:// adapter
    unmatched = session.get_adapter("https://other.example.org/v1/endpoint")
    assert unmatched is default_https_adapter

    session.close()


def test_retry_exception_translation_and_type_hierarchy() -> None:
    # Verify exception type hierarchy contract
    assert issubclass(RetryError, RequestException)
    assert not issubclass(RetryError, RequestsConnectionError)

    adapter = HTTPAdapter(max_retries=Retry(total=2, status_forcelist=[503]))
    mock_pool = MagicMock()
    mock_pool.urlopen.side_effect = MaxRetryError(
        pool=mock_pool,
        url="/service",
        reason=ResponseError("too many 503 error responses"),
    )
    adapter.get_connection_with_tls_context = MagicMock(return_value=mock_pool)
    req = Request("GET", "http://example.com/service").prepare()

    # When urllib3 raises MaxRetryError with ResponseError, HTTPAdapter raises RetryError
    try:
        adapter.send(req)
        assert False, "Expected RetryError to be raised"
    except RetryError as exc:
        assert isinstance(exc, RequestException)
        assert not isinstance(exc, RequestsConnectionError)
        assert "too many 503 error responses" in str(exc)

    # Test ConnectTimeoutError translation to ConnectTimeout (subclass of RequestsConnectionError)
    mock_pool.urlopen.side_effect = MaxRetryError(
        pool=mock_pool,
        url="/service",
        reason=ConnectTimeoutError(mock_pool, "Connection timed out"),
    )
    try:
        adapter.send(req)
        assert False, "Expected ConnectTimeout to be raised"
    except ConnectTimeout as exc:
        assert isinstance(exc, RequestsConnectionError)

    # Test Urllib3SSLError translation to RequestsSSLError (subclass of RequestsConnectionError)
    mock_pool.urlopen.side_effect = MaxRetryError(
        pool=mock_pool,
        url="/service",
        reason=Urllib3SSLError("certificate verify failed"),
    )
    try:
        adapter.send(req)
        assert False, "Expected RequestsSSLError to be raised"
    except RequestsSSLError as exc:
        assert isinstance(exc, RequestsConnectionError)
        assert "certificate verify failed" in str(exc)

    adapter.close()


def main() -> None:
    assert version("requests") == EXPECTED_VERSION
    assert requests.__version__ == EXPECTED_VERSION

    test_default_initialization()
    test_custom_retry_configuration()
    test_connection_pool_routing_and_reuse()
    test_build_response()
    test_request_url_formatting()
    test_session_adapter_mounting()
    test_retry_exception_translation_and_type_hierarchy()
    print("CONTRACT PASS: requests.adapters.HTTPAdapter contract verified")


if __name__ == "__main__":
    main()