feat(hxprobe): httpx-based HTTP probe — full standalone package
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>
This commit is contained in:
0
hxprobe/tests/__init__.py
Normal file
0
hxprobe/tests/__init__.py
Normal file
260
hxprobe/tests/test_cli.py
Normal file
260
hxprobe/tests/test_cli.py
Normal file
@@ -0,0 +1,260 @@
|
||||
"""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()
|
||||
122
hxprobe/tests/test_integration.py
Normal file
122
hxprobe/tests/test_integration.py
Normal file
@@ -0,0 +1,122 @@
|
||||
"""Integration tests against live internet services for hxprobe.
|
||||
|
||||
Run with:
|
||||
make hx-test-integration
|
||||
cd hxprobe && uv run pytest tests -m integration -v
|
||||
|
||||
Marked with `pytest.mark.integration` (see module-level `pytestmark` below)
|
||||
so it's excluded from the default `hx-test`/`make test` gate, which runs
|
||||
`pytest -m "not integration"` — these hit the real internet.
|
||||
|
||||
The point of this file is specifically hxprobe's headline capabilities: real
|
||||
HTTP/2 negotiation and redirect-following.
|
||||
"""
|
||||
|
||||
import io
|
||||
import socket
|
||||
import unittest
|
||||
|
||||
import pytest
|
||||
|
||||
from hxprobe.cli import EXIT_DNS, EXIT_OK, run
|
||||
from hxprobe.probe import Options, measure
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def _online() -> bool:
|
||||
try:
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.settimeout(3)
|
||||
s.connect(("8.8.8.8", 53))
|
||||
s.close()
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
_NEEDS_NET = unittest.skipUnless(_online(), "no internet connectivity")
|
||||
|
||||
|
||||
def _run(args: list[str]) -> tuple[int, str, str]:
|
||||
out, err = io.StringIO(), io.StringIO()
|
||||
code = run(args, out, err)
|
||||
return code, out.getvalue(), err.getvalue()
|
||||
|
||||
|
||||
@_NEEDS_NET
|
||||
class TestHTTP2Negotiation(unittest.TestCase):
|
||||
"""The headline Go-parity feature: real ALPN HTTP/2, not forced HTTP/1.1."""
|
||||
|
||||
def test_negotiates_h2_against_cloudflare(self):
|
||||
r = measure("https://example.com", Options(verbose=True))
|
||||
self.assertIsNone(r.err, r.err)
|
||||
self.assertEqual(r.detail.http_version, "HTTP/2")
|
||||
|
||||
def test_no_http2_flag_forces_http1(self):
|
||||
code, out, _ = _run(["--no-http2", "-v", "https://example.com"])
|
||||
self.assertEqual(code, EXIT_OK)
|
||||
self.assertIn("HTTP/1.1", out)
|
||||
self.assertNotIn("HTTP/2", out)
|
||||
|
||||
|
||||
@_NEEDS_NET
|
||||
class TestRedirectsLive(unittest.TestCase):
|
||||
"""The other headline Go-parity feature: redirects followed by default."""
|
||||
|
||||
def test_follows_http_to_https_redirect(self):
|
||||
code, out, _ = _run(["-v", "http://github.com"])
|
||||
self.assertEqual(code, EXIT_OK)
|
||||
self.assertIn("200", out)
|
||||
self.assertIn("redirect", out)
|
||||
|
||||
def test_no_follow_redirects_reports_redirect_status(self):
|
||||
code, out, _ = _run(["--no-follow-redirects", "http://github.com"])
|
||||
self.assertEqual(code, EXIT_OK)
|
||||
self.assertNotIn("redirect_count", out) # text mode never shows the raw key
|
||||
self.assertTrue(any(code_str in out for code_str in ("301", "302", "307", "308")))
|
||||
|
||||
|
||||
@_NEEDS_NET
|
||||
class TestSuccess(unittest.TestCase):
|
||||
def test_https_all_phases_present(self):
|
||||
r = measure("https://example.com")
|
||||
self.assertIsNone(r.err, r.err)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertTrue(r.dns.present)
|
||||
self.assertTrue(r.connect.present)
|
||||
self.assertTrue(r.tls.present)
|
||||
self.assertTrue(r.ttfb.present)
|
||||
self.assertTrue(r.transfer.present)
|
||||
self.assertTrue(r.total.present)
|
||||
|
||||
def test_http_no_tls_phase(self):
|
||||
r = measure("http://example.com")
|
||||
self.assertIsNone(r.err, r.err)
|
||||
self.assertFalse(r.tls.present)
|
||||
|
||||
def test_timings_are_positive(self):
|
||||
r = measure("https://example.com")
|
||||
for attr in ("dns", "connect", "tls", "ttfb", "transfer", "total"):
|
||||
ph = getattr(r, attr)
|
||||
if ph.present:
|
||||
self.assertGreater(ph.ms, 0, f"{attr}.ms should be > 0")
|
||||
|
||||
|
||||
@_NEEDS_NET
|
||||
class TestDNSFailure(unittest.TestCase):
|
||||
def test_invalid_tld_phase(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_cli_dns_failure_exit_code(self):
|
||||
code, out, _ = _run(["http://no.such.host.invalid"])
|
||||
self.assertEqual(code, EXIT_DNS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
256
hxprobe/tests/test_probe.py
Normal file
256
hxprobe/tests/test_probe.py
Normal file
@@ -0,0 +1,256 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user