net_benchmark.http_bench.core

Core HTTP benchmarking functionality.

Classes

HTTPBenchmarkEngine([max_concurrent, ...])
HTTPDigestAuth(username, password)	httpx-native Digest authentication (MD5 or SHA-256, qop=auth).
HTTPProtocol(value[, names, module, ...])	Negotiated application protocol — captured from the live connection.
HTTPResult(target, method, start_time, ...)	Result of a single HTTP request
MetricsCapturingTransport([verify, cert, ...])	AsyncHTTPTransport that captures per-connection timing, TLS metrics, connection reuse info, and (best-effort) HTTP/2 pushes.
PushDetectingPool([ssl_context, proxy, ...])	AsyncConnectionPool that injects push-tracking H2 connections.
PushTrackingH2Connection([config])	An H2 connection that records promised stream headers.
TargetManager(targets)	Parse and validate HTTP target URLs.
TimingNetworkBackend([local_address])	Wraps AutoBackend to inject per-connection TCP and TLS timing.
TimingNetworkStream(stream, metrics[, ...])	Wraps an AsyncNetworkStream to capture TCP-connect and TLS-handshake times for exactly this connection.

class net_benchmark.http_bench.core.TimingNetworkStream(stream, metrics, session_registry=None)

Bases: AsyncNetworkStream

Wraps an AsyncNetworkStream to capture TCP-connect and TLS-handshake times for exactly this connection.

# Each TimingNetworkStream owns its own metrics dict (created fresh in TimingNetworkBackend.connect_tcp — never shared/mutated by other connections). Callers that need this specific connection’s metrics should retrieve this stream instance (e.g. via httpx.Response.extensions[“network_stream”]) and call get_metrics(), rather than reading TimingNetworkBackend.metrics, which only reflects whichever connection was opened most recently on this backend and is unsafe to read under concurrency (see MetricsCapturingTransport docs).

async read(max_bytes, timeout=None)

Return type:

bytes

async write(buffer, timeout=None)

Return type:

None

async aclose()

Return type:

None

async start_tls(ssl_context, server_hostname=None, timeout=None)

Return type:

AsyncNetworkStream

get_extra_info(info)

Return type:

Any

get_metrics()

This connection’s own metrics dict — safe to read even when other connections are concurrently in flight on the same backend.

Return type:

Dict[str, Any]

class net_benchmark.http_bench.core.TimingNetworkBackend(local_address=None)

Bases: AsyncNetworkBackend

Wraps AutoBackend to inject per-connection TCP and TLS timing.

One backend instance per origin. self.metrics reflects only the most recently opened connection and is kept for backward compatibility with MetricsCapturingTransport.get_connection_metrics() (used as a fallback when the per-request stream isn’t retrievable) — callers that need correct per-request attribution under concurrency should prefer the per-stream metrics via TimingNetworkStream.get_metrics().

async connect_tcp(host, port, timeout=None, local_address=None, socket_options=None)

Return type:

AsyncNetworkStream

async connect_unix_socket(path, timeout=None, socket_options=None)

Return type:

AsyncNetworkStream

async sleep(seconds)

Return type:

None

class net_benchmark.http_bench.core.PushTrackingH2Connection(config=None)

Bases: H2Connection

An H2 connection that records promised stream headers.

receive_data(data)

Pass some received HTTP/2 data to the connection for handling.

Parameters:

data (bytes) – The data received from the remote peer on the network.

Return type:

List[Event]

Returns:

A list of events that the remote peer triggered by sending this data.

class net_benchmark.http_bench.core.PushDetectingPool(ssl_context=None, proxy=None, max_connections=10, max_keepalive_connections=None, keepalive_expiry=None, http1=True, http2=False, retries=0, local_address=None, uds=None, network_backend=None, socket_options=None)

Bases: AsyncConnectionPool

AsyncConnectionPool that injects push-tracking H2 connections.

This relies on httpcore private internals (_init_connection, _h2_connection) that are not part of httpcore’s public API and are not guaranteed stable across versions. If either is missing or raises, push detection silently degrades to “no pushes recorded” rather than crashing the request — see _init_connection below.

class net_benchmark.http_bench.core.MetricsCapturingTransport(verify=True, cert=None, http2=False, sni_hostname=None, mtls_cert=None, mtls_key=None, local_address=None, enable_push_detection=False)

Bases: AsyncHTTPTransport

AsyncHTTPTransport that captures per-connection timing, TLS metrics, connection reuse info, and (best-effort) HTTP/2 pushes.

# get_connection_metrics() returns only the most-recently-opened connection’s metrics for this transport and is unsafe to rely on when multiple requests to the same origin are in flight concurrently — it exists as a fallback for callers that can’t retrieve the per-request stream (e.g. sequential/single-concurrency use, or older code). Prefer reading httpx.Response.extensions[“network_stream”].get_metrics() when available; request_single() below does this automatically.

get_connection_metrics(connection_id=None)

Pass connection_id to get that specific connection’s metrics safely, even under concurrency (looked up from TimingNetworkBackend.metrics_by_id, unaffected by other connections opening/closing). Without connection_id, falls back to the old “most recently opened” snapshot — still unsafe under concurrency, kept only for callers with no id to give (e.g. a totally fresh transport with zero requests sent).

Return type:

Dict[str, Any]

get_connections_opened()

Cumulative count of new TCP connections opened by this transport’s pool.

Return type:

int

get_recent_push_promises()

Return any HTTP/2 push promises recorded by the push-detecting pool. Empty list if push detection is unavailable/disabled, or if the httpcore internals it depends on aren’t present in this version — never raises for that reason. Only AttributeError is swallowed here (private-API shape drift); anything else propagates so real bugs in this loop aren’t hidden alongside “h2 unavailable”.

Return type:

List[Dict[str, Any]]

class net_benchmark.http_bench.core.HTTPProtocol(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Negotiated application protocol — captured from the live connection.

HTTP1 = 'HTTP/1.1'

HTTP2 = 'HTTP/2'

UNKNOWN = 'unknown'

class net_benchmark.http_bench.core.HTTPDigestAuth(username, password)

Bases: Auth

httpx-native Digest authentication (MD5 or SHA-256, qop=auth).

Replaces a previous version that borrowed requests.auth.HTTPDigestAuth outside its intended Session.send() retry flow (it never actually computed a digest response from the real WWW-Authenticate challenge). This parses the real 401 and computes HA1/HA2/response directly, no requests dependency.

auth_flow(request)

Return type:

Generator[Request, Response, None]

class net_benchmark.http_bench.core.HTTPResult(target, method, start_time, end_time, total_ms, status, iteration=1, attempt_number=1, query_id=<factory>, http_status_code=None, error_message=None, redirect_count=0, final_url='', ttfb_ms=None, ttlb_ms=None, protocol=HTTPProtocol.UNKNOWN, alpn_negotiated=None, response_size_bytes=None, compressed=False, content_encoding=None, content_type=None, security_headers=<factory>, cdn_fingerprint=None, server_header=None, cert_expiry_days=None, cert_cn=None, alt_svc=None, ip_version=None, cert_issuer_cn=None, cert_sans=<factory>, cert_wildcard=False, downgrade_detected=False, redirect_urls=<factory>, tls_handshake_ms=None, tcp_connect_ms=None, dns_resolve_ms=None, dns_resolver_ip=None, compressed_size_bytes=None, http2_expected=False, http2_downgraded=False, redirect_timings=<factory>, query_params=<factory>, cache_control=None, etag=None, last_modified=None, age=None, request_id=None, assertion_results=<factory>, connection_id=None, connection_reused=False, tcp_fast_open=None, tls_resumed=False, tls_session_id=None, session_ticket=False, http2_push_count=0, http2_pushes=<factory>, upload_size_bytes=None, upload_time_ms=None, upload_throughput_mbps=None, websocket_handshake_ms=None)

Bases: object

Result of a single HTTP request

target: str

method: str

start_time: float

end_time: float

total_ms: float

status: QueryStatus

iteration: int = 1

attempt_number: int = 1

query_id: str

http_status_code: Optional[int] = None

error_message: Optional[str] = None

redirect_count: int = 0

final_url: str = ''

ttfb_ms: Optional[float] = None

ttlb_ms: Optional[float] = None

protocol: HTTPProtocol = 'unknown'

alpn_negotiated: Optional[str] = None

response_size_bytes: Optional[int] = None

compressed: bool = False

content_encoding: Optional[str] = None

content_type: Optional[str] = None

security_headers: Dict[str, Optional[str]]

cdn_fingerprint: Optional[str] = None

server_header: Optional[str] = None

cert_expiry_days: Optional[int] = None

cert_cn: Optional[str] = None

alt_svc: Optional[str] = None

ip_version: Optional[str] = None

cert_issuer_cn: Optional[str] = None

cert_sans: List[str]

cert_wildcard: bool = False

downgrade_detected: bool = False

redirect_urls: List[str]

tls_handshake_ms: Optional[float] = None

tcp_connect_ms: Optional[float] = None

dns_resolve_ms: Optional[float] = None

dns_resolver_ip: Optional[str] = None

compressed_size_bytes: Optional[int] = None

http2_expected: bool = False

http2_downgraded: bool = False

redirect_timings: List[Dict[str, Any]]

query_params: Dict[str, str]

cache_control: Optional[str] = None

etag: Optional[str] = None

last_modified: Optional[str] = None

age: Optional[str] = None

request_id: Optional[str] = None

assertion_results: Dict[str, bool]

connection_id: Optional[str] = None

connection_reused: bool = False

tcp_fast_open: Optional[bool] = None

tls_resumed: bool = False

tls_session_id: Optional[str] = None

session_ticket: bool = False

http2_push_count: int = 0

http2_pushes: List[str]

upload_size_bytes: Optional[int] = None

upload_time_ms: Optional[float] = None

upload_throughput_mbps: Optional[float] = None

websocket_handshake_ms: Optional[float] = None

to_dict()

Return type:

Dict[str, Any]

class net_benchmark.http_bench.core.TargetManager(targets)

Bases: object

Parse and validate HTTP target URLs.

DEFAULT_TARGETS: List[str] = ['https://www.cloudflare.com', 'https://www.google.com', 'https://www.github.com', 'https://www.wikipedia.org', 'https://www.apple.com']

property targets: List[str]

classmethod get_default_targets()

Return type:

List[str]

classmethod parse_targets_input(input_value)

Return type:

TargetManager

class net_benchmark.http_bench.core.HTTPBenchmarkEngine(max_concurrent=50, timeout=10.0, connect_timeout=None, read_timeout=None, write_timeout=None, max_retries=2, retry_backoff_multiplier=0.1, retry_backoff_base=2.0, method='GET', headers=None, follow_redirects=True, verify_ssl=True, http2=True, auth=None, cookies=None, proxy=None, sni_hostname=None, mtls_cert=None, mtls_key=None, inject_request_id=False, assertions=None, query_params=None, body=None, local_address=None, use_cookie_jar=False, enable_push_detection=False, enable_connection_reuse=False, enable_tfo_detection=False, enable_tls_resumption=False, enable_session_ticket=False)

Bases: object

set_progress_callback(callback)

Return type:

None

async close()

Close all pooled clients. Must be awaited after run_benchmark.

Return type:

None

get_failed_targets()

Return type:

Dict[str, int]

get_connection_stats(target, connection_id=None)

Pass connection_id (from HTTPResult.connection_id, requires enable_connection_reuse=True) to safely inspect one specific connection’s metrics. Without it, “latest_metrics” is a best-effort snapshot only reliable when no concurrent requests are in flight.

Return type:

Dict[str, Any]

async request_single(target, iteration=1, multipart_file_size=0)

Execute a single HTTP request with retry logic.

Return type:

HTTPResult

async websocket_single(target, iteration=1)

Perform a WebSocket handshake and measure the time to establish.

Requires aiohttp to be installed (pip install net-benchmark[websocket]).

Return type:

HTTPResult

async run_benchmark(targets, iterations=1, warmup=False, warmup_fast=False)

Run benchmark across all targets for N iterations.

Return type:

List[HTTPResult]