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>
261 lines
8.2 KiB
Python
261 lines
8.2 KiB
Python
"""Hermetic tests for hxprobe.cli.run() — a standalone CLI (its own argparse,
|
|
exit codes, and text/JSON rendering, not shared with any other package).
|
|
These tests spot-check the core rendering paths (single URL, --fail, JSON,
|
|
DNS/connect failures) plus what's specific to hxprobe: the --no-http2/
|
|
--no-follow-redirects flags and that redirects/http_version are really
|
|
wired through end to end."""
|
|
|
|
import http.server
|
|
import io
|
|
import json
|
|
import os
|
|
import socket
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
from hxprobe.cli import EXIT_CONNECT, EXIT_DNS, EXIT_HTTP, EXIT_OK, EXIT_USAGE, run
|
|
|
|
|
|
def _free_port() -> int:
|
|
with socket.socket() as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
class _OKHandler(http.server.BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1" # keep-alive is in play — always send Content-Length
|
|
|
|
def do_GET(self):
|
|
body = b"hello hxprobe"
|
|
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
|
|
|
|
|
|
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 _invoke(args: list[str]) -> tuple[int, str, str]:
|
|
out, err = io.StringIO(), io.StringIO()
|
|
code = run(args, out, err)
|
|
return code, out.getvalue(), err.getvalue()
|
|
|
|
|
|
class TestCLIBasics(unittest.TestCase):
|
|
def test_help_shows_hxprobe_prog_name(self):
|
|
code, out, err = _invoke(["-h"])
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertIn("hxprobe", out)
|
|
|
|
def test_help_lists_protocol_flags(self):
|
|
code, out, err = _invoke(["-h"])
|
|
self.assertIn("--no-http2", out)
|
|
self.assertIn("--no-follow-redirects", out)
|
|
|
|
|
|
class TestCLISuccess(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.ok_server, cls.ok_port = _start_server(_OKHandler)
|
|
cls.nf_server, cls.nf_port = _start_server(_NotFoundHandler)
|
|
cls.redirect_server, cls.redirect_port = _start_server(_RedirectHandler)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.ok_server.shutdown()
|
|
cls.nf_server.shutdown()
|
|
cls.redirect_server.shutdown()
|
|
|
|
def _url(self, port=None, path=""):
|
|
return f"http://127.0.0.1:{port or self.ok_port}{path}"
|
|
|
|
def test_single_url_success(self):
|
|
code, out, _ = _invoke([self._url()])
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertIn("200", out)
|
|
self.assertIn("DNS lookup", out)
|
|
|
|
def test_fail_flag_on_404(self):
|
|
code, out, _ = _invoke(["--fail", self._url(self.nf_port)])
|
|
self.assertEqual(code, EXIT_HTTP)
|
|
self.assertIn("✗", out)
|
|
|
|
def test_redirect_followed_by_default(self):
|
|
code, out, _ = _invoke(["-v", self._url(self.redirect_port, "/redirect")])
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertIn("200", out)
|
|
self.assertIn("1 redirect", out)
|
|
|
|
def test_no_follow_redirects_flag(self):
|
|
code, out, _ = _invoke(
|
|
["--no-follow-redirects", self._url(self.redirect_port, "/redirect")]
|
|
)
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertIn("302", out)
|
|
|
|
def test_verbose_shows_protocol_row(self):
|
|
code, out, _ = _invoke(["-v", self._url()])
|
|
self.assertIn("Protocol", out)
|
|
self.assertIn("HTTP/1.1", out)
|
|
|
|
def test_json_includes_http_version_and_redirect_count(self):
|
|
code, out, _ = _invoke(["--json", "-v", self._url(self.redirect_port, "/redirect")])
|
|
data = json.loads(out)
|
|
verbose = data[0]["verbose"]
|
|
self.assertEqual(verbose["redirect_count"], 1)
|
|
self.assertIn("http_version", verbose)
|
|
|
|
def test_no_http2_flag_forces_http1(self):
|
|
code, out, _ = _invoke(["--no-http2", "-v", self._url()])
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertIn("HTTP/1.1", out)
|
|
|
|
|
|
class TestCLIFileInput(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_reads_urls_from_file(self):
|
|
with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f:
|
|
f.write(f"# comment\n\n{self._url()}\n{self._url()}\n")
|
|
path = f.name
|
|
try:
|
|
code, out, _ = _invoke(["-f", path])
|
|
finally:
|
|
os.unlink(path)
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertEqual(out.count("200"), 2)
|
|
|
|
def test_missing_file_is_usage_error(self):
|
|
code, out, err = _invoke(["-f", "/no/such/file/hxprobe-test"])
|
|
self.assertEqual(code, EXIT_USAGE)
|
|
self.assertIn("cannot read", err)
|
|
|
|
def test_empty_file_is_usage_error(self):
|
|
with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
code, out, err = _invoke(["-f", path])
|
|
finally:
|
|
os.unlink(path)
|
|
self.assertEqual(code, EXIT_USAGE)
|
|
self.assertIn("no URLs found", err)
|
|
|
|
def test_both_sources_given_is_usage_error(self):
|
|
code, out, err = _invoke(["-f", "/tmp/whatever", self._url()])
|
|
self.assertEqual(code, EXIT_USAGE)
|
|
self.assertIn("cannot combine", err)
|
|
|
|
def test_no_sources_given_is_usage_error(self):
|
|
code, out, err = _invoke([])
|
|
self.assertEqual(code, EXIT_USAGE)
|
|
self.assertIn("no URLs given", err)
|
|
|
|
|
|
class TestCLIFailures(unittest.TestCase):
|
|
def test_dns_failure(self):
|
|
code, out, _ = _invoke(["http://no.such.host.invalid"])
|
|
self.assertEqual(code, EXIT_DNS)
|
|
self.assertIn("FAILED", out)
|
|
|
|
def test_connection_refused(self):
|
|
port = _free_port()
|
|
code, out, _ = _invoke([f"http://127.0.0.1:{port}"])
|
|
self.assertEqual(code, EXIT_CONNECT)
|
|
self.assertIn("FAILED", out)
|
|
|
|
|
|
class TestCLIRunSummary(unittest.TestCase):
|
|
"""The end-of-run footer: only shown for multi-URL runs, since the single
|
|
worst-code exit can't show the mix of failure classes behind it."""
|
|
|
|
@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_multi_url_mixed_shows_summary(self):
|
|
port = _free_port()
|
|
code, out, _ = _invoke([self._url(), f"http://127.0.0.1:{port}"])
|
|
self.assertEqual(code, EXIT_CONNECT)
|
|
self.assertIn("Summary: 2 URLs", out)
|
|
self.assertIn("1 ok", out)
|
|
self.assertIn("1 failed", out)
|
|
self.assertIn("connect : 1", out)
|
|
self.assertIn("→ exit 3", out)
|
|
|
|
def test_multi_url_all_ok_summary(self):
|
|
code, out, _ = _invoke([self._url(), self._url()])
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertIn("Summary: 2 URLs — 2 ok", out)
|
|
self.assertNotIn("✗", out)
|
|
|
|
def test_single_url_has_no_summary(self):
|
|
code, out, _ = _invoke([self._url()])
|
|
self.assertEqual(code, EXIT_OK)
|
|
self.assertNotIn("Summary:", out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|