urllib3 2.7.0: urllib3

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 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

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:6309e9cbce93cfe5de0ec7bd7f5b2656721f26c05257d088e6a5fac454bdc271","contract":["urllib3.request executes top-level HTTP GET requests returning BaseHTTPResponse without manual pool management","urllib3.request sends top-level HTTP POST requests with automatic JSON serialization and custom headers","urllib3.request handles query parameters and custom headers with response decoding"],"goal":"verify urllib3 in pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"subject":"pkg:pypi/urllib3@2.7.0","symbols":["urllib3"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Sample demonstrating top-level HTTP request functionality using urllib3."""

import json
from typing import Any, Dict, Optional
import urllib3
from urllib3.response import BaseHTTPResponse


def execute_get_request(
    url: str,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 3.0,
) -> BaseHTTPResponse:
    """Execute a top-level HTTP GET request using urllib3.request."""
    return urllib3.request("GET", url, headers=headers, timeout=timeout)


def execute_post_json(
    url: str,
    json_payload: Any,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 3.0,
) -> BaseHTTPResponse:
    """Execute a top-level HTTP POST request with automated JSON payload serialization."""
    return urllib3.request("POST", url, json=json_payload, headers=headers, timeout=timeout)


def execute_request_with_fields(
    method: str,
    url: str,
    fields: Dict[str, Any],
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 3.0,
) -> BaseHTTPResponse:
    """Execute a top-level HTTP request with query/form fields using urllib3.request."""
    return urllib3.request(method, url, fields=fields, headers=headers, timeout=timeout)


def parse_response_body(response: BaseHTTPResponse) -> Any:
    """Decode and parse the JSON payload from a BaseHTTPResponse."""
    return json.loads(response.data.decode("utf-8"))

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

"""Contract verification test for top-level urllib3 request functionality."""

import json
import os
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import urllib3
from urllib3.response import BaseHTTPResponse
from sample import (
    execute_get_request,
    execute_post_json,
    execute_request_with_fields,
    parse_response_body,
)


class MockHTTPHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/health":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Service-Name", "urllib3-contract-test")
            self.end_headers()
            payload = {"status": "ok", "service": "urllib3-sample"}
            self.wfile.write(json.dumps(payload).encode("utf-8"))
        elif parsed.path == "/query":
            query_params = parse_qs(parsed.query)
            params_flat = {k: v[0] if len(v) == 1 else v for k, v in query_params.items()}
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Echo-Header", self.headers.get("X-Echo-Header", ""))
            self.end_headers()
            payload = {"received_params": params_flat}
            self.wfile.write(json.dumps(payload).encode("utf-8"))
        else:
            self.send_response(404)
            self.end_headers()

    def do_POST(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/submit":
            content_length = int(self.headers.get("Content-Length", 0))
            body_bytes = self.rfile.read(content_length) if content_length > 0 else b""
            content_type = self.headers.get("Content-Type", "")
            data = json.loads(body_bytes.decode("utf-8")) if body_bytes else {}
            client_id = self.headers.get("X-Client-Id", "")

            self.send_response(201)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            payload = {
                "created": True,
                "content_type": content_type,
                "received_data": data,
                "client_id": client_id,
            }
            self.wfile.write(json.dumps(payload).encode("utf-8"))
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format: str, *args: object) -> None:
        # Suppress server access log to stderr during test execution
        pass


def start_mock_server() -> tuple[HTTPServer, str]:
    server = HTTPServer(("127.0.0.1", 0), MockHTTPHandler)
    host, port = server.server_address
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, f"http://127.0.0.1:{port}"


def test_top_level_get_request(base_url: str) -> None:
    """Assertion 1: urllib3.request executes top-level HTTP GET requests returning BaseHTTPResponse without manual pool management."""
    url = f"{base_url}/health"
    response = execute_get_request(url)

    assert isinstance(response, BaseHTTPResponse), "Expected instance of BaseHTTPResponse"
    assert response.status == 200, f"Expected HTTP 200, got {response.status}"
    assert response.headers.get("X-Service-Name") == "urllib3-contract-test"

    body = parse_response_body(response)
    assert body["status"] == "ok"
    assert body["service"] == "urllib3-sample"


def test_top_level_post_json(base_url: str) -> None:
    """Assertion 2: urllib3.request sends top-level HTTP POST requests with automatic JSON serialization and custom headers."""
    url = f"{base_url}/submit"
    payload = {"task": "contract-validation", "retries": 3}
    headers = {"X-Client-Id": "client-abc"}

    response = execute_post_json(url, json_payload=payload, headers=headers)

    assert isinstance(response, BaseHTTPResponse), "Expected instance of BaseHTTPResponse"
    assert response.status == 201, f"Expected HTTP 201 Created, got {response.status}"

    body = parse_response_body(response)
    assert body["created"] is True
    assert "application/json" in body["content_type"]
    assert body["received_data"] == payload
    assert body["client_id"] == "client-abc"


def test_top_level_request_fields_and_headers(base_url: str) -> None:
    """Assertion 3: urllib3.request handles query parameters and custom headers with response decoding."""
    url = f"{base_url}/query"
    fields = {"category": "network", "page": "2"}
    headers = {"X-Echo-Header": "header-passed"}

    response = execute_request_with_fields("GET", url, fields=fields, headers=headers)

    assert isinstance(response, BaseHTTPResponse), "Expected instance of BaseHTTPResponse"
    assert response.status == 200, f"Expected HTTP 200, got {response.status}"
    assert response.headers.get("X-Echo-Header") == "header-passed"

    body = parse_response_body(response)
    assert body["received_params"] == {"category": "network", "page": "2"}


def main() -> None:
    server, base_url = start_mock_server()
    try:
        test_top_level_get_request(base_url)
        test_top_level_post_json(base_url)
        test_top_level_request_fields_and_headers(base_url)
        print("All contract assertions passed successfully.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()