urllib3 2.7.0: util

# urllib3.util API

This sample pins `urllib3` 2.7.0 and verifies core utility components in `urllib3.util`.

The contract verifies:
- `parse_url` parses URL strings into `Url` named tuples, correctly extracting and normalizing components including scheme, host, port, path, query, and user auth.
- `make_headers` generates HTTP headers dictionary supporting basic auth encoding, keep-alive connection flags, user-agent, and cache control directives.
- `Timeout` manages connect, read, and total timeout values, raising `TimeoutStateError` when calculating unstarted read timeout from total.
- `Retry` manages retry attempts, exponential backoff computation, status code forcelists, and allowed HTTP methods, raising `MaxRetryError` when exhausted.
- `SKIP_HEADER` and `SKIPPABLE_HEADERS` provide sentinel values to suppress default request headers such as host, user-agent, and accept-encoding.

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 urllib3.util in pkg:pypi/urllib3@2.7.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/urllib3@2.7.0
Demonstrate these symbols/APIs:
  - urllib3.util

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:54a4f7b12e7fdffd7df0b3934d9eeb81c607e92857b2e9734b301286ab5ae528","contract":["parse_url parses URL strings into Url named tuples, correctly extracting and normalizing components including scheme, host, port, path, query, and user auth.","make_headers generates HTTP headers dictionary supporting basic auth encoding, keep-alive connection flags, user-agent, and cache control directives.","Timeout manages connect, read, and total timeout values, raising TimeoutStateError when calculating unstarted read timeout from total.","Retry manages retry attempts, exponential backoff computation, status code forcelists, and allowed HTTP methods, raising MaxRetryError when exhausted.","SKIP_HEADER and SKIPPABLE_HEADERS provide sentinel values to suppress default request headers such as host, user-agent, and accept-encoding."],"goal":"verify urllib3.util in pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.util"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"subject":"pkg:pypi/urllib3@2.7.0","symbols":["urllib3.util"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Sample demonstrating utility functions and helpers in urllib3.util."""

from typing import Dict, List, Optional
from urllib3.util import (
    SKIP_HEADER,
    SKIPPABLE_HEADERS,
    Retry,
    Timeout,
    Url,
    make_headers,
    parse_url,
)


def parse_endpoint(url: str) -> Url:
    """Parse a URL string into a Url named tuple."""
    return parse_url(url)


def build_request_headers(
    basic_auth: Optional[str] = None,
    keep_alive: bool = True,
    user_agent: Optional[str] = None,
    disable_cache: bool = False,
) -> Dict[str, str]:
    """Construct an HTTP headers mapping using make_headers."""
    return make_headers(
        keep_alive=keep_alive,
        basic_auth=basic_auth,
        user_agent=user_agent,
        disable_cache=disable_cache,
    )


def create_timeout_config(
    total: Optional[float] = None,
    connect: Optional[float] = None,
    read: Optional[float] = None,
) -> Timeout:
    """Create a configured Timeout instance."""
    return Timeout(total=total, connect=connect, read=read)


def create_retry_policy(
    total: int = 3,
    backoff_factor: float = 0.5,
    status_forcelist: Optional[List[int]] = None,
    allowed_methods: Optional[List[str]] = None,
) -> Retry:
    """Create a configured Retry policy."""
    kwargs = {
        "total": total,
        "backoff_factor": backoff_factor,
        "status_forcelist": status_forcelist or [500, 502, 503, 504],
    }
    if allowed_methods is not None:
        kwargs["allowed_methods"] = allowed_methods
    return Retry(**kwargs)

{
  "schemaVersion": 1,
  "goal": "verify urllib3.util in pkg:pypi/urllib3@2.7.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/urllib3@2.7.0"
  ],
  "symbols": [
    "urllib3.util"
  ]
}

"""Contract verification test for urllib3.util in urllib3 2.7.0."""

import base64
import os
import sys

# Ensure sample module can be imported
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import urllib3
from urllib3.exceptions import MaxRetryError, TimeoutStateError
from urllib3.util import (
    SKIP_HEADER,
    SKIPPABLE_HEADERS,
    Retry,
    Timeout,
    Url,
    make_headers,
    parse_url,
)
from sample import (
    build_request_headers,
    create_retry_policy,
    create_timeout_config,
    parse_endpoint,
)


def test_parse_url() -> None:
    # 1. parse_url parses URL strings into Url named tuples, correctly extracting and normalizing components
    url_str = "HTTPS://user:pass@example.com:8443/api/v1/resource?filter=active&sort=desc#details"
    parsed = parse_endpoint(url_str)
    assert isinstance(parsed, Url), "parse_url must return a Url named tuple"
    assert parsed.scheme == "https", f"Scheme must be lowercased, got: {parsed.scheme}"
    assert parsed.auth == "user:pass"
    assert parsed.host == "example.com"
    assert parsed.port == 8443
    assert parsed.path == "/api/v1/resource"
    assert parsed.query == "filter=active&sort=desc"
    assert parsed.fragment == "details"
    assert parsed.request_uri == "/api/v1/resource?filter=active&sort=desc"
    assert parsed.authority == "user:pass@example.com:8443"
    assert parsed.netloc == "example.com:8443"

    # Minimal relative url
    rel = parse_url("/relative/path?q=1")
    assert rel.scheme is None
    assert rel.host is None
    assert rel.path == "/relative/path"
    assert rel.query == "q=1"
    assert rel.request_uri == "/relative/path?q=1"


def test_make_headers() -> None:
    # 2. make_headers generates HTTP headers dictionary supporting basic auth encoding, keep-alive connection flags, user-agent, and cache control directives
    headers = build_request_headers(
        keep_alive=True,
        basic_auth="user:pass",
        user_agent="TestAgent/1.0",
        disable_cache=True,
    )
    expected_auth = "Basic " + base64.b64encode(b"user:pass").decode("latin1")
    assert headers["authorization"] == expected_auth
    assert headers["connection"] == "keep-alive"
    assert headers["user-agent"] == "TestAgent/1.0"
    assert headers["cache-control"] == "no-cache"

    # Accept-encoding
    ae_headers = make_headers(accept_encoding=True)
    assert "accept-encoding" in ae_headers
    assert "gzip" in ae_headers["accept-encoding"]


def test_timeout() -> None:
    # 3. Timeout manages connect, read, and total timeout values, raising TimeoutStateError when calculating unstarted read timeout from total
    t = create_timeout_config(total=10.0, connect=3.0, read=5.0)
    assert t.total == 10.0
    assert t.connect_timeout == 3.0
    assert t.read_timeout == 5.0

    # Clone with overrides
    t_cloned = t.clone()
    assert t_cloned.total == 10.0
    assert t_cloned.connect_timeout == 3.0
    assert t_cloned.read_timeout == 5.0

    # Timeout with total only raises TimeoutStateError when reading read_timeout before start_connect
    t_total_only = Timeout(total=5.0)
    assert t_total_only.connect_timeout == 5.0
    try:
        _ = t_total_only.read_timeout
        raise AssertionError("Expected TimeoutStateError before start_connect")
    except TimeoutStateError:
        pass

    # After start_connect, read_timeout is calculated from remaining duration
    t_total_only.start_connect()
    assert t_total_only.read_timeout <= 5.0
    assert t_total_only.get_connect_duration() >= 0.0


def test_retry() -> None:
    # 4. Retry manages retry attempts, exponential backoff computation, status code forcelists, and allowed HTTP methods, raising MaxRetryError when exhausted
    retry = create_retry_policy(
        total=2,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503],
        allowed_methods=["GET", "PUT"],
    )
    assert retry.total == 2
    assert retry.backoff_factor == 0.5
    assert 500 in retry.status_forcelist
    assert retry.is_retry("GET", status_code=502) is True
    assert retry.is_retry("PUT", status_code=500) is True
    assert retry.is_retry("POST", status_code=500) is False  # not in allowed_methods
    assert retry.is_retry("GET", status_code=200) is False

    # Backoff calculation
    backoff_1 = retry.get_backoff_time()
    assert backoff_1 >= 0.0

    # Increment state and verify decrementing retries
    r1 = retry.increment(method="GET", error=Exception("connection reset"))
    assert r1.total == 1
    r2 = r1.increment(method="GET", error=Exception("connection reset"))
    assert r2.total == 0

    try:
        r2.increment(method="GET", error=Exception("connection reset"))
        raise AssertionError("Expected MaxRetryError when total retries exhausted")
    except MaxRetryError:
        pass


def test_skip_headers() -> None:
    # 5. SKIP_HEADER and SKIPPABLE_HEADERS provide sentinel values to suppress default request headers such as host, user-agent, and accept-encoding
    assert isinstance(SKIP_HEADER, str)
    assert isinstance(SKIPPABLE_HEADERS, (set, frozenset))
    assert "user-agent" in SKIPPABLE_HEADERS
    assert "host" in SKIPPABLE_HEADERS
    assert "accept-encoding" in SKIPPABLE_HEADERS


def main() -> None:
    test_parse_url()
    test_make_headers()
    test_timeout()
    test_retry()
    test_skip_headers()
    print("CONTRACT PASS: urllib3.util verified offline")


if __name__ == "__main__":
    main()