httpcore2 2.10.0: ConnectionPool, request

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/httpcore2@2.10.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/httpcore2@2.10.0
Demonstrate these symbols/APIs:
  - httpcore2.ConnectionPool
  - httpcore2.request

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:e1186ded2041ab7a87d8b1f940f0e6485628b6695fdcf52da031b63a59872a91","contract":["httpcore2.URL parses scheme, host, port and target path","httpcore2.ConnectionPool sends HTTP requests via network backend and returns Response with status and body","httpcore2.ConnectionPool.request sends request and automatically closes response after reading content","httpcore2.request sends synchronous HTTP GET and POST requests and returns Response","httpcore2.request raises ConnectError or NetworkError when connecting to an unreachable port"],"goal":"verify pkg:pypi/httpcore2@2.10.0","kind":"HOW","packages":["pkg:pypi/httpcore2@2.10.0"],"schemaVersion":1,"symbols":["httpcore2.ConnectionPool","httpcore2.request"]},"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/httpcore2@2.10.0"],"schemaVersion":1,"subject":"pkg:pypi/httpcore2@2.10.0","symbols":["httpcore2.ConnectionPool","httpcore2.request"],"verifierAdapter":"python@1"}

[project]
name = "verify-httpcore2"
version = "0.1.0"
dependencies = [
    "h11==0.16.0",
    "httpcore2==2.10.0",
    "truststore==0.10.4",
]

h11==0.16.0
httpcore2==2.10.0
truststore==0.10.4

{
  "schemaVersion": 1,
  "goal": "verify pkg:pypi/httpcore2@2.10.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/httpcore2@2.10.0"
  ],
  "symbols": [
    "httpcore2.ConnectionPool",
    "httpcore2.request"
  ]
}

import httpcore2
import http.server
import threading


def test_url_parsing():
    url = httpcore2.URL("http://example.com:8080/path?query=1")
    assert url.scheme == b"http"
    assert url.host == b"example.com"
    assert url.port == 8080
    assert url.target == b"/path?query=1"


def test_connection_pool_mock_backend():
    body = b"Hello Pool Response"
    headers = [
        (b"Content-Type", b"text/plain"),
        (b"Content-Length", str(len(body)).encode("ascii")),
    ]
    header_block = b"".join(k + b": " + v + b"\r\n" for k, v in headers)
    raw_response = b"HTTP/1.1 200 OK\r\n" + header_block + b"\r\n" + body

    backend = httpcore2.MockBackend(buffer=[raw_response])
    pool = httpcore2.ConnectionPool(network_backend=backend)
    request = httpcore2.Request(
        "GET",
        "http://example.com/mock-endpoint",
        headers=[(b"host", b"example.com")],
    )

    response = pool.handle_request(request)
    assert response.status == 200
    headers_dict = dict(response.headers)
    assert headers_dict[b"Content-Type"] == b"text/plain"
    assert response.read() == body
    pool.close()


def test_connection_pool_request_helper():
    json_body = b'{"status": "success", "code": 201}'
    headers = [
        (b"Content-Type", b"application/json"),
        (b"Content-Length", str(len(json_body)).encode("ascii")),
    ]
    header_block = b"".join(k + b": " + v + b"\r\n" for k, v in headers)
    raw_response = b"HTTP/1.1 201 Created\r\n" + header_block + b"\r\n" + json_body

    backend = httpcore2.MockBackend(buffer=[raw_response])
    with httpcore2.ConnectionPool(network_backend=backend) as pool:
        response = pool.request("POST", "http://example.com/api/items", content=b"item-data")
        assert response.status == 201
        assert response.content == json_body


def test_httpcore_request_with_local_server():
    class TestHandler(http.server.BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            content = b"local-get-ok"
            self.send_header("Content-Length", str(len(content)))
            self.end_headers()
            self.wfile.write(content)

        def do_POST(self):
            length = int(self.headers.get("Content-Length", 0))
            payload = self.rfile.read(length) if length > 0 else b""
            self.send_response(200)
            self.send_header("Content-Type", "application/octet-stream")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

        def log_message(self, format, *args):
            pass

    server = http.server.HTTPServer(("127.0.0.1", 0), TestHandler)
    port = server.server_port
    thread = threading.Thread(target=server.serve_forever)
    thread.daemon = True
    thread.start()

    try:
        # GET request
        resp_get = httpcore2.request("GET", f"http://127.0.0.1:{port}/ping")
        assert resp_get.status == 200
        assert resp_get.content == b"local-get-ok"

        # POST request
        resp_post = httpcore2.request("POST", f"http://127.0.0.1:{port}/echo", content=b"echo-payload-bytes")
        assert resp_post.status == 200
        assert resp_post.content == b"echo-payload-bytes"
    finally:
        server.shutdown()


def test_error_handling_unreachable_port():
    try:
        httpcore2.request("GET", "http://127.0.0.1:1")
        assert False, "Expected ConnectError or NetworkError on unreachable port"
    except (httpcore2.ConnectError, httpcore2.NetworkError) as exc:
        assert isinstance(exc, (httpcore2.ConnectError, httpcore2.NetworkError))


def main():
    test_url_parsing()
    test_connection_pool_mock_backend()
    test_connection_pool_request_helper()
    test_httpcore_request_with_local_server()
    test_error_handling_unreachable_port()
    print("All contract assertions passed successfully.")


if __name__ == "__main__":
    main()