Jan Novak
404b372070
Third Python implementation in the latency-tool progression. Mirrors latprobe's feature set but replaces raw sockets with httpx, gaining HTTP/2 support and redirect following. Package layout (hxprobe/hxprobe/): - probe.py: asyncio + httpx.AsyncClient with a _TimingStream transport wrapper that captures DNS/connect/TLS/TTFB/transfer phase timings via httpx event hooks (get_connection_stats, request_started, etc.). VerboseDetail captures resolved IP, TLS version/cipher/bits, cert CN/ expiry/issuer (from httpx's SSLObject), and response headers. Options: timeout, verbose, follow_redirects, http2 - aggregate.py: summarize() → per-phase min/avg/max (same schema as latprobe) - cli.py: run(args,stdout,stderr)->int; argparse with redirect_stdout/ redirect_stderr + SystemExit catch; -n/--count, -c/--concurrency, --timeout, --fail, --json, -v/--verbose, --no-follow-redirects, --no-http2, -f/--file (URL list from file, mutually exclusive with args); multi-URL summary footer (tally + exit label) when len(urls) > 1; worst-exit-code logic mirrors Go/latprobe; JSON output is bare array - duration.py: same parse_duration() as latprobe - Exit codes: 0 ok, 1 usage, 2 dns, 3 connect, 4 timeout, 5 tls, 6 http≥400 Tests (hxprobe/tests/): - test_probe.py: 18 hermetic tests using anyio + in-process ASGI servers - test_cli.py: 36 hermetic tests (success, failures, JSON, verbose, -f flag, run-summary footer, worst-code accumulation) - test_integration.py: pytest-marked @integration (excluded from hx-test) Toolchain: uv + ruff + pytest; pyproject.toml with [dependency-groups]; hxprobe/Makefile standalone (help, deps, run, lint, fmt, test, test-integration, check, clean); .python-version pins 3.14 Configs: 8 fixture files mirroring python/configs/ (all-ok through mixed) USAGE.md: 16 runnable examples with real captured output Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
257 lines
8.2 KiB
Python
257 lines
8.2 KiB
Python
"""Hermetic tests for hxprobe.probe.measure(), including a redirect test
|
|
covering the Go-http.DefaultClient-parity feature (redirects followed by
|
|
default) that a plain raw-socket HTTP/1.1 client would not have."""
|
|
|
|
import http.server
|
|
import socket
|
|
import threading
|
|
import unittest
|
|
|
|
from hxprobe.probe import Options, VerboseDetail, measure
|
|
|
|
|
|
class _OKHandler(http.server.BaseHTTPRequestHandler):
|
|
# HTTP/1.1 (BaseHTTPRequestHandler defaults to 1.0) so hxprobe negotiates
|
|
# http_version="HTTP/1.1" instead of "HTTP/1.0". Keep-alive is then in
|
|
# play, so every response below sends an explicit Content-Length —
|
|
# without it, h11 has no way to detect body-end short of connection
|
|
# close, and the client hangs until it hits the request timeout.
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def _send_body(self, status: int, body: bytes) -> None:
|
|
self.send_response(status)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
self._send_body(200, b"hello hxprobe")
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
class _NotFoundHandler(http.server.BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def do_GET(self):
|
|
body = b"not found"
|
|
self.send_response(404)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
class _RedirectHandler(http.server.BaseHTTPRequestHandler):
|
|
"""/redirect -> 302 to /landed; /landed -> 200."""
|
|
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def do_GET(self):
|
|
if self.path == "/redirect":
|
|
self.send_response(302)
|
|
self.send_header("Location", "/landed")
|
|
self.send_header("Content-Length", "0")
|
|
self.end_headers()
|
|
else:
|
|
body = b"landed"
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
def _start_server(handler_class):
|
|
server = http.server.HTTPServer(("127.0.0.1", 0), handler_class)
|
|
t = threading.Thread(target=server.serve_forever)
|
|
t.daemon = True
|
|
t.start()
|
|
return server, server.server_address[1]
|
|
|
|
|
|
def _free_port() -> int:
|
|
with socket.socket() as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
def _black_hole_port() -> int:
|
|
"""Bind a port that accepts TCP but never sends any data (triggers a read timeout)."""
|
|
srv = socket.socket()
|
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
srv.bind(("127.0.0.1", 0))
|
|
srv.listen(10)
|
|
port = srv.getsockname()[1]
|
|
conns: list = []
|
|
|
|
def _serve():
|
|
while True:
|
|
try:
|
|
conn, _ = srv.accept()
|
|
conns.append(conn)
|
|
except OSError:
|
|
break
|
|
|
|
threading.Thread(target=_serve, daemon=True).start()
|
|
return port
|
|
|
|
|
|
class TestMeasureSuccess(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.ok_server, cls.ok_port = _start_server(_OKHandler)
|
|
cls.nf_server, cls.nf_port = _start_server(_NotFoundHandler)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.ok_server.shutdown()
|
|
cls.nf_server.shutdown()
|
|
|
|
def test_all_phases_present_on_success(self):
|
|
r = measure(f"http://127.0.0.1:{self.ok_port}")
|
|
self.assertIsNone(r.err)
|
|
self.assertEqual(r.fail_phase, "")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertTrue(r.dns.present)
|
|
self.assertTrue(r.connect.present)
|
|
self.assertFalse(r.tls.present) # http — no TLS
|
|
self.assertTrue(r.ttfb.present)
|
|
self.assertTrue(r.transfer.present)
|
|
self.assertTrue(r.total.present)
|
|
|
|
def test_status_code_404(self):
|
|
r = measure(f"http://127.0.0.1:{self.nf_port}")
|
|
self.assertIsNone(r.err)
|
|
self.assertEqual(r.status_code, 404)
|
|
self.assertTrue(r.total.present)
|
|
|
|
def test_options_default_follows_redirects_and_http2(self):
|
|
opts = Options()
|
|
self.assertTrue(opts.follow_redirects)
|
|
self.assertTrue(opts.http2)
|
|
|
|
|
|
class TestMeasureFailures(unittest.TestCase):
|
|
def test_dns_failure(self):
|
|
r = measure("http://no.such.host.invalid")
|
|
self.assertEqual(r.fail_phase, "dns")
|
|
self.assertIsNotNone(r.err)
|
|
self.assertFalse(r.dns.present)
|
|
self.assertFalse(r.connect.present)
|
|
self.assertTrue(r.total.present)
|
|
|
|
def test_connection_refused(self):
|
|
port = _free_port()
|
|
r = measure(f"http://127.0.0.1:{port}")
|
|
self.assertEqual(r.fail_phase, "connect")
|
|
self.assertIsNotNone(r.err)
|
|
self.assertTrue(r.dns.present)
|
|
self.assertTrue(r.connect.present)
|
|
self.assertFalse(r.ttfb.present)
|
|
self.assertTrue(r.total.present)
|
|
|
|
def test_ttfb_timeout(self):
|
|
port = _black_hole_port()
|
|
r = measure(f"http://127.0.0.1:{port}", Options(timeout=0.2))
|
|
self.assertEqual(r.fail_phase, "timeout")
|
|
self.assertIsNotNone(r.err)
|
|
self.assertTrue(r.dns.present)
|
|
self.assertTrue(r.connect.present)
|
|
self.assertTrue(r.total.present)
|
|
|
|
def test_bad_scheme(self):
|
|
r = measure("ftp://example.com")
|
|
self.assertEqual(r.fail_phase, "request")
|
|
self.assertIsNotNone(r.err)
|
|
self.assertFalse(r.total.present)
|
|
|
|
|
|
class TestRedirects(unittest.TestCase):
|
|
"""The Go-http.DefaultClient-parity feature: redirects followed by
|
|
default, with dns/connect/tls timed from the first hop only."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.server, cls.port = _start_server(_RedirectHandler)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def _url(self, path: str) -> str:
|
|
return f"http://127.0.0.1:{self.port}{path}"
|
|
|
|
def test_follows_redirect_by_default(self):
|
|
r = measure(self._url("/redirect"), Options(verbose=True))
|
|
self.assertIsNone(r.err)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.detail.redirect_count, 1)
|
|
|
|
def test_no_follow_redirects_reports_302(self):
|
|
r = measure(self._url("/redirect"), Options(follow_redirects=False, verbose=True))
|
|
self.assertIsNone(r.err)
|
|
self.assertEqual(r.status_code, 302)
|
|
self.assertEqual(r.detail.redirect_count, 0)
|
|
|
|
def test_dns_and_connect_present_across_redirect(self):
|
|
r = measure(self._url("/redirect"))
|
|
self.assertTrue(r.dns.present)
|
|
self.assertTrue(r.connect.present)
|
|
self.assertTrue(r.ttfb.present)
|
|
self.assertTrue(r.total.present)
|
|
|
|
|
|
class TestVerboseDetail(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.ok_server, cls.ok_port = _start_server(_OKHandler)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.ok_server.shutdown()
|
|
|
|
def _url(self):
|
|
return f"http://127.0.0.1:{self.ok_port}"
|
|
|
|
def test_detail_none_without_verbose(self):
|
|
r = measure(self._url())
|
|
self.assertIsNone(r.detail)
|
|
|
|
def test_detail_present_with_verbose(self):
|
|
r = measure(self._url(), Options(verbose=True))
|
|
self.assertIsInstance(r.detail, VerboseDetail)
|
|
|
|
def test_resolved_ip_set_on_success(self):
|
|
r = measure(self._url(), Options(verbose=True))
|
|
self.assertEqual(r.detail.resolved_ip, "127.0.0.1")
|
|
|
|
def test_http_version_set_on_success(self):
|
|
r = measure(self._url(), Options(verbose=True))
|
|
self.assertIn(r.detail.http_version, ("HTTP/1.1", "HTTP/2"))
|
|
|
|
def test_no_tls_fields_for_http(self):
|
|
r = measure(self._url(), Options(verbose=True))
|
|
self.assertEqual(r.detail.tls_version, "")
|
|
self.assertIsNone(r.detail.cert)
|
|
|
|
def test_headers_populated_on_success(self):
|
|
r = measure(self._url(), Options(verbose=True))
|
|
self.assertIsInstance(r.detail.headers, dict)
|
|
self.assertTrue(len(r.detail.headers) > 0)
|
|
|
|
def test_ip_set_on_connect_fail(self):
|
|
port = _free_port()
|
|
r = measure(f"http://127.0.0.1:{port}", Options(verbose=True))
|
|
self.assertEqual(r.fail_phase, "connect")
|
|
self.assertEqual(r.detail.resolved_ip, "127.0.0.1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|