net_benchmark.http_bench.core

Core HTTP benchmarking functionality.

Classes

`HTTPBenchmarkEngine`([max_concurrent, ...])
`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 and TLS metrics.
`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.

class net_benchmark.http_bench.core.TimingNetworkStream(stream, metrics)[source]

Bases: AsyncNetworkStream

Wraps an AsyncNetworkStream to capture TCP-connect and TLS-handshake times.

The metrics dict is shared with TimingNetworkBackend so writes made inside start_tls() are immediately visible there without any extra plumbing.

async read(max_bytes, timeout=None)[source]

Return type:: bytes

async write(buffer, timeout=None)[source]

Return type:: None

async aclose()[source]

Return type:: None

async start_tls(ssl_context, server_hostname=None, timeout=None)[source]

Return type:: AsyncNetworkStream

get_extra_info(info)[source]

Return type:: Any

class net_benchmark.http_bench.core.TimingNetworkBackend(local_address=None)[source]

Bases: AsyncNetworkBackend

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

One backend instance per origin so metrics are not clobbered across origins.

async connect_tcp(host, port, timeout=None, local_address=None, socket_options=None)[source]

Return type:: AsyncNetworkStream

async connect_unix_socket(path, timeout=None, socket_options=None)[source]

Return type:: AsyncNetworkStream

async sleep(seconds)[source]

Return type:: None

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)[source]

Bases: AsyncHTTPTransport

AsyncHTTPTransport that captures per-connection timing and TLS metrics.

Rather than overriding the non-existent _init_pool hook, we build the httpcore.AsyncConnectionPool ourselves and pass the custom TimingNetworkBackend directly.

get_connection_metrics()[source]

Return a snapshot of metrics from the most recently established connection.

Return type:: Dict[str, Any]

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

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.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>)[source]

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]

to_dict()[source]

Return type:: Dict[str, Any]

class net_benchmark.http_bench.core.TargetManager(targets)[source]

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()[source]

Return type:: List[str]

classmethod parse_targets_input(input_value)[source]

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)[source]

Bases: object

set_progress_callback(callback)[source]

Return type:: None

async close()[source]

Close all pooled clients. Must be awaited after run_benchmark.

Return type:: None

get_failed_targets()[source]

Return type:: Dict[str, int]

async request_single(target, iteration=1)[source]

Execute a single HTTP request with retry logic.

Return type:: HTTPResult

async run_benchmark(targets, iterations=1, warmup=False, warmup_fast=False)[source]

Run benchmark across all targets for N iterations.

Return type:: List[HTTPResult]