File size: 3,959 Bytes
53b2ac9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | # SPDX-License-Identifier: MIT
"""Tests for the http_bench plugin."""
import http.server
import os
import sys
import threading
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "plugins"))
from http_bench import run, _percentile
# ── Fixtures ──
@pytest.fixture(scope="module")
def local_http_server():
"""Start a simple HTTP server on a random port."""
handler = http.server.SimpleHTTPRequestHandler
# Suppress log output
class QuietHandler(handler):
def log_message(self, format, *args):
pass
srv = http.server.HTTPServer(("127.0.0.1", 0), QuietHandler)
port = srv.server_address[1]
t = threading.Thread(target=srv.serve_forever, daemon=True)
t.start()
yield port
srv.shutdown()
# ── _percentile helper ──
def test_percentile_basic():
data = list(range(1, 101)) # 1..100
assert _percentile(data, 50) == pytest.approx(50.5, abs=1)
assert _percentile(data, 95) == pytest.approx(95.05, abs=1)
assert _percentile(data, 99) == pytest.approx(99.01, abs=1)
def test_percentile_single():
assert _percentile([42.0], 50) == 42.0
assert _percentile([42.0], 99) == 42.0
def test_percentile_empty():
assert _percentile([], 50) == 0.0
# ── Basic functionality ──
def test_missing_url():
result = run()
assert "Error" in result
def test_bench_local_server(local_http_server):
result = run(url=f"http://127.0.0.1:{local_http_server}", requests=5, concurrency=2, timeout=5)
assert "Latency" in result
assert "p50" in result
assert "p95" in result
assert "p99" in result
assert "Throughput" in result
assert "Success" in result
def test_bench_reports_failures():
# Port 1 is almost certainly not an HTTP server
result = run(url="http://127.0.0.1:1", requests=2, concurrency=1, timeout=0.5)
assert "Failed" in result
def test_small_request_count(local_http_server):
result = run(url=f"http://127.0.0.1:{local_http_server}", requests=1)
assert "Total:" in result
assert "1" in result
def test_auto_prepend_http(local_http_server):
"""Should auto-prepend http:// if missing."""
result = run(url=f"127.0.0.1:{local_http_server}", requests=2)
assert "Latency" in result or "Failed" in result
def test_clamp_concurrency(local_http_server):
"""Concurrency over 20 should be clamped."""
result = run(url=f"http://127.0.0.1:{local_http_server}", requests=3, concurrency=100)
# Should still work, just clamped to 20
assert "Total:" in result
def test_clamp_requests(local_http_server):
"""Requests over 500 should be clamped."""
result = run(url=f"http://127.0.0.1:{local_http_server}", requests=9999, concurrency=1)
# Should report 500 max
assert "500" in result
# ── TLS verification ──
def _captured_contexts(monkeypatch, **run_kwargs):
import ssl
import urllib.error
import http_bench
contexts = []
def fake_urlopen(req, timeout=None, context=None):
contexts.append(context)
raise urllib.error.URLError("stubbed")
monkeypatch.setattr(http_bench.urllib.request, "urlopen", fake_urlopen)
output = run(url="https://example.invalid", requests=2, concurrency=1, **run_kwargs)
assert len(contexts) == 2
return ssl, contexts, output
def test_tls_verified_by_default(monkeypatch):
ssl, contexts, output = _captured_contexts(monkeypatch)
for ctx in contexts:
assert ctx.verify_mode == ssl.CERT_REQUIRED
assert ctx.check_hostname is True
assert "verification disabled" not in output
@pytest.mark.parametrize("flag", [True, "true"])
def test_tls_insecure_opt_in(monkeypatch, flag):
ssl, contexts, output = _captured_contexts(monkeypatch, insecure=flag)
for ctx in contexts:
assert ctx.verify_mode == ssl.CERT_NONE
assert ctx.check_hostname is False
assert "verification disabled" in output
|