urllib3 2.7.0: PoolManager, BaseHTTPResponse

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

Use EXACTLY these public packages and versions:
  - pkg:pypi/urllib3@2.7.0

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:86eecb5bd77abb026f0e0b48c959a16835fc414434d727b2e4f1138746025a19","contract":["urllib3.PoolManager executes HTTP GET requests and retrieves response status and data","urllib3.PoolManager sends HTTP POST requests with JSON payload and custom headers","urllib3.PoolManager handles query parameters and connection pooling across multiple requests"],"goal":"verify pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.PoolManager","urllib3.BaseHTTPResponse"]},"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,"symbols":["urllib3.PoolManager","urllib3.BaseHTTPResponse"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Sample demonstrating HTTP request handling and connection pooling using urllib3.PoolManager."""

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


def create_pool_manager(num_pools: int = 10, maxsize: int = 10) -> urllib3.PoolManager:
    """Create and configure a urllib3.PoolManager instance."""
    return urllib3.PoolManager(num_pools=num_pools, maxsize=maxsize)


def send_get_request(
    http: urllib3.PoolManager,
    url: str,
    headers: Optional[Dict[str, str]] = None,
    fields: Optional[Dict[str, Any]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP GET request with optional headers and query parameters."""
    return http.request("GET", url, headers=headers, fields=fields)


def send_post_json(
    http: urllib3.PoolManager,
    url: str,
    payload: Dict[str, Any],
    headers: Optional[Dict[str, str]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP POST request containing a JSON body."""
    req_headers = {"Content-Type": "application/json"}
    if headers:
        req_headers.update(headers)
    encoded_body = json.dumps(payload).encode("utf-8")
    return http.request("POST", url, headers=req_headers, body=encoded_body)


def parse_response_json(response: BaseHTTPResponse) -> Dict[str, Any]:
    """Decode and parse a JSON response from a BaseHTTPResponse."""
    if response.status < 200 or response.status >= 300:
        raise ValueError(f"HTTP request failed with status {response.status}")
    return json.loads(response.data.decode("utf-8"))

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

"""Contract verification test for urllib3.PoolManager."""

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 sample import (
    create_pool_manager,
    parse_response_json,
    send_get_request,
    send_post_json,
)


class MockHTTPHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed = urlparse(self.path)
        if parsed.path == "/status":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Custom-Header", "test-header-value")
            self.end_headers()
            response_data = {"status": "ok", "message": "server ready"}
            self.wfile.write(json.dumps(response_data).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.end_headers()
            response_data = {"received_query": params_flat}
            self.wfile.write(json.dumps(response_data).encode("utf-8"))
        else:
            self.send_response(404)
            self.end_headers()

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

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

    def log_message(self, format, *args):
        # Suppress logging output to stderr during tests
        pass


def run_test_server():
    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_get_requests_and_json_parsing(base_url: str):
    """Assertion 1: urllib3.PoolManager executes HTTP GET requests and retrieves response status and data."""
    http = create_pool_manager()
    url = f"{base_url}/status"
    resp = send_get_request(http, url)

    assert resp.status == 200, f"Expected 200 OK, got {resp.status}"
    assert resp.headers.get("X-Custom-Header") == "test-header-value"

    data = parse_response_json(resp)
    assert data["status"] == "ok"
    assert data["message"] == "server ready"


def test_post_json_and_headers(base_url: str):
    """Assertion 2: urllib3.PoolManager sends HTTP POST requests with JSON payload and custom headers."""
    http = create_pool_manager()
    url = f"{base_url}/submit"
    payload = {"item": "notebook", "quantity": 5}
    custom_headers = {"X-Request-Id": "req-12345"}

    resp = send_post_json(http, url, payload, headers=custom_headers)

    assert resp.status == 201, f"Expected 201 Created, got {resp.status}"
    data = parse_response_json(resp)
    assert data["created"] is True
    assert data["data"] == payload
    assert data["request_id"] == "req-12345"


def test_query_params_and_pool_reuse(base_url: str):
    """Assertion 3: urllib3.PoolManager handles query parameters and connection pooling across multiple requests."""
    http = create_pool_manager(num_pools=5, maxsize=5)

    # First request with query parameters
    url = f"{base_url}/query"
    params = {"search": "test", "limit": "10"}
    resp1 = send_get_request(http, url, fields=params)
    assert resp1.status == 200
    data1 = parse_response_json(resp1)
    assert data1["received_query"] == {"search": "test", "limit": "10"}

    # Second request reusing the pool
    resp2 = send_get_request(http, f"{base_url}/status")
    assert resp2.status == 200
    data2 = parse_response_json(resp2)
    assert data2["status"] == "ok"

    # Verify pool manager pool count
    assert len(http.pools) >= 1, "PoolManager should have initialized connection pools"


def main():
    server, base_url = run_test_server()
    try:
        test_get_requests_and_json_parsing(base_url)
        test_post_json_and_headers(base_url)
        test_query_params_and_pool_reuse(base_url)
        print("All contract assertions passed successfully.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()