net_benchmark.http_bench.load_test

Load testing engine (0.5.1): throughput, sustained rate, ramp-up. Stats reuse HTTPAnalyzer/TargetStats from analysis.py — no separate percentile/latency implementation.

Classes

ConnectionReuseStats(total_requests, ...)	Keep-alive / connection-reuse detection (roadmap item 4).
IntervalStats(window_index, stats, ...)	One time bucket (default 1s) of results, for time-series charts (roadmap item 14).
LoadTestEngine(target[, http_engine])	Wraps a single-target HTTPBenchmarkEngine with load-shaping strategies.
LoadTestMode(value[, names, module, ...])
LoadTestSummary(mode, target, duration_s, ...)

class net_benchmark.http_bench.load_test.ConnectionReuseStats(total_requests, connections_opened)

Bases: object

Keep-alive / connection-reuse detection (roadmap item 4).

Kept separate from TargetStats.connection_reuse_rate (analysis.py) — that field is a rate over completed requests, computed from the connection_reused flag on each HTTPResult. This dataclass instead holds the raw TCP-connection-open count from the transport layer (HTTPBenchmarkEngine.get_connection_stats), which TargetStats has no equivalent for.

total_requests: int

connections_opened: int

property connections_reused: int

property reuse_rate: float

0-1 fraction, unlike TargetStats.connection_reuse_rate which is 0-100 — kept as a fraction here since this predates and is independent of the analysis.py field; exporters/CLI should be explicit about which one they’re reading.

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

Bases: str, Enum

THROUGHPUT = 'throughput'

SUSTAINED = 'sustained'

RAMP_UP = 'ramp_up'

class net_benchmark.http_bench.load_test.IntervalStats(window_index, stats, status_code_distribution)

Bases: object

One time bucket (default 1s) of results, for time-series charts (roadmap item 14). stats is a full TargetStats computed by running HTTPAnalyzer over just this bucket’s results — same percentile/latency math as everywhere else, not a separate calculation.

window_index: int

stats: TargetStats

status_code_distribution: List[Dict[str, Any]]

class net_benchmark.http_bench.load_test.LoadTestSummary(mode, target, duration_s, target_rps, stats, status_code_distribution, connection_reuse, intervals, results)

Bases: object

mode: LoadTestMode

target: str

duration_s: float

target_rps: Optional[float]

stats: TargetStats

status_code_distribution: List[Dict[str, Any]]

connection_reuse: ConnectionReuseStats

intervals: List[IntervalStats]

results: List[HTTPResult]

property achieved_rps: float

to_dict()

Return type:

Dict[str, Any]

class net_benchmark.http_bench.load_test.LoadTestEngine(target, http_engine=None)

Bases: object

Wraps a single-target HTTPBenchmarkEngine with load-shaping strategies.

One LoadTestEngine == one target. For multi-target load tests, construct one instance per target (each gets its own pooled client/transport via the underlying HTTPBenchmarkEngine, so origins never share connections).

async close()

Return type:

None

async run_throughput(duration_s=10.0, max_concurrency=200)

Saturate the target with up to max_concurrency concurrent requests for duration_s and report the achieved RPS.

Return type:

LoadTestSummary

async run_sustained(target_rps, duration_s, max_concurrency=None)

Fire requests at a fixed target_rps for duration_s using a simple token-bucket pacer.

Return type:

LoadTestSummary

async run_ramp_up(start_concurrency, max_concurrency, ramp_duration_s, hold_duration_s=0.0, step_interval_s=1.0, max_total_rps=None)

Step concurrency up linearly from start_concurrency to max_concurrency over ramp_duration_s, then hold at max_concurrency for hold_duration_s. Each concurrency “slot” keeps issuing back-to-back requests for as long as it’s alive, so RPS scales with concurrency naturally rather than being separately paced.

max_total_rps is a safety ceiling, not a target rate (use run_sustained for that). It exists because, unlike run_throughput (bounded by its semaphore) and run_sustained (bounded by its pacer), nothing here otherwise limits how fast slots fire against a very fast target (e.g. localhost, a CDN edge, or a mocked client in tests) — a single slot can spin as fast as the event loop allows. Default ceiling is generous (concurrency * 50 rps) so it only kicks in for genuinely pathological cases; pass None to disable.

Return type:

LoadTestSummary