| |
| """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 |
|
|
|
|
| |
|
|
| @pytest.fixture(scope="module") |
| def local_http_server(): |
| """Start a simple HTTP server on a random port.""" |
| handler = http.server.SimpleHTTPRequestHandler |
|
|
| |
| 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() |
|
|
|
|
| |
|
|
| def test_percentile_basic(): |
| data = list(range(1, 101)) |
| 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 |
|
|
|
|
| |
|
|
| 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(): |
| |
| 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) |
| |
| 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) |
| |
| assert "500" in result |
|
|
|
|
| |
|
|
| 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 |
|
|
