urllib3 2.7.0: connectionpool.HTTPConnectionPool

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.connectionpool.HTTPConnectionPool 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.connectionpool.HTTPConnectionPool

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:9beefea3f6d1bf267eb7f30e28f348bbe81c735d54b38e4bd1b7076fbcd64fa6","contract":["urllib3.connectionpool.HTTPConnectionPool sends HTTP GET requests and decodes response status, body, and headers","urllib3.connectionpool.HTTPConnectionPool sends HTTP POST requests with request body and custom headers","urllib3.connectionpool.HTTPConnectionPool applies pool-level default headers to outgoing requests and closes cleanly as a context manager","urllib3.connectionpool.HTTPConnectionPool.is_same_host verifies whether a target URL matches the pool host, port, and scheme"],"goal":"verify urllib3.connectionpool.HTTPConnectionPool in pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.connectionpool.HTTPConnectionPool"]},"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.connectionpool.HTTPConnectionPool"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Demonstration of HTTPConnectionPool in urllib3."""

from __future__ import annotations

from typing import Mapping, Optional
import urllib3
from urllib3.connectionpool import HTTPConnectionPool
from urllib3.response import BaseHTTPResponse


def create_connection_pool(
    host: str,
    port: Optional[int] = None,
    maxsize: int = 5,
    headers: Optional[Mapping[str, str]] = None,
) -> HTTPConnectionPool:
    """Create an HTTPConnectionPool for a specific host and port."""
    return HTTPConnectionPool(host=host, port=port, maxsize=maxsize, headers=headers)


def send_get_request(
    pool: HTTPConnectionPool,
    path: str,
    headers: Optional[Mapping[str, str]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP GET request using the provided HTTPConnectionPool."""
    return pool.request("GET", path, headers=headers)


def send_post_request(
    pool: HTTPConnectionPool,
    path: str,
    body: bytes,
    headers: Optional[Mapping[str, str]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP POST request with a payload body using the HTTPConnectionPool."""
    return pool.request("POST", path, body=body, headers=headers)


def check_url_same_host(pool: HTTPConnectionPool, url: str) -> bool:
    """Check whether a given URL targets the same host, port, and scheme as the pool."""
    return pool.is_same_host(url)

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

"""Contract verification test for urllib3.connectionpool.HTTPConnectionPool."""

from __future__ import annotations

import http.server
import os
import sys
import threading

# Ensure root directory is in sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from urllib3.connectionpool import HTTPConnectionPool
from sample import (
    check_url_same_host,
    create_connection_pool,
    send_get_request,
    send_post_request,
)


class MockHTTPHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path == "/hello":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("X-Test-Header", "hello-test")
            self.end_headers()
            self.wfile.write(b"Hello HTTPConnectionPool!")
        elif self.path == "/headers":
            default_hdr = self.headers.get("X-Pool-Default", "")
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("X-Echo-Default", default_hdr)
            self.end_headers()
            self.wfile.write(b"headers-received")
        else:
            self.send_response(404)
            self.end_headers()

    def do_POST(self) -> None:
        if self.path == "/echo":
            content_len = int(self.headers.get("Content-Length", 0))
            payload = self.rfile.read(content_len)
            self.send_response(201)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            self.wfile.write(b"Echo: " + payload)
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format: str, *args: object) -> None:
        # Suppress server access logs during test run
        pass


def test_get_requests(port: int) -> None:
    """Assertion 1: urllib3.connectionpool.HTTPConnectionPool sends HTTP GET requests and decodes response status, body, and headers."""
    pool = HTTPConnectionPool("127.0.0.1", port=port)
    try:
        resp = pool.request("GET", "/hello")
        assert resp.status == 200, f"Expected status 200, got {resp.status}"
        assert resp.data == b"Hello HTTPConnectionPool!", f"Unexpected response body: {resp.data!r}"
        assert resp.headers.get("X-Test-Header") == "hello-test", (
            f"Unexpected header value: {resp.headers.get('X-Test-Header')}"
        )

        # Verify helper function
        sample_resp = send_get_request(pool, "/hello")
        assert sample_resp.status == 200
        assert sample_resp.data == b"Hello HTTPConnectionPool!"
    finally:
        pool.close()


def test_post_requests(port: int) -> None:
    """Assertion 2: urllib3.connectionpool.HTTPConnectionPool sends HTTP POST requests with request body and custom headers."""
    pool = create_connection_pool("127.0.0.1", port=port)
    try:
        payload = b"test payload"
        resp = pool.request("POST", "/echo", body=payload, headers={"Content-Type": "text/plain"})
        assert resp.status == 201, f"Expected status 201, got {resp.status}"
        assert resp.data == b"Echo: test payload", f"Unexpected response body: {resp.data!r}"

        # Verify helper function
        sample_resp = send_post_request(pool, "/echo", body=payload, headers={"Content-Type": "text/plain"})
        assert sample_resp.status == 201
        assert sample_resp.data == b"Echo: test payload"
    finally:
        pool.close()


def test_pool_headers_and_context_manager(port: int) -> None:
    """Assertion 3: urllib3.connectionpool.HTTPConnectionPool applies pool-level default headers to outgoing requests and closes cleanly as a context manager."""
    pool_ref = None
    with HTTPConnectionPool("127.0.0.1", port=port, headers={"X-Pool-Default": "pool-configured-header"}) as pool:
        pool_ref = pool
        assert pool.pool is not None, "Pool queue should be initialized"
        resp = pool.request("GET", "/headers")
        assert resp.status == 200, f"Expected status 200, got {resp.status}"
        assert resp.headers.get("X-Echo-Default") == "pool-configured-header", (
            f"Expected pool default header, got {resp.headers.get('X-Echo-Default')}"
        )

    # After exiting the context manager, the pool should be closed
    assert pool_ref is not None
    assert pool_ref.pool is None, "Pool queue should be None after close"


def test_is_same_host(port: int) -> None:
    """Assertion 4: urllib3.connectionpool.HTTPConnectionPool.is_same_host verifies whether a target URL matches the pool host, port, and scheme."""
    pool = HTTPConnectionPool("127.0.0.1", port=port)
    try:
        assert pool.is_same_host(f"http://127.0.0.1:{port}/hello") is True
        assert pool.is_same_host(f"http://127.0.0.1:{port + 1}/hello") is False
        assert pool.is_same_host(f"https://127.0.0.1:{port}/hello") is False
        assert pool.is_same_host(f"http://example.com:{port}/hello") is False

        # Verify helper function
        assert check_url_same_host(pool, f"http://127.0.0.1:{port}/hello") is True
        assert check_url_same_host(pool, "http://example.com/hello") is False
    finally:
        pool.close()


def main() -> None:
    server = http.server.HTTPServer(("127.0.0.1", 0), MockHTTPHandler)
    port = server.server_address[1]
    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()

    try:
        test_get_requests(port)
        test_post_requests(port)
        test_pool_headers_and_context_manager(port)
        test_is_same_host(port)
        print("All contract assertions passed successfully.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()