urllib3 2.7.0: connectionpool.ConnectionPool

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

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:6b0ac817343cf35ac77743293bf2b9ec92f7e4fdee5356f1f5669262e4328665","contract":["ConnectionPool initializes with normalized host and optional port attributes and formats descriptive string representation","ConnectionPool raises LocationValueError when initialized without a valid host string","ConnectionPool supports context manager protocol and calls close upon exit","ConnectionPool serves as the base class for HTTPConnectionPool and HTTPSConnectionPool connection pool implementations"],"goal":"verify urllib3.connectionpool.ConnectionPool in pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.connectionpool.ConnectionPool"]},"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.ConnectionPool"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Sample demonstrating the usage and extension of urllib3.connectionpool.ConnectionPool."""

from typing import Any, Dict, Optional
from urllib3.connectionpool import ConnectionPool, HTTPConnectionPool


class ManagedServicePool(ConnectionPool):
    """Custom service pool extending ConnectionPool with lifecycle tracking."""

    def __init__(self, host: str, port: Optional[int] = None) -> None:
        super().__init__(host=host, port=port)
        self.is_closed: bool = False

    def close(self) -> None:
        """Close pool resources and update state."""
        super().close()
        self.is_closed = True


def create_connection_pool(host: str, port: Optional[int] = None) -> ConnectionPool:
    """Instantiate a base ConnectionPool instance with host normalization."""
    return ConnectionPool(host=host, port=port)


def is_connection_pool_instance(pool: object) -> bool:
    """Check if an object is an instance of ConnectionPool."""
    return isinstance(pool, ConnectionPool)


def inspect_pool_target(pool: ConnectionPool) -> Dict[str, Any]:
    """Retrieve normalized target host, port, and string representation from a ConnectionPool."""
    return {
        "host": pool.host,
        "port": pool.port,
        "description": str(pool),
    }

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

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

import os
import sys

# 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__))))

from urllib3.connectionpool import ConnectionPool, HTTPConnectionPool, HTTPSConnectionPool
from urllib3.exceptions import LocationValueError
from sample import (
    ManagedServicePool,
    create_connection_pool,
    inspect_pool_target,
    is_connection_pool_instance,
)


def test_connection_pool_initialization_and_normalization() -> None:
    """Assertion 1: ConnectionPool initializes with normalized host and optional port attributes and formats descriptive string representation."""
    pool = create_connection_pool("EXAMPLE.COM", 8080)
    assert pool.host == "example.com", f"Expected 'example.com', got {pool.host}"
    assert pool.port == 8080, f"Expected 8080, got {pool.port}"
    assert str(pool) == "ConnectionPool(host='example.com', port=8080)"

    info = inspect_pool_target(pool)
    assert info["host"] == "example.com"
    assert info["port"] == 8080
    assert info["description"] == "ConnectionPool(host='example.com', port=8080)"


def test_connection_pool_invalid_host_raises() -> None:
    """Assertion 2: ConnectionPool raises LocationValueError when initialized without a valid host string."""
    raised_empty = False
    try:
        ConnectionPool(host="")
    except LocationValueError:
        raised_empty = True
    assert raised_empty, "Expected LocationValueError when host is empty string"

    raised_none = False
    try:
        ConnectionPool(host=None)  # type: ignore[arg-type]
    except LocationValueError:
        raised_none = True
    assert raised_none, "Expected LocationValueError when host is None"


def test_connection_pool_context_manager() -> None:
    """Assertion 3: ConnectionPool supports context manager protocol and calls close upon exit."""
    custom_pool = ManagedServicePool("localhost", 9000)
    assert not custom_pool.is_closed

    with custom_pool as entered_pool:
        assert entered_pool is custom_pool
        assert not entered_pool.is_closed

    assert custom_pool.is_closed, "Expected pool.close() to be called on context manager exit"

    # Also test base ConnectionPool directly in context manager
    base_pool = ConnectionPool("localhost", 80)
    with base_pool as p:
        assert p is base_pool
    # Base close() can be called cleanly
    base_pool.close()


def test_connection_pool_subclass_hierarchy() -> None:
    """Assertion 4: ConnectionPool serves as the base class for HTTPConnectionPool and HTTPSConnectionPool connection pool implementations."""
    assert issubclass(HTTPConnectionPool, ConnectionPool), "HTTPConnectionPool must subclass ConnectionPool"
    assert issubclass(HTTPSConnectionPool, ConnectionPool), "HTTPSConnectionPool must subclass ConnectionPool"

    http_pool = HTTPConnectionPool("localhost", port=8080)
    assert is_connection_pool_instance(http_pool)
    assert http_pool.host == "localhost"
    assert http_pool.port == 8080
    assert "HTTPConnectionPool(host='localhost', port=8080)" in str(http_pool)
    http_pool.close()

    https_pool = HTTPSConnectionPool("localhost", port=8443)
    assert is_connection_pool_instance(https_pool)
    assert https_pool.host == "localhost"
    assert https_pool.port == 8443
    assert "HTTPSConnectionPool(host='localhost', port=8443)" in str(https_pool)
    https_pool.close()


def main() -> None:
    test_connection_pool_initialization_and_normalization()
    test_connection_pool_invalid_host_raises()
    test_connection_pool_context_manager()
    test_connection_pool_subclass_hierarchy()
    print("All contract assertions passed successfully.")


if __name__ == "__main__":
    main()