Jan Novak
24ea9c9e71
Adds the complete Python port of the Go latprobe tool plus diagnostic verbose output that the Go version does not yet have. Package (python/latprobe/): - probe.py: raw-socket HTTP timing with CertInfo/VerboseDetail dataclasses; TTFB loop accumulates to \r\n\r\n so headers are parseable without changing t_first_byte semantics; captures resolved IP, TLS version/cipher/bits, verified cert (via getpeercert()), and all response headers when verbose=True - aggregate.py: summarize() → per-phase min/avg/max PhaseStats - cli.py: injectable run(args,stdout,stderr)->int; argparse with injected streams; ThreadPoolExecutor concurrency across URLs; four text-rendering branches; JSON output; worst-exit-code accumulation; -v/--verbose flag appends IP/TLS/cert/header block after every timing table; JSON extended with "verbose" object (omitted when flag absent) - duration.py: parse Go-style duration strings (10s, 500ms, 2m, bare seconds) - Exit codes: 0 ok, 1 usage, 2 dns, 3 connect, 4 timeout, 5 tls, 6 http≥400 Tests: - test_probe.py: 18 hermetic tests (success, 404, DNS/connect/timeout/scheme failures, partial-phase invariants, verbose detail fields) - test_cli.py: 38 hermetic tests (usage errors, single/aggregate/failure text, worst-code, --fail, JSON schema/ordering/grouping, verbose text and JSON) - test_integration.py: 45 tests against live internet services (badssl.com for TLS errors, real cert/IP/header validation in verbose mode) Makefile: PYTHONPATH fix, fnmatch pattern test_[!i]*.py to exclude integration tests from py-test, new py-test-integration target Docs: docs/usage/py-latprobe.md, python/configs/usage-latprobe.md (runnable reference with captured output), plans for both the package and verbose feature Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
431 lines
16 KiB
Python
431 lines
16 KiB
Python
"""Integration tests against live internet services.
|
||
|
||
Run with:
|
||
make py-test-integration # from repo root
|
||
PYTHONPATH=python python3.14 python/tests/test_integration.py -v
|
||
|
||
These tests require network access. The full suite takes roughly 20–40 s
|
||
because TLS handshakes to badssl.com are slow and the timeout test waits
|
||
2 s for a non-routable IP to time out.
|
||
|
||
badssl.com tests are marked with a note — that host occasionally resets
|
||
connections mid-handshake, so the failure may show up as 'connect' rather
|
||
than 'tls'. Both are accepted.
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import socket
|
||
import sys
|
||
import unittest
|
||
|
||
from latprobe.cli import EXIT_DNS, EXIT_HTTP, EXIT_OK, EXIT_TIMEOUT, EXIT_TLS, run
|
||
from latprobe.probe import Options, VerboseDetail, measure
|
||
|
||
|
||
# ── connectivity guard ────────────────────────────────────────────────────────
|
||
|
||
|
||
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")
|
||
|
||
|
||
# ── helpers ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
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()
|
||
|
||
|
||
# ── success ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@_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_iana_org(self):
|
||
r = measure("https://www.iana.org")
|
||
self.assertIsNone(r.err, r.err)
|
||
self.assertEqual(r.status_code, 200)
|
||
|
||
def test_google_com(self):
|
||
r = measure("https://www.google.com")
|
||
self.assertIsNone(r.err, r.err)
|
||
self.assertIn(r.status_code, (200, 301, 302))
|
||
|
||
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")
|
||
|
||
def test_total_covers_all_present_phases(self):
|
||
r = measure("https://example.com")
|
||
phase_sum = sum(
|
||
getattr(r, a).ms
|
||
for a in ("dns", "connect", "tls", "ttfb", "transfer")
|
||
if getattr(r, a).present
|
||
)
|
||
self.assertGreaterEqual(r.total.ms, phase_sum * 0.9)
|
||
|
||
|
||
# ── DNS failure ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
@_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_invalid_tld_https(self):
|
||
r = measure("https://this-host-does-not-exist.invalid")
|
||
self.assertEqual(r.fail_phase, "dns")
|
||
self.assertFalse(r.tls.present)
|
||
|
||
def test_cli_exit_code(self):
|
||
code, out, _ = _run(["http://no.such.host.invalid"])
|
||
self.assertEqual(code, EXIT_DNS)
|
||
self.assertIn("FAILED", out)
|
||
self.assertIn("dns", out)
|
||
|
||
def test_cli_json_errors_field(self):
|
||
code, out, _ = _run(["--json", "http://no.such.host.invalid"])
|
||
self.assertEqual(code, EXIT_DNS)
|
||
data = json.loads(out)
|
||
self.assertEqual(data[0]["succeeded"], 0)
|
||
self.assertEqual(data[0]["failed"], 1)
|
||
self.assertNotIn("phases", data[0])
|
||
self.assertEqual(data[0]["errors"][0]["phase"], "dns")
|
||
|
||
|
||
# ── TLS errors (badssl.com) ───────────────────────────────────────────────────
|
||
|
||
|
||
@_NEEDS_NET
|
||
class TestTLSErrors(unittest.TestCase):
|
||
"""badssl.com occasionally resets mid-handshake; 'connect' is also accepted."""
|
||
|
||
def _assert_tls_or_connect_fail(self, url: str) -> None:
|
||
r = measure(url)
|
||
self.assertIsNotNone(r.err, f"{url} — expected failure")
|
||
self.assertIn(r.fail_phase, ("tls", "connect"),
|
||
f"unexpected phase for {url}: {r.fail_phase}")
|
||
self.assertTrue(r.dns.present, "dns should have completed")
|
||
|
||
def test_expired_cert(self):
|
||
self._assert_tls_or_connect_fail("https://expired.badssl.com/")
|
||
|
||
def test_self_signed_cert(self):
|
||
self._assert_tls_or_connect_fail("https://self-signed.badssl.com/")
|
||
|
||
def test_incomplete_chain(self):
|
||
self._assert_tls_or_connect_fail("https://incomplete-chain.badssl.com/")
|
||
|
||
def test_cli_exit_code_tls(self):
|
||
code, out, _ = _run(["https://expired.badssl.com/"])
|
||
# badssl may reset connection → connect (3) or tls (5); both > 0
|
||
self.assertGreater(code, 0)
|
||
self.assertIn("FAILED", out)
|
||
|
||
def test_partial_phases_dns_and_connect_present(self):
|
||
r = measure("https://expired.badssl.com/")
|
||
self.assertTrue(r.dns.present)
|
||
self.assertTrue(r.connect.present)
|
||
# tls phase is recorded even when it fails
|
||
self.assertTrue(r.tls.present)
|
||
|
||
|
||
# ── HTTP 4xx / 5xx ───────────────────────────────────────────────────────────
|
||
|
||
|
||
@_NEEDS_NET
|
||
class TestHTTP4xx(unittest.TestCase):
|
||
|
||
def test_google_404_probe_succeeds_at_network_level(self):
|
||
r = measure("https://www.google.com/this-page-does-not-exist-at-all-1234567890")
|
||
self.assertIsNone(r.err, r.err)
|
||
self.assertEqual(r.status_code, 404)
|
||
self.assertTrue(r.total.present)
|
||
|
||
def test_github_404_all_phases_present(self):
|
||
r = measure("https://github.com/this-repo-does-not-exist-abcxyz123/no-way")
|
||
self.assertIsNone(r.err, r.err)
|
||
self.assertEqual(r.status_code, 404)
|
||
for attr in ("dns", "connect", "tls", "ttfb", "transfer"):
|
||
self.assertTrue(getattr(r, attr).present, f"{attr} should be present")
|
||
|
||
def test_cli_exit_ok_without_fail_flag(self):
|
||
code, out, _ = _run(
|
||
["https://www.google.com/this-page-does-not-exist-at-all-1234567890"]
|
||
)
|
||
self.assertEqual(code, EXIT_OK)
|
||
self.assertIn("404", out)
|
||
self.assertNotIn("✗", out)
|
||
|
||
def test_cli_exit_http_with_fail_flag(self):
|
||
code, out, _ = _run([
|
||
"--fail",
|
||
"https://www.google.com/this-page-does-not-exist-at-all-1234567890",
|
||
])
|
||
self.assertEqual(code, EXIT_HTTP)
|
||
self.assertIn("✗", out)
|
||
|
||
def test_cli_json_404_status(self):
|
||
code, out, _ = _run([
|
||
"--json",
|
||
"https://www.google.com/this-page-does-not-exist-at-all-1234567890",
|
||
])
|
||
self.assertEqual(code, EXIT_OK)
|
||
data = json.loads(out)
|
||
self.assertEqual(data[0]["status"], 404)
|
||
self.assertEqual(data[0]["succeeded"], 1)
|
||
self.assertNotIn("errors", data[0])
|
||
|
||
def test_iana_404(self):
|
||
r = measure("https://www.iana.org/this-page-does-not-exist-either")
|
||
self.assertIsNone(r.err, r.err)
|
||
self.assertEqual(r.status_code, 404)
|
||
|
||
|
||
# ── timeout ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@_NEEDS_NET
|
||
class TestTimeout(unittest.TestCase):
|
||
"""Uses a non-routable IP (RFC 5737).
|
||
|
||
On networks that silently drop packets the connect phase waits the full
|
||
timeout → fail_phase='timeout'. On networks that return ICMP unreachable
|
||
immediately the connect fails with an OS error → fail_phase='connect'.
|
||
Both outcomes are accepted; the important assertion is that the probe fails
|
||
and the exit code is ≥ EXIT_CONNECT.
|
||
"""
|
||
|
||
_URL = "http://10.255.255.1/"
|
||
_OPTS = Options(timeout=2.0)
|
||
_FAIL_PHASES = ("timeout", "connect")
|
||
|
||
def test_probe_fail_phase(self):
|
||
r = measure(self._URL, self._OPTS)
|
||
self.assertIn(r.fail_phase, self._FAIL_PHASES)
|
||
self.assertIsNotNone(r.err)
|
||
self.assertTrue(r.dns.present) # IP literal — no real DNS lookup
|
||
self.assertTrue(r.connect.present)
|
||
self.assertTrue(r.total.present)
|
||
|
||
def test_cli_exit_code(self):
|
||
from latprobe.cli import EXIT_CONNECT
|
||
code, out, _ = _run(["--timeout", "2s", self._URL])
|
||
self.assertIn(code, (EXIT_CONNECT, EXIT_TIMEOUT))
|
||
self.assertIn("FAILED", out)
|
||
|
||
def test_cli_json_error_phase(self):
|
||
from latprobe.cli import EXIT_CONNECT
|
||
code, out, _ = _run(["--json", "--timeout", "2s", self._URL])
|
||
self.assertIn(code, (EXIT_CONNECT, EXIT_TIMEOUT))
|
||
data = json.loads(out)
|
||
self.assertIn(data[0]["errors"][0]["phase"], self._FAIL_PHASES)
|
||
|
||
|
||
# ── end-to-end multi-URL ─────────────────────────────────────────────────────
|
||
|
||
|
||
@_NEEDS_NET
|
||
class TestEndToEnd(unittest.TestCase):
|
||
|
||
def test_multi_url_worst_code_is_dns(self):
|
||
"""success (0) + dns failure (2) → worst exit = 2."""
|
||
code, out, _ = _run([
|
||
"https://example.com",
|
||
"http://no.such.host.invalid",
|
||
])
|
||
self.assertEqual(code, EXIT_DNS)
|
||
self.assertIn("200", out)
|
||
self.assertIn("FAILED", out)
|
||
|
||
def test_multi_url_output_ordered(self):
|
||
"""URLs appear in input order regardless of which resolves faster."""
|
||
code, out, _ = _run([
|
||
"https://www.iana.org",
|
||
"https://example.com",
|
||
])
|
||
self.assertEqual(code, EXIT_OK)
|
||
iana_pos = out.find("iana.org")
|
||
example_pos = out.find("example.com")
|
||
self.assertLess(iana_pos, example_pos)
|
||
|
||
def test_sampling_json_schema(self):
|
||
code, out, _ = _run(["--json", "-n", "3", "https://example.com"])
|
||
self.assertEqual(code, EXIT_OK)
|
||
data = json.loads(out)
|
||
entry = data[0]
|
||
self.assertEqual(entry["succeeded"], 3)
|
||
self.assertIn("tls", entry["phases"])
|
||
for phase, stats in entry["phases"].items():
|
||
self.assertLessEqual(stats["min_ms"], stats["avg_ms"])
|
||
self.assertLessEqual(stats["avg_ms"], stats["max_ms"])
|
||
|
||
def test_sampling_aggregate_text(self):
|
||
code, out, _ = _run(["-n", "3", "https://example.com"])
|
||
self.assertEqual(code, EXIT_OK)
|
||
self.assertIn("3 samples", out)
|
||
self.assertIn("min", out)
|
||
self.assertIn("max", out)
|
||
|
||
def test_concurrency_flag(self):
|
||
code, out, _ = _run([
|
||
"-c", "2",
|
||
"https://example.com",
|
||
"https://www.iana.org",
|
||
])
|
||
self.assertEqual(code, EXIT_OK)
|
||
self.assertIn("example.com", out)
|
||
self.assertIn("iana.org", out)
|
||
|
||
def test_mixed_fail_flag_exit_code(self):
|
||
"""200 (ok) + 404 with --fail → worst = 6."""
|
||
code, out, _ = _run([
|
||
"--fail",
|
||
"https://example.com",
|
||
"https://www.google.com/this-page-does-not-exist-at-all-1234567890",
|
||
])
|
||
self.assertEqual(code, EXIT_HTTP)
|
||
|
||
|
||
# ── verbose mode ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
@_NEEDS_NET
|
||
class TestVerboseIntegration(unittest.TestCase):
|
||
|
||
def test_https_verbose_has_ip(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertIsNotNone(r.detail)
|
||
self.assertRegex(r.detail.resolved_ip, r"^\d+\.\d+\.\d+\.\d+$",
|
||
"expected IPv4 address")
|
||
|
||
def test_https_verbose_has_tls_version(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertIn("TLS", r.detail.tls_version,
|
||
f"expected TLSv1.x, got: {r.detail.tls_version!r}")
|
||
|
||
def test_https_verbose_has_cipher(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertNotEqual(r.detail.tls_cipher, "")
|
||
|
||
def test_https_verbose_cert_cn(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertIsNotNone(r.detail.cert)
|
||
self.assertIn("example.com", r.detail.cert.cn)
|
||
|
||
def test_https_verbose_cert_expiry_format(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
expiry = r.detail.cert.expiry
|
||
self.assertRegex(expiry, r"^\d{4}-\d{2}-\d{2}$",
|
||
f"expected YYYY-MM-DD, got: {expiry!r}")
|
||
|
||
def test_https_verbose_cert_verified(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertTrue(r.detail.cert.verified)
|
||
|
||
def test_https_verbose_cert_issuer_set(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertNotEqual(r.detail.cert.issuer_cn, "")
|
||
|
||
def test_https_verbose_headers_present(self):
|
||
r = measure("https://example.com", Options(verbose=True))
|
||
self.assertGreater(len(r.detail.headers), 0)
|
||
# example.com always sends Content-Type
|
||
self.assertIn("Content-Type", r.detail.headers)
|
||
|
||
def test_http_verbose_no_tls(self):
|
||
r = measure("http://example.com", Options(verbose=True))
|
||
self.assertEqual(r.detail.tls_version, "")
|
||
self.assertIsNone(r.detail.cert)
|
||
|
||
def test_redirect_verbose_shows_location(self):
|
||
r = measure("http://example.com", Options(verbose=True))
|
||
# example.com HTTP redirects to HTTPS — Location header should be present
|
||
if r.status_code in (301, 302, 307, 308):
|
||
self.assertIn("Location", r.detail.headers)
|
||
|
||
def test_dns_fail_verbose_no_ip(self):
|
||
r = measure("http://no.such.host.invalid", Options(verbose=True))
|
||
self.assertIsNotNone(r.detail)
|
||
self.assertEqual(r.detail.resolved_ip, "")
|
||
self.assertEqual(r.detail.headers, {})
|
||
|
||
def test_cli_verbose_text_shows_ip(self):
|
||
code, out, _ = _run(["--verbose", "https://example.com"])
|
||
self.assertEqual(code, EXIT_OK)
|
||
self.assertIn("IP", out)
|
||
self.assertIn("TLS", out)
|
||
self.assertIn("Cert", out)
|
||
self.assertIn("Content-Type", out)
|
||
|
||
def test_cli_verbose_json_schema(self):
|
||
code, out, _ = _run(["--verbose", "--json", "https://example.com"])
|
||
self.assertEqual(code, EXIT_OK)
|
||
data = json.loads(out)
|
||
v = data[0]["verbose"]
|
||
self.assertIn("ip", v)
|
||
self.assertIn("tls_version", v)
|
||
self.assertIn("tls_cipher", v)
|
||
self.assertIn("tls_bits", v)
|
||
self.assertIn("cert", v)
|
||
self.assertIn("headers", v)
|
||
|
||
def test_cli_verbose_json_cert_fields(self):
|
||
code, out, _ = _run(["--verbose", "--json", "https://example.com"])
|
||
data = json.loads(out)
|
||
cert = data[0]["verbose"]["cert"]
|
||
self.assertIn("cn", cert)
|
||
self.assertIn("expiry", cert)
|
||
self.assertIn("issuer_cn", cert)
|
||
self.assertTrue(cert["verified"])
|
||
|
||
def test_cli_verbose_tls_fail_shows_ip(self):
|
||
code, out, _ = _run(["--verbose", "https://expired.badssl.com/"])
|
||
self.assertGreater(code, 0)
|
||
# IP should be shown even when TLS fails (DNS + TCP succeeded)
|
||
self.assertIn("IP", out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|