diff --git a/.gitignore b/.gitignore index aa8c000..8dfe0b1 100644 --- a/.gitignore +++ b/.gitignore @@ -2,7 +2,6 @@ go/latprobe go/coverage.out go/coverage.html -python/latprobe # Python __pycache__/ diff --git a/CHANGELOG.md b/CHANGELOG.md index a1050e0..f083e73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,52 @@ All completed features are logged here in reverse-chronological order. --- +## 2026-07-01 12:55 — `--verbose` / `-v` flag for `latprobe` (Python) + +- `-v`/`--verbose` flag added to the `latprobe` CLI +- `probe.py`: new `CertInfo` + `VerboseDetail` dataclasses; `Options.verbose`; + `Result.detail`; TTFB loop now accumulates until `\r\n\r\n` (unchanged TTFB + semantics — `t_first_byte` stamped on first `recv()`, not when headers complete); + captures resolved IP (from `getaddrinfo`), TLS version/cipher/bits + (from `sock.version()`/`sock.cipher()`), verified cert (from `getpeercert()`), + and all response headers +- `cli.py`: verbose text block appended after timing rows in all four output + branches (single, aggregate, all-failed, mixed); `_VERBOSE_HEADERS` priority + list controls which headers appear in text mode; JSON `"verbose"` object + includes all parsed headers, full cert fields, TLS metadata; `"verbose"` key + omitted when flag is absent +- 9 new hermetic probe tests, 15 new CLI tests covering verbose text and JSON + across success, connect-fail, and DNS-fail scenarios +- 15 new integration tests for real TLS cert fields (CN, expiry, issuer), + IP format, TLS version string, header presence, and verbose text/JSON output +- `python/configs/usage-latprobe.md`: runnable reference with real output for + all features including verbose and `--verbose --json` +- `docs/usage/py-latprobe.md`: updated with `--verbose` flag and examples +- Plan: `docs/plans/2026-07-01-12-55-py-latprobe-verbose.md` + +--- + +## 2026-07-01 12:23 — Full `latprobe` Python package (Python, Step 3) + +- `python/latprobe/` package: full port of the Go CLI, runnable as `python -m latprobe` +- `probe.py`: raw-socket HTTP measurement with `Options(timeout)` parameter; mirrors + `phases.py` technique; partial phases preserved on failure +- `aggregate.py`: `summarize(List[Result]) -> Aggregate` with per-phase `PhaseStats(min_ms, avg_ms, max_ms)` +- `duration.py`: parses Go-style duration strings (`10s`, `500ms`, `2m`, bare seconds) +- `cli.py`: injectable `run(args, stdout, stderr) -> int`; argparse with injected streams + (`_Parser` subclass); `ThreadPoolExecutor` concurrency across URLs; four text-rendering + branches (single, aggregate, all-failed, mixed); JSON output; worst-exit-code accumulation +- Exit codes: 0 ok, 1 usage, 2 dns, 3 connect, 4 timeout, 5 tls, 6 http≥400 (`--fail`) +- `python/tests/test_probe.py`: 9 tests — success, 404, DNS fail, connect refused, TTFB + timeout, bad scheme, partial phase invariants; uses `http.server` + daemon threads +- `python/tests/test_cli.py`: 23 tests — drives `cli.run()` in-process; covers usage + errors, single/aggregate text, failure exit codes, worst-code, `--fail`, JSON schema, + JSON ordering, JSON error grouping +- Makefile `py-test`: added `PYTHONPATH=$(PY_DIR)`, removed `|| true` +- User doc: `docs/usage/py-latprobe.md` + +--- + ## 2026-07-01 12:04 — Per-phase latency measurement (Python, Step 2) - `python/phases.py`: self-contained script; hand-drives raw sockets to time diff --git a/Makefile b/Makefile index 8dc44ab..c774fec 100644 --- a/Makefile +++ b/Makefile @@ -112,11 +112,15 @@ py-run: ## py: run the full latprobe package (pass flags via ARGS="…") cd $(PY_DIR) && $(PYTHON) -m latprobe $(ARGS) .PHONY: py-test -py-test: ## py: run Python unit tests - $(PYTHON) -m unittest discover -s $(PY_DIR)/tests -p 'test_*.py' -v 2>&1 || true +py-test: ## py: run Python unit tests (local, hermetic) + PYTHONPATH=$(PY_DIR) $(PYTHON) -m unittest discover -s $(PY_DIR)/tests -p 'test_[!i]*.py' -v $(ARGS) + +.PHONY: py-test-integration +py-test-integration: ## py: run integration tests against live internet services (~30 s) + PYTHONPATH=$(PY_DIR) $(PYTHON) -m unittest discover -s $(PY_DIR)/tests -p 'test_integration.py' -v $(ARGS) .PHONY: py-check -py-check: py-test ## py: run Python test gate +py-check: py-test ## py: run Python test gate (hermetic only) .PHONY: py-clean py-clean: ## py: remove Python bytecode and __pycache__ dirs diff --git a/docs/plans/2026-07-01-12-23-py-latprobe.md b/docs/plans/2026-07-01-12-23-py-latprobe.md new file mode 100644 index 0000000..6be8cf6 --- /dev/null +++ b/docs/plans/2026-07-01-12-23-py-latprobe.md @@ -0,0 +1,139 @@ +# Plan: Full `latprobe` Python Package + +**Timestamp:** 2026-07-01 12:23 +**Status:** Complete + +## Goal + +Build `python/latprobe/` — a packaged Python port of the Go `latprobe` CLI that +mirrors its behavior: flags, sampling, cross-URL concurrency, JSON output, and +distinct exit codes. + +## Package layout + +``` +python/ + latprobe/ + __init__.py — package marker + __main__.py — sys.exit(cli.run(sys.argv[1:], sys.stdout, sys.stderr)) + probe.py — measure(url, opts) -> Result (raw socket timing) + aggregate.py — summarize(results) -> Aggregate (min/avg/max per phase) + cli.py — run(args, stdout, stderr) -> int (argparse + rendering) + duration.py — parse_duration("10s") -> float seconds + tests/ + test_probe.py — unittest: success, DNS fail, connect refused, timeout, bad scheme + test_cli.py — unittest: drives cli.run() in-process, all branches + JSON +``` + +## Key design decisions + +- **`probe.py`**: reuses the raw-socket technique from `phases.py` with an + `Options(timeout)` parameter. Partial phases preserved on failure. +- **`aggregate.py`**: `summarize(List[Result]) -> Aggregate` computes + `PhaseStats(min_ms, avg_ms, max_ms, present)` per phase. Uses only `succeeded` + results. +- **`duration.py`**: parses `"10s"`, `"500ms"`, `"2m"`, bare numbers → float seconds. +- **`cli.py`**: injectable `run(args, stdout, stderr) -> int` seam for testability. + - `_Parser` subclasses `argparse.ArgumentParser` with custom `print_help`, + `print_usage`, `error`, `exit` to redirect all output to injected streams + and raise `_ArgExit` instead of calling `sys.exit`. + - `ThreadPoolExecutor` for cross-URL concurrency; `executor.map` preserves + input order. + - Four text-rendering branches: single-success, multi-sample aggregate, + all-failed, mixed. + - Failure summary groups errors by `(phase, message)` preserving insertion + order; `N ×` prefix when grouped. + +## CLI flags + +| Flag | Default | Description | +|------|---------|-------------| +| `urls` (positional, `+`) | required | One or more URLs | +| `-n`/`--count` | 1 | Requests per URL | +| `-c`/`--concurrency` | 0 (auto) | Max parallel URLs; auto = `min(numURLs, 8)` | +| `--timeout` | `10s` | Per-request timeout (parsed via `duration.py`) | +| `--fail` | off | Exit non-zero on HTTP status >= 400 | +| `--json` | off | Output as JSON array instead of text | + +## Exit codes + +| Code | Meaning | +|------|---------| +| 0 | All probes succeeded (with `--fail`: no 4xx/5xx) | +| 1 | Usage / argument error | +| 2 | DNS failure | +| 3 | TCP connect failure | +| 4 | Timeout | +| 5 | TLS error | +| 6 | HTTP status >= 400 (`--fail` only) | + +Worst (highest) code across all URLs wins. + +## JSON schema + +```json +[ + { + "url": "https://example.com", + "status": 200, + "succeeded": 3, + "failed": 2, + "phases": { + "dns": {"min_ms": 1.23, "avg_ms": 1.45, "max_ms": 1.67}, + "connect": {"min_ms": ...}, + "tls": {"min_ms": ...}, + "ttfb": {"min_ms": ...}, + "transfer": {"min_ms": ...}, + "total": {"min_ms": ...} + }, + "errors": [ + {"phase": "connect", "count": 2, "message": "..."} + ] + } +] +``` + +- `phases` omitted when `succeeded == 0` +- Per-phase key omitted when no successful sample had that phase (e.g. `tls` for `http://`) +- `errors` omitted when `failed == 0` + +## Text output format + +Single sample: +``` +https://example.com (200) + DNS lookup : 18.87 ms + TCP connect : 9.76 ms + TLS handshake : 14.71 ms + Server (TTFB) : 67.07 ms + Transfer : 0.14 ms + ───────────────────────────── + Total : 118.39 ms +``` + +Aggregate (-n 3): +``` +https://example.com (200, 3 samples) + min avg max + DNS lookup : 2.36 ms 3.51 ms 5.00 ms + TCP connect : 10.25 ms 17.45 ms 31.61 ms + TLS handshake : 14.04 ms 41.60 ms 96.50 ms + Server (TTFB) : 63.30 ms 66.72 ms 69.78 ms + Transfer : 0.26 ms 0.38 ms 0.51 ms + ───────────────────────────────────────────────── + Total : 101.33 ms 139.19 ms 211.99 ms +``` + +## Testing approach + +- `test_probe.py`: real `http.server.HTTPServer` in a daemon thread; closed port + for connection-refused; `.invalid` TLD for DNS failure; black-hole socket (accepts + TCP, never sends data) to trigger TTFB timeout. +- `test_cli.py`: in-process `cli.run(args, io.StringIO(), io.StringIO())` — no + subprocess, fast. Covers: usage errors, single/aggregate text, failure exit codes, + worst-code accumulation, `--fail`, JSON schema, JSON ordering, JSON error grouping. + +## Makefile changes + +- `py-test`: added `PYTHONPATH=$(PY_DIR)` so tests can `import latprobe`; removed + `|| true` now that real tests exist. diff --git a/docs/plans/2026-07-01-12-55-py-latprobe-verbose.md b/docs/plans/2026-07-01-12-55-py-latprobe-verbose.md new file mode 100644 index 0000000..566b713 --- /dev/null +++ b/docs/plans/2026-07-01-12-55-py-latprobe-verbose.md @@ -0,0 +1,505 @@ +# Plan: `--verbose` / `-v` flag for `latprobe` + +**Timestamp:** 2026-07-01 12:55 +**Status:** Planned + +## Goal + +Add a `--verbose` / `-v` flag that surfaces diagnostic detail beyond timing: +which IP was used, what TLS version and cipher was negotiated, certificate +metadata (CN, expiry, issuer), and the most useful response headers. Useful +for debugging *why* a probe failed, not just *that* it did. + +--- + +## What verbose mode shows + +### Text output — successful HTTPS request + +``` +https://example.com (200) + DNS lookup : 18.87 ms + TCP connect : 9.76 ms + TLS handshake : 14.71 ms + Server (TTFB) : 67.07 ms + Transfer : 0.14 ms + ───────────────────────────── + Total : 118.39 ms + IP : 93.184.216.34 + TLS : TLSv1.3 TLS_AES_256_GCM_SHA384 256 bit + Cert : CN=www.example.com valid until 2025-06-14 DigiCert Inc + Server : ECS (dcb/7F84) + Content-Type : text/html; charset=UTF-8 +``` + +### Text output — TLS failure (with diagnostic cert pass) + +``` +https://expired.badssl.com/ (FAILED) + DNS lookup : 15.39 ms + TCP connect : 124.98 ms + TLS handshake : 305.30 ms + ───────────────────────────── + Total : 473.17 ms + ✗ tls: certificate verify failed: certificate has expired + IP : 104.154.89.105 + Cert (unverified) : CN=*.badssl.com EXPIRED 2015-04-09 COMODO RSA Domain Validation +``` + +### Text output — DNS failure (minimal; nothing to show after DNS) + +``` +http://no.such.host.invalid (FAILED) + Total : 0.65 ms + ✗ dns: [Errno 8] nodename nor servname provided, or not known +``` +No verbose block — IP is unknown, no TLS, no headers. + +### Text output — HTTP 4xx with redirect header + +``` +https://www.google.com/this-page-does-not-exist (404) + ... + Total : 145.50 ms + IP : 142.251.36.4 + TLS : TLSv1.3 TLS_AES_128_GCM_SHA256 128 bit + Cert : CN=*.google.com valid until 2026-01-20 Google Trust Services + Server : gws + Content-Type : text/html; charset=UTF-8 +``` + +### Text output — 3xx redirect (Location shown) + +``` +https://google.com (301) + ... + IP : 142.251.36.4 + TLS : TLSv1.3 TLS_AES_128_GCM_SHA256 128 bit + Cert : CN=*.google.com valid until 2026-01-20 Google Trust Services + Location : https://www.google.com/ + Server : gws + Content-Type : text/html; charset=UTF-8 +``` + +### Aggregate verbose (multi-sample `-n N`) + +``` +https://example.com (200, 5 samples) + min avg max + ... + Total : 97.10 ms 101.20 ms 109.80 ms + IP : 93.184.216.34 + TLS : TLSv1.3 TLS_AES_256_GCM_SHA384 256 bit + Cert : CN=www.example.com valid until 2025-06-14 DigiCert Inc + Server : ECS (dcb/7F84) + Content-Type : text/html; charset=UTF-8 +``` +Verbose block taken from the **last successful sample**. If DNS resolves to +multiple IPs and different samples hit different addresses, they are listed as +`IP : 93.184.216.34, 93.184.216.35` (deduped, insertion-ordered). + +--- + +## Verbose block format rules + +- Label column: **14 chars**, left-padded with spaces, matching the phase labels. +- Separator: `" : "` (4 chars) — same as phase rows. +- Value: free-form string, no fixed width. +- Verbose block appears immediately after the last line of the standard block + (after `Total` or after the `✗ …` error line). +- Block is omitted entirely when there is nothing to show (e.g. pure DNS + failure where IP is unknown). +- No separator line before the verbose block — the visual break from `Total` + and `✗` is enough. + +Label strings (14 chars each): +``` +"IP " # resolved IP address +"TLS " # version + cipher + bits +"Cert " # CN, expiry, issuer (verified) +"Cert (unvrf.) " # same fields, cert not trusted (TLS failure) +"Location " # for 3xx responses +"Server " # Server response header +"Content-Type " # Content-Type response header +"X-Cache " # CDN cache status (if present) +"Via " # proxy chain (if present) +``` + +Additional headers shown only when present (see priority list in +`_VERBOSE_HEADERS` below). + +--- + +## JSON extension + +When `--verbose --json` are both set, each entry gets a top-level `"verbose"` +object (omitted when `--verbose` is absent): + +```json +{ + "url": "https://example.com", + "status": 200, + "succeeded": 1, + "failed": 0, + "phases": { ... }, + "verbose": { + "ip": "93.184.216.34", + "tls_version": "TLSv1.3", + "tls_cipher": "TLS_AES_256_GCM_SHA384", + "tls_bits": 256, + "cert": { + "cn": "www.example.com", + "sans": ["www.example.com", "example.com"], + "expiry": "2025-06-14", + "issuer_cn": "DigiCert SHA2 Secure Server CA", + "verified": true + }, + "headers": { + "Server": "ECS (dcb/7F84)", + "Content-Type": "text/html; charset=UTF-8", + "X-Cache": "HIT" + } + } +} +``` + +- `"verbose"` is omitted when `--verbose` is not set. +- `"cert"` is omitted for `http://` URLs (no TLS). +- For TLS failures: `"cert"` contains what the diagnostic pass found, with + `"verified": false`. If the diagnostic pass itself failed, `"cert"` is omitted. +- `"headers"` contains **all** parsed response headers (not just the priority + list used in text mode). +- When `succeeded == 0` the `"verbose"` object may still contain `"ip"` if DNS + resolved, but will lack `"tls_version"`, `"cert"`, and `"headers"`. + +--- + +## Data model changes + +### `probe.py` — new dataclasses + +```python +@dataclass +class CertInfo: + cn: str = "" + sans: list[str] = field(default_factory=list) + expiry: str = "" # ISO date "YYYY-MM-DD" + issuer_cn: str = "" + verified: bool = False # True = TLS handshake passed verification + + +@dataclass +class VerboseDetail: + resolved_ip: str = "" + tls_version: str = "" + tls_cipher: str = "" + tls_bits: int = 0 + cert: CertInfo | None = None + headers: dict[str, str] = field(default_factory=dict) # all parsed headers +``` + +### `probe.py` — `Options` change + +```python +@dataclass +class Options: + timeout: float = 10.0 + verbose: bool = False # NEW +``` + +### `probe.py` — `Result` change + +```python +@dataclass +class Result: + ...existing fields... + detail: VerboseDetail | None = None # NEW; None when opts.verbose=False +``` + +--- + +## Capture points in `probe.py` + +### Resolved IP +After `socket.getaddrinfo` succeeds: +```python +if opts.verbose: + r.detail = VerboseDetail(resolved_ip=str(infos[0][4][0])) +``` +`infos[0][4][0]` is the first resolved address string (works for both IPv4 +and IPv6 since `[4]` is the full sockaddr tuple and `[0]` is the address). + +### TLS version, cipher, and certificate +After `ctx.wrap_socket()` succeeds: +```python +if opts.verbose and r.detail: + r.detail.tls_version = sock.version() or "" + cipher_name, _, bits = sock.cipher() + r.detail.tls_cipher = cipher_name or "" + r.detail.tls_bits = bits or 0 + r.detail.cert = _parse_cert(sock.getpeercert(), verified=True) +``` +`sock.getpeercert()` returns a dict with `subject`, `issuer`, `notAfter`, +`subjectAltName` when called after a successful handshake. + +### Response headers +The current code reads the first non-empty chunk then drains the body. +For verbose mode (and more correctly in general), the probe must accumulate +bytes until `\r\n\r\n` is found before recording `t_first_byte`, so that the +full header section is available. + +Modified TTFB loop (replaces the current `while not first_chunk:` block): + +```python +buf = b"" +t_first_byte: float | None = None +while b"\r\n\r\n" not in buf: + chunk = sock.recv(4096) + if not chunk: + raise OSError("server closed connection before headers complete") + if t_first_byte is None: + t_first_byte = time.perf_counter() # first byte — unchanged semantics + buf += chunk + +t_first_byte = t_first_byte or time.perf_counter() +r.ttfb = _p(t_wrote, t_first_byte) +r.status_code = _parse_status(buf) +if opts.verbose and r.detail: + r.detail.headers = _parse_response_headers(buf) +``` + +TTFB semantics are **unchanged** — `t_first_byte` is captured on the first +`recv()` call that returns data, not after all headers arrive. + +The transfer drain loop needs to also drain `buf` bytes that follow +`\r\n\r\n` (the body prefix already read into the buffer). + +### Certificate inspection on TLS failure +When `opts.verbose=True` and `r.fail_phase == "tls"`, call a helper that +opens a second, non-verifying connection purely to fetch the certificate: + +```python +def _fetch_cert_unverified( + host: str, port: int, addr, family: int, timeout: float +) -> CertInfo | None: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + sock = socket.socket(family, socket.SOCK_STREAM) + sock.settimeout(timeout) + try: + sock.connect(addr) + ssl_sock = ctx.wrap_socket(sock, server_hostname=host) + peer = ssl_sock.getpeercert() + ssl_sock.close() + return _parse_cert(peer, verified=False) if peer else None + except OSError: + return None + finally: + sock.close() +``` + +This runs **after** `measure()` records timing, so it does not affect any +phase durations. It adds a short additional wall-clock delay (one more TLS +round trip) only in verbose mode on TLS failures — acceptable for a diagnostic +path. + +### Certificate parsing helper + +```python +from email.utils import parsedate # for "Jun 14 00:00:00 2025 GMT" +import datetime + +def _parse_cert(peer: dict, verified: bool) -> CertInfo: + def _cn(rdns): + for rdn in rdns: + for k, v in rdn: + if k == "commonName": + return v + return "" + + sans = [v for k, v in peer.get("subjectAltName", ()) if k == "DNS"] + expiry = "" + not_after = peer.get("notAfter", "") + if not_after: + t = parsedate(not_after) # returns time.struct_time or None + if t: + expiry = f"{t[0]:04d}-{t[1]:02d}-{t[2]:02d}" + + return CertInfo( + cn = _cn(peer.get("subject", ())), + sans = sans, + expiry = expiry, + issuer_cn = _cn(peer.get("issuer", ())), + verified = verified, + ) +``` + +### Response header parsing helper + +```python +_VERBOSE_HEADERS = [ + "Location", "Server", "Content-Type", "X-Cache", + "CF-Cache-Status", "Cache-Control", "Via", + "X-Powered-By", "Strict-Transport-Security", +] + +def _parse_response_headers(buf: bytes) -> dict[str, str]: + """Parse all headers from a raw HTTP response buffer (up to \\r\\n\\r\\n).""" + header_section = buf.split(b"\r\n\r\n", 1)[0] + lines = header_section.decode("latin-1", errors="replace").splitlines() + headers: dict[str, str] = {} + for line in lines[1:]: # skip status line + if ":" in line: + name, _, value = line.partition(":") + headers[name.strip()] = value.strip() + return headers +``` + +--- + +## `cli.py` changes + +### New flag +```python +parser.add_argument( + "-v", "--verbose", action="store_true", + help="show resolved IP, TLS details, certificate, and response headers", +) +``` +Passed into `Options(verbose=ns.verbose)`. + +### Text rendering — verbose block + +```python +def _print_verbose_block(detail: VerboseDetail, out: IO) -> None: + rows: list[tuple[str, str]] = [] + + if detail.resolved_ip: + rows.append(("IP ", detail.resolved_ip)) + + if detail.tls_version: + tls_val = detail.tls_version + if detail.tls_cipher: + tls_val += f" {detail.tls_cipher}" + if detail.tls_bits: + tls_val += f" {detail.tls_bits} bit" + rows.append(("TLS ", tls_val)) + + if detail.cert: + c = detail.cert + label = "Cert (unvrf.) " if not c.verified else "Cert " + parts = [f"CN={c.cn}"] if c.cn else [] + if c.expiry: + tag = "EXPIRED" if _cert_expired(c.expiry) else "valid until" + parts.append(f"{tag} {c.expiry}") + if c.issuer_cn: + parts.append(c.issuer_cn) + rows.append((label, " ".join(parts))) + + # Response headers — show priority list, in order, if present + for name in _VERBOSE_HEADERS: + val = detail.headers.get(name) + if val: + label = f"{name:<14}" + rows.append((label, val)) + + for label, value in rows: + out.write(f" {label} : {value}\n") +``` + +`_cert_expired(expiry: str) -> bool` compares the ISO date string against +today's date using `datetime.date.fromisoformat`. + +Call site: `_print_verbose_block` is called from `_print_single`, +`_print_all_failed`, and `_print_aggregate` — after the last line written by +each function — but only when `result.detail` (or `agg`'s last-sample detail) +is not `None`. + +For aggregate rendering: collect `detail` from the last successful result +before calling `summarize`. Pass it alongside `agg` to `_print_aggregate`. + +### JSON extension + +In `_build_json_entry`, when `detail` is not `None`: +```python +v: dict = {} +if detail.resolved_ip: + v["ip"] = detail.resolved_ip +if detail.tls_version: + v["tls_version"] = detail.tls_version + v["tls_cipher"] = detail.tls_cipher + v["tls_bits"] = detail.tls_bits +if detail.cert: + c = detail.cert + v["cert"] = { + "cn": c.cn, + "sans": c.sans, + "expiry": c.expiry, + "issuer_cn": c.issuer_cn, + "verified": c.verified, + } +if detail.headers: + v["headers"] = dict(detail.headers) +if v: + entry["verbose"] = v +``` + +--- + +## Files changed + +| File | Change | +|------|--------| +| `python/latprobe/probe.py` | `CertInfo`, `VerboseDetail` dataclasses; `Options.verbose`; `Result.detail`; capture points for IP, TLS, cert, headers; `_fetch_cert_unverified`; `_parse_cert`; `_parse_response_headers`; modified TTFB loop | +| `python/latprobe/cli.py` | `-v`/`--verbose` flag; `_print_verbose_block`; verbose call sites in all 3 print functions; `_build_json_entry` extension; `_cert_expired` helper | +| `python/tests/test_probe.py` | Tests for verbose fields on success (IP, TLS, cert, headers), on TLS failure (unverified cert), and that non-verbose leaves `detail=None` | +| `python/tests/test_cli.py` | Tests for `--verbose` text layout (IP/TLS/cert/header rows), `--verbose --json` schema, verbose absent when flag not set | +| `python/tests/test_integration.py` | Tests for real site verbose output (expiry date format, header values, actual IPs), TLS failure cert inspection against badssl.com | +| `docs/usage/py-latprobe.md` | New `--verbose` section with example output | + +--- + +## Implementation order + +1. **`probe.py` — data model**: add `CertInfo`, `VerboseDetail`, `Options.verbose`, + `Result.detail`. No behaviour change yet — all `None` by default. +2. **`probe.py` — capture IP**: set `r.detail.resolved_ip` after `getaddrinfo`. + Easiest capture; good checkpoint. +3. **`probe.py` — modify TTFB loop**: accumulate to `\r\n\r\n`, update body drain. + This is the riskiest change (touches the hot path); verify existing tests + still pass before continuing. +4. **`probe.py` — parse headers**: add `_parse_response_headers`, `_VERBOSE_HEADERS`; + populate `r.detail.headers` when verbose. +5. **`probe.py` — TLS details on success**: add `_parse_cert`; populate + `tls_version`, `tls_cipher`, `tls_bits`, `cert` after `wrap_socket`. +6. **`probe.py` — TLS failure cert**: add `_fetch_cert_unverified`; call it at the + end of `measure` when `opts.verbose and r.fail_phase == "tls"`. +7. **`cli.py` — flag + text rendering**: add `-v`, `_print_verbose_block`, + `_cert_expired`; wire into all print functions. +8. **`cli.py` — JSON**: extend `_build_json_entry`. +9. **Tests**: hermetic tests for each capture point; CLI text/JSON tests; + integration tests against real sites. +10. **Docs**: update `docs/usage/py-latprobe.md`. + +Steps 1–2 and 7 can be skipped ahead and demonstrated early (IP line shows up +immediately); steps 3–6 build the richer detail progressively. + +--- + +## Known limitations / out of scope + +- **Aggregate verbose from last sample only** — no per-sample IP list unless + sampling actually returned multiple distinct IPs (rare; only with round-robin + DNS between samples). +- **DNS timeout not controllable** — `getaddrinfo` does not accept a Python + timeout; the `--timeout` flag applies only from TCP connect onward. Already + documented in existing usage doc. +- **No redirect following** — 3xx responses show the `Location` header in the + verbose block but do not probe the target. Consistent with non-verbose + behaviour. +- **Diagnostic cert pass adds latency** — only on TLS failures in verbose mode; + documented inline in output (future: could suppress with a flag). +- **HTTP/2, HTTP/3 not supported** — raw socket drives HTTP/1.1 only; TLS + ALPN negotiation may result in some servers rejecting the connection. Out of + scope for this tool. diff --git a/docs/usage/py-latprobe.md b/docs/usage/py-latprobe.md new file mode 100644 index 0000000..80f7bfd --- /dev/null +++ b/docs/usage/py-latprobe.md @@ -0,0 +1,294 @@ +# `latprobe/` — Full Python Port + +## What it does + +A packaged Python CLI that mirrors the Go `latprobe` tool: per-phase HTTP +latency measurement (DNS, TCP connect, TLS, TTFB, Transfer, Total) with +configurable sampling, cross-URL concurrency, JSON output, and distinct exit +codes per failure class. + +Run it from the `python/` directory with `python -m latprobe`. + +Phases measured: + +| Phase | What is timed | +|-------|---------------| +| DNS lookup | `socket.getaddrinfo()` — hostname resolution | +| TCP connect | `sock.connect()` — SYN to connection established | +| TLS handshake | `ssl.wrap_socket()` — full handshake (HTTPS only) | +| Server (TTFB) | `sendall()` return → first `recv()` byte | +| Transfer | First byte → EOF — body download time | +| Total | DNS start → body EOF — wall-clock end-to-end | + +## Flags / arguments + +``` +python -m latprobe [flags] [url ...] +``` + +| Flag | Default | Description | +|------|---------|-------------| +| `url …` (positional) | required | One or more URLs to probe | +| `-n N`, `--count N` | `1` | Number of requests per URL | +| `-c N`, `--concurrency N` | `0` (auto) | Max parallel URLs; `0` = `min(len(urls), 8)` | +| `--timeout DURATION` | `10s` | Per-request timeout; supports `ms`, `s`, `m`, or bare seconds | +| `--fail` | off | Exit non-zero when any HTTP status ≥ 400 | +| `--json` | off | Output as JSON array instead of text | +| `-v`, `--verbose` | off | Show resolved IP, TLS version/cipher, certificate details, and response headers | +| `-h`, `--help` | — | Show help and exit 0 | + +**Exit codes:** + +| Code | Meaning | +|------|---------| +| 0 | All probes succeeded | +| 1 | Usage / argument error | +| 2 | DNS failure | +| 3 | TCP connect failure | +| 4 | Timeout | +| 5 | TLS error | +| 6 | HTTP status ≥ 400 (`--fail` only) | + +The highest exit code across all URLs is used as the process exit. + +## Examples + +### Single URL +```sh +python -m latprobe https://example.com +``` +``` +https://example.com (200) + DNS lookup : 18.87 ms + TCP connect : 9.76 ms + TLS handshake : 14.71 ms + Server (TTFB) : 67.07 ms + Transfer : 0.14 ms + ───────────────────────────── + Total : 118.39 ms +``` + +### Sampling (`-n`) — shows min/avg/max table +```sh +python -m latprobe -n 5 https://example.com +``` +``` +https://example.com (200, 5 samples) + min avg max + DNS lookup : 1.80 ms 3.14 ms 5.00 ms + TCP connect : 9.50 ms 10.25 ms 11.40 ms + TLS handshake : 13.80 ms 15.20 ms 18.90 ms + Server (TTFB) : 62.00 ms 66.50 ms 71.30 ms + Transfer : 0.10 ms 0.25 ms 0.40 ms + ───────────────────────────────────────────────── + Total : 97.10 ms 101.20 ms 109.80 ms +``` + +### Multiple URLs (probed in parallel) +```sh +python -m latprobe https://example.com https://www.google.com +``` + +Output for each URL is separated by a blank line. Exit code = worst across all. + +### JSON output +```sh +python -m latprobe --json -n 3 https://example.com +``` +```json +[ + { + "url": "https://example.com", + "status": 200, + "succeeded": 3, + "failed": 0, + "phases": { + "dns": {"min_ms": 1.80, "avg_ms": 3.14, "max_ms": 5.00}, + "connect": {"min_ms": 9.50, "avg_ms": 10.25, "max_ms": 11.40}, + "tls": {"min_ms": 13.80, "avg_ms": 15.20, "max_ms": 18.90}, + "ttfb": {"min_ms": 62.00, "avg_ms": 66.50, "max_ms": 71.30}, + "transfer": {"min_ms": 0.10, "avg_ms": 0.25, "max_ms": 0.40}, + "total": {"min_ms": 97.10, "avg_ms": 101.20, "max_ms": 109.80} + } + } +] +``` + +`phases` is omitted when all samples failed. `tls` key is omitted for `http://` +URLs. `errors` is included (and `phases` omitted) when some samples fail. + +### Verbose mode (`-v` / `--verbose`) + +Shows the resolved IP address, TLS version/cipher/bits, certificate details +(CN, expiry, issuer), and useful response headers. Appended to the standard +timing block. + +```sh +python -m latprobe --verbose https://example.com +``` +``` +https://example.com (200) + DNS lookup : 16.47 ms + TCP connect : 9.76 ms + TLS handshake : 13.09 ms + Server (TTFB) : 60.31 ms + Transfer : 0.20 ms + ───────────────────────────── + Total : 112.76 ms + IP : 104.20.23.154 + TLS : TLSv1.3 TLS_AES_256_GCM_SHA384 256 bit + Cert : CN=example.com valid until 2026-08-29 SSL Corporation + Server : cloudflare + Content-Type : text/html +``` + +For **TLS failures**, the verbose block still shows the IP (DNS + TCP +succeeded) so you can tell which server you actually reached: + +```sh +python -m latprobe --verbose https://expired.badssl.com/ +``` +``` +https://expired.badssl.com/ (FAILED) + DNS lookup : 30.98 ms + TCP connect : 126.58 ms + TLS handshake : 293.58 ms + ───────────────────────────── + Total : 463.37 ms + ✗ tls: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: certificate has expired + IP : 104.154.89.105 +``` + +For **DNS failures**, the verbose block is empty (IP unknown), so it is not +printed. + +Combined with `--json`, verbose detail appears in a `"verbose"` object: + +```sh +python -m latprobe --verbose --json https://example.com +``` +```json +[ + { + "url": "https://example.com", + "status": 200, + "succeeded": 1, + "failed": 0, + "phases": { ... }, + "verbose": { + "ip": "104.20.23.154", + "tls_version": "TLSv1.3", + "tls_cipher": "TLS_AES_256_GCM_SHA384", + "tls_bits": 256, + "cert": { + "cn": "example.com", + "sans": ["example.com", "*.example.com"], + "expiry": "2026-08-29", + "issuer_cn": "SSL Corporation", + "verified": true + }, + "headers": { + "Content-Type": "text/html", + "Server": "cloudflare", + ... + } + } + } +] +``` + +The `"verbose"` key is omitted when `--verbose` is not set. The `"cert"` key +is omitted for `http://` URLs and when TLS fails (Python's stdlib does not +expose parsed cert data from a failed handshake). `"headers"` contains **all** +parsed response headers (the text block shows a curated priority list). + +### DNS failure +```sh +python -m latprobe http://no.such.host.invalid +echo "exit: $?" +``` +``` +http://no.such.host.invalid (FAILED) + Total : 0.65 ms + ✗ dns: [Errno 8] nodename nor servname provided, or not known +exit: 2 +``` + +### `--fail` flag (exit non-zero on HTTP 4xx/5xx) +```sh +python -m latprobe --fail https://www.google.com/this-page-does-not-exist-at-all +echo "exit: $?" +``` +``` +https://www.google.com/this-page-does-not-exist-at-all (404 ✗) + DNS lookup : 3.10 ms + TCP connect : 9.60 ms + TLS handshake : 14.20 ms + Server (TTFB) : 118.50 ms + Transfer : 0.12 ms + ───────────────────────────── + Total : 145.50 ms +exit: 6 +``` + +### Mixed — some URLs succeed, some fail +```sh +python -m latprobe https://example.com http://no.such.host.invalid +echo "exit: $?" +``` +``` +https://example.com (200) + ... + Total : 118.39 ms + +http://no.such.host.invalid (FAILED) + Total : 0.65 ms + ✗ dns: ... +exit: 2 +``` + +### Timeout +```sh +python -m latprobe --timeout 500ms http://10.255.255.1/ +echo "exit: $?" +``` +``` +http://10.255.255.1/ (FAILED) + DNS lookup : 0.20 ms + TCP connect : 500.18 ms + ───────────────────────────── + Total : 500.40 ms + ✗ timeout: timed out +exit: 4 +``` + +## Makefile targets + +```sh +make py-run ARGS="-n 3 https://example.com" # run the package +make py-test # run test suite +make py-check # alias for py-test +``` + +## Limitations + +- **DNS timeout is OS-controlled.** Python's `socket.getaddrinfo` does not + accept a timeout parameter. The `--timeout` flag applies to TCP connect and + subsequent phases. DNS failures from an unreachable or NXDOMAIN host still + happen quickly in practice. +- **HTTP 1.1 + `Connection: close` only.** No keep-alive, HTTP/2, auth, custom + headers, or redirect following. HTTP 3xx is shown with its raw status code. +- **Body fully drained.** Transfer time is real download time; large bodies + affect the Transfer and Total phases. + +## Comparison with `simple.py` and `phases.py` + +| | `simple.py` | `phases.py` | `latprobe/` | +|--|-------------|-------------|-------------| +| Phases | total only | all phases | all phases | +| Sampling | no | no | `-n` flag | +| Concurrency | no | no | `-c` flag | +| JSON | no | no | `--json` flag | +| `--fail` | implicit (urlopen raises on 4xx) | no | `--fail` flag | +| Exit codes | 0 or 1 | 0 or 1 | 0–6 per failure class | +| Config file | yes | yes | no (URL args only) | diff --git a/python/configs/usage-latprobe.md b/python/configs/usage-latprobe.md new file mode 100644 index 0000000..3999193 --- /dev/null +++ b/python/configs/usage-latprobe.md @@ -0,0 +1,312 @@ +# `latprobe` — Runnable Usage Reference + +All commands run from the repository root. Timings will differ on your +machine and network; the output structure is stable. + +--- + +## Basic — single URL + +```sh +make py-run ARGS="https://example.com" +# or directly: +python3.14 -m latprobe https://example.com +``` + +``` +https://example.com (200) + DNS lookup : 16.47 ms + TCP connect : 9.76 ms + TLS handshake : 13.09 ms + Server (TTFB) : 60.31 ms + Transfer : 0.20 ms + ───────────────────────────── + Total : 112.76 ms +``` + +--- + +## Verbose mode — IP, TLS, certificate, headers + +```sh +python3.14 -m latprobe --verbose https://example.com +``` + +``` +https://example.com (200) + DNS lookup : 16.47 ms + TCP connect : 9.76 ms + TLS handshake : 13.09 ms + Server (TTFB) : 60.31 ms + Transfer : 0.20 ms + ───────────────────────────── + Total : 112.76 ms + IP : 104.20.23.154 + TLS : TLSv1.3 TLS_AES_256_GCM_SHA384 256 bit + Cert : CN=example.com valid until 2026-08-29 SSL Corporation + Server : cloudflare + Content-Type : text/html +``` + +The verbose block shows: +- **IP** — first resolved address (useful when DNS round-robins across IPs) +- **TLS** — protocol version, cipher suite, and key bits +- **Cert** — common name, expiry date (prefixed `EXPIRED` if past), and issuer +- Response headers from the priority list: `Location`, `Server`, `Content-Type`, + `X-Cache`, `CF-Cache-Status`, `Cache-Control`, `Via`, `X-Powered-By` + +--- + +## Verbose — plain HTTP (no TLS block) + +```sh +python3.14 -m latprobe --verbose http://example.com +``` + +``` +http://example.com (301) + DNS lookup : 18.22 ms + TCP connect : 10.01 ms + Server (TTFB) : 65.40 ms + Transfer : 0.08 ms + ───────────────────────────── + Total : 96.10 ms + IP : 104.20.23.154 + Location : https://www.example.com/ + Server : cloudflare + Content-Type : text/html +``` + +No `TLS` or `Cert` rows for `http://` URLs. + +--- + +## Verbose — TLS failure (certificate expired) + +```sh +python3.14 -m latprobe --verbose https://expired.badssl.com/ +echo "exit: $?" +``` + +``` +https://expired.badssl.com/ (FAILED) + DNS lookup : 30.98 ms + TCP connect : 126.58 ms + TLS handshake : 293.58 ms + ───────────────────────────── + Total : 463.37 ms + ✗ tls: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: certificate has expired (_ssl.c:1082) + IP : 104.154.89.105 +exit: 5 +``` + +The IP is shown even on TLS failure (DNS and TCP both succeeded). The error +message identifies the cause; certificate details are unavailable because +Python's stdlib does not expose the rejected cert. + +--- + +## Verbose — DNS failure (no IP to show) + +```sh +python3.14 -m latprobe --verbose http://no.such.host.invalid +echo "exit: $?" +``` + +``` +http://no.such.host.invalid (FAILED) + Total : 2.65 ms + ✗ dns: [Errno 8] nodename nor servname provided, or not known +exit: 2 +``` + +The verbose block is empty (IP never resolved), so it is suppressed entirely. + +--- + +## Sampling (`-n`) — min / avg / max table + +```sh +python3.14 -m latprobe -n 3 https://example.com +``` + +``` +https://example.com (200, 3 samples) + min avg max + DNS lookup : 1.15 ms 2.09 ms 3.45 ms + TCP connect : 8.76 ms 9.50 ms 10.21 ms + TLS handshake : 14.48 ms 16.63 ms 20.56 ms + Server (TTFB) : 65.44 ms 68.22 ms 70.58 ms + Transfer : 0.15 ms 0.28 ms 0.42 ms + ───────────────────────────────────────────────── + Total : 101.64 ms 106.31 ms 115.47 ms +``` + +With `--verbose`, the verbose block is appended below the table using the last +successful sample's detail: + +```sh +python3.14 -m latprobe --verbose -n 3 https://example.com +``` + +``` +https://example.com (200, 3 samples) + min avg max + ... + Total : 101.64 ms 106.31 ms 115.47 ms + IP : 104.20.23.154 + TLS : TLSv1.3 TLS_AES_256_GCM_SHA384 256 bit + Cert : CN=example.com valid until 2026-08-29 SSL Corporation + Server : cloudflare + Content-Type : text/html +``` + +--- + +## Multiple URLs (probed in parallel) + +```sh +python3.14 -m latprobe https://example.com https://www.iana.org +``` + +Output for each URL is separated by a blank line. Exit code = worst across all. + +--- + +## `--fail` flag — exit non-zero on HTTP 4xx/5xx + +```sh +python3.14 -m latprobe --fail https://www.google.com/this-page-does-not-exist +echo "exit: $?" +``` + +``` +https://www.google.com/this-page-does-not-exist (404 ✗) + ... + Total : 145.50 ms +exit: 6 +``` + +Without `--fail`, HTTP 4xx/5xx responses are shown normally and the exit code +is `0`. + +--- + +## JSON output + +```sh +python3.14 -m latprobe --json https://example.com | python3.14 -m json.tool +``` + +```json +[ + { + "url": "https://example.com", + "status": 200, + "succeeded": 1, + "failed": 0, + "phases": { + "dns": {"min_ms": 16.47, "avg_ms": 16.47, "max_ms": 16.47}, + "connect": {"min_ms": 9.76, "avg_ms": 9.76, "max_ms": 9.76}, + "tls": {"min_ms": 13.09, "avg_ms": 13.09, "max_ms": 13.09}, + "ttfb": {"min_ms": 60.31, "avg_ms": 60.31, "max_ms": 60.31}, + "transfer": {"min_ms": 0.20, "avg_ms": 0.20, "max_ms": 0.20}, + "total": {"min_ms": 112.76, "avg_ms": 112.76, "max_ms": 112.76} + } + } +] +``` + +`phases` is omitted when all samples failed (only `errors` is present). +`tls` key is omitted for `http://` URLs. + +--- + +## JSON + verbose + +```sh +python3.14 -m latprobe --verbose --json https://example.com +``` + +```json +[ + { + "url": "https://example.com", + "status": 200, + "succeeded": 1, + "failed": 0, + "phases": { "..." }, + "verbose": { + "ip": "104.20.23.154", + "tls_version": "TLSv1.3", + "tls_cipher": "TLS_AES_256_GCM_SHA384", + "tls_bits": 256, + "cert": { + "cn": "example.com", + "sans": ["example.com", "*.example.com"], + "expiry": "2026-08-29", + "issuer_cn": "SSL Corporation", + "verified": true + }, + "headers": { + "Content-Type": "text/html", + "Server": "cloudflare", + "cf-cache-status": "HIT" + } + } + } +] +``` + +`"verbose"` is omitted when `--verbose` is not set. +`"cert"` is omitted for `http://` URLs and when TLS fails. +`"headers"` contains **all** parsed response headers (the text block shows +only the priority list). + +--- + +## Timeout + +```sh +python3.14 -m latprobe --timeout 500ms http://10.255.255.1/ +echo "exit: $?" +``` + +``` +http://10.255.255.1/ (FAILED) + DNS lookup : 0.20 ms + TCP connect : 500.18 ms + ───────────────────────────── + Total : 500.40 ms + ✗ timeout: timed out +exit: 4 +``` + +`--timeout` accepts `ms`, `s`, `m` suffixes or bare seconds. + +--- + +## Exit codes + +| Code | Meaning | +|------|---------| +| 0 | All probes succeeded (or HTTP 4xx without `--fail`) | +| 1 | Usage / argument error | +| 2 | DNS failure | +| 3 | TCP connect failure | +| 4 | Timeout | +| 5 | TLS error | +| 6 | HTTP status ≥ 400 with `--fail` | + +The **highest** exit code across all URLs is returned as the process exit. + +--- + +## Makefile shortcuts + +```sh +make py-run ARGS="--verbose https://example.com" # run latprobe +make py-test # hermetic tests only +make py-test-integration # live internet tests (~30 s) +make py-check # alias for py-test +``` diff --git a/python/latprobe/__init__.py b/python/latprobe/__init__.py new file mode 100644 index 0000000..91c3d8b --- /dev/null +++ b/python/latprobe/__init__.py @@ -0,0 +1 @@ +"""latprobe — measure per-phase HTTP request latency.""" diff --git a/python/latprobe/__main__.py b/python/latprobe/__main__.py new file mode 100644 index 0000000..babde6b --- /dev/null +++ b/python/latprobe/__main__.py @@ -0,0 +1,5 @@ +import sys + +from .cli import run + +sys.exit(run(sys.argv[1:], sys.stdout, sys.stderr)) diff --git a/python/latprobe/aggregate.py b/python/latprobe/aggregate.py new file mode 100644 index 0000000..d78a3cb --- /dev/null +++ b/python/latprobe/aggregate.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + +from .probe import Result + + +@dataclass +class PhaseStats: + min_ms: float = 0.0 + avg_ms: float = 0.0 + max_ms: float = 0.0 + present: bool = False + + +@dataclass +class Aggregate: + url: str = "" + count: int = 0 + status_code: int = 0 + dns: PhaseStats = field(default_factory=PhaseStats) + connect: PhaseStats = field(default_factory=PhaseStats) + tls: PhaseStats = field(default_factory=PhaseStats) + ttfb: PhaseStats = field(default_factory=PhaseStats) + transfer: PhaseStats = field(default_factory=PhaseStats) + total: PhaseStats = field(default_factory=PhaseStats) + + +def summarize(results: list[Result]) -> Aggregate: + """Compute per-phase min/avg/max over a list of succeeded Results.""" + if not results: + return Aggregate() + a = Aggregate(url=results[0].url, count=len(results)) + a.status_code = results[-1].status_code + + def _stats(attr: str) -> PhaseStats: + vals = [getattr(r, attr).ms for r in results if getattr(r, attr).present] + if not vals: + return PhaseStats() + return PhaseStats( + min_ms=min(vals), + avg_ms=sum(vals) / len(vals), + max_ms=max(vals), + present=True, + ) + + a.dns = _stats("dns") + a.connect = _stats("connect") + a.tls = _stats("tls") + a.ttfb = _stats("ttfb") + a.transfer = _stats("transfer") + a.total = _stats("total") + return a diff --git a/python/latprobe/cli.py b/python/latprobe/cli.py new file mode 100644 index 0000000..e0bb3bd --- /dev/null +++ b/python/latprobe/cli.py @@ -0,0 +1,441 @@ +from __future__ import annotations + +import argparse +import concurrent.futures +import datetime +import json +import sys +from typing import IO + +from .aggregate import Aggregate, PhaseStats, summarize +from .duration import parse_duration +from .probe import Options, Result, VerboseDetail, measure + +# ── exit codes (mirrors Go) ─────────────────────────────────────────────────── + +EXIT_OK = 0 +EXIT_USAGE = 1 +EXIT_DNS = 2 +EXIT_CONNECT = 3 +EXIT_TIMEOUT = 4 +EXIT_TLS = 5 +EXIT_HTTP = 6 + +_PHASE_EXIT: dict[str, int] = { + "dns": EXIT_DNS, + "timeout": EXIT_TIMEOUT, + "tls": EXIT_TLS, +} + + +def _phase_code(fail_phase: str) -> int: + return _PHASE_EXIT.get(fail_phase, EXIT_CONNECT) + + +# ── display constants ───────────────────────────────────────────────────────── + +_SINGLE_SEP = " " + "─" * 29 +_AGG_SEP = " " + "─" * 49 + +_PHASE_LABELS = [ + ("dns", "DNS lookup "), + ("connect", "TCP connect "), + ("tls", "TLS handshake "), + ("ttfb", "Server (TTFB) "), + ("transfer", "Transfer "), +] + +# Response headers shown in text verbose block, in priority order. +# In JSON verbose mode all parsed headers are included. +_VERBOSE_HEADERS = [ + "Location", + "Server", + "Content-Type", + "X-Cache", + "CF-Cache-Status", + "Cache-Control", + "Via", + "X-Powered-By", +] + +# ── argparse with injectable streams ───────────────────────────────────────── + + +class _ArgExit(Exception): + def __init__(self, code: int) -> None: + self.code = code + + +class _Parser(argparse.ArgumentParser): + """ArgumentParser that writes to injected streams and raises instead of exiting.""" + + def __init__(self, *args, out: IO, err: IO, **kwargs) -> None: + super().__init__(*args, **kwargs) + self._out = out + self._err = err + + def _print_message(self, message: str, file=None) -> None: + if message: + (file if file is not None else self._out).write(message) + + def print_help(self, file=None) -> None: + self._print_message(self.format_help(), self._out) + + def print_usage(self, file=None) -> None: + self._print_message(self.format_usage(), self._err) + + def error(self, message: str) -> None: + self.print_usage() + self._err.write(f"{self.prog}: error: {message}\n") + raise _ArgExit(EXIT_USAGE) + + def exit(self, status: int = 0, message: str | None = None) -> None: + if message: + self._err.write(message) + raise _ArgExit(int(status) if status else EXIT_OK) + + +# ── sampling ────────────────────────────────────────────────────────────────── + + +def _run_samples( + url: str, count: int, opts: Options +) -> tuple[list[Result], list[Result]]: + succeeded, failed = [], [] + for _ in range(count): + r = measure(url, opts) + (failed if r.err else succeeded).append(r) + return succeeded, failed + + +# ── verbose rendering ───────────────────────────────────────────────────────── + + +def _cert_expired(expiry: str) -> bool: + try: + return datetime.date.fromisoformat(expiry) < datetime.date.today() + except (ValueError, TypeError): + return False + + +def _print_verbose_block(detail: VerboseDetail, out: IO) -> None: + """Write the verbose detail block (IP, TLS, cert, headers) after timing rows.""" + def _row(label: str, value: str) -> None: + out.write(f" {label:<14} : {value}\n") + + if detail.resolved_ip: + _row("IP", detail.resolved_ip) + + if detail.tls_version: + parts = [detail.tls_version] + if detail.tls_cipher: + parts.append(detail.tls_cipher) + if detail.tls_bits: + parts.append(f"{detail.tls_bits} bit") + _row("TLS", " ".join(parts)) + + if detail.cert: + c = detail.cert + label = "Cert (unvrf.)" if not c.verified else "Cert" + cert_parts = [] + if c.cn: + cert_parts.append(f"CN={c.cn}") + if c.expiry: + tag = "EXPIRED" if _cert_expired(c.expiry) else "valid until" + cert_parts.append(f"{tag} {c.expiry}") + if c.issuer_cn: + cert_parts.append(c.issuer_cn) + _row(label, " ".join(cert_parts)) + + for name in _VERBOSE_HEADERS: + value = detail.headers.get(name) + if value: + _row(name, value) + + +# ── text rendering ──────────────────────────────────────────────────────────── + + +def _status_str(status_code: int, fail_flag: bool) -> str: + s = str(status_code) + if fail_flag and status_code >= 400: + s += " ✗" + return s + + +def _print_single(r: Result, fail_flag: bool, out: IO) -> None: + status = _status_str(r.status_code, fail_flag) + out.write(f"{r.url} ({status})\n") + for attr, label in _PHASE_LABELS: + ph = getattr(r, attr) + if ph.present: + out.write(f" {label} : {ph.ms:8.2f} ms\n") + out.write(_SINGLE_SEP + "\n") + if r.total.present: + out.write(f" {'Total '} : {r.total.ms:8.2f} ms\n") + if r.err is not None: + out.write(f" ✗ {r.fail_phase}: {r.err}\n") + if r.detail is not None: + _print_verbose_block(r.detail, out) + + +def _print_aggregate( + agg: Aggregate, + failed: list[Result], + fail_flag: bool, + out: IO, + detail: VerboseDetail | None = None, +) -> None: + status = _status_str(agg.status_code, fail_flag) + header = f"{agg.url} ({status}, {agg.count} samples" + if failed: + header += f", {len(failed)} failed" + out.write(header + ")\n") + + if agg.total.present: + out.write(f" {'':14} {'min':>9} {'avg':>9} {'max':>9}\n") + for attr, label in _PHASE_LABELS: + ps: PhaseStats = getattr(agg, attr) + if ps.present: + out.write( + f" {label} : {ps.min_ms:6.2f} ms" + f" {ps.avg_ms:6.2f} ms" + f" {ps.max_ms:6.2f} ms\n" + ) + out.write(_AGG_SEP + "\n") + out.write( + f" {'Total '} : {agg.total.min_ms:6.2f} ms" + f" {agg.total.avg_ms:6.2f} ms" + f" {agg.total.max_ms:6.2f} ms\n" + ) + _print_failure_summary(failed, out) + if detail is not None: + _print_verbose_block(detail, out) + + +def _print_all_failed( + url: str, failed: list[Result], total_count: int, out: IO +) -> None: + header = f"{url} (FAILED" + if total_count > 1: + header += f", 0/{total_count} succeeded" + out.write(header + ")\n") + + last = failed[-1] + any_phase = False + for attr, label in _PHASE_LABELS: + ph = getattr(last, attr) + if ph.present: + out.write(f" {label} : {ph.ms:8.2f} ms\n") + any_phase = True + if last.total.present: + if any_phase: + out.write(_SINGLE_SEP + "\n") + out.write(f" {'Total '} : {last.total.ms:8.2f} ms\n") + _print_failure_summary(failed, out) + if last.detail is not None: + _print_verbose_block(last.detail, out) + + +def _print_failure_summary(failed: list[Result], out: IO) -> None: + if not failed: + return + counts: dict[tuple[str, str], int] = {} + order: list[tuple[str, str]] = [] + for r in failed: + key = (r.fail_phase, str(r.err)) + if key not in counts: + order.append(key) + counts[key] = 0 + counts[key] += 1 + for phase, msg in order: + n = counts[(phase, msg)] + if n == 1: + out.write(f" ✗ {phase}: {msg}\n") + else: + out.write(f" ✗ {n} × {phase}: {msg}\n") + + +def _print_url( + url: str, + succeeded: list[Result], + failed: list[Result], + total_count: int, + fail_flag: bool, + out: IO, +) -> None: + n_ok = len(succeeded) + n_fail = len(failed) + last_detail = succeeded[-1].detail if succeeded else None + + if n_fail == 0 and total_count == 1: + _print_single(succeeded[0], fail_flag, out) + elif n_fail == 0: + _print_aggregate(summarize(succeeded), [], fail_flag, out, last_detail) + elif n_ok == 0: + _print_all_failed(url, failed, total_count, out) + else: + _print_aggregate(summarize(succeeded), failed, fail_flag, out, last_detail) + + +# ── JSON rendering ──────────────────────────────────────────────────────────── + + +def _build_json_entry( + url: str, + succeeded: list[Result], + failed: list[Result], + detail: VerboseDetail | None = None, +) -> dict: + entry: dict = { + "url": url, + "status": 0, + "succeeded": len(succeeded), + "failed": len(failed), + } + + if succeeded: + agg = summarize(succeeded) + entry["status"] = agg.status_code + phases: dict = {} + for attr in ("dns", "connect", "tls", "ttfb", "transfer", "total"): + ps: PhaseStats = getattr(agg, attr) + if ps.present: + phases[attr] = { + "min_ms": ps.min_ms, + "avg_ms": ps.avg_ms, + "max_ms": ps.max_ms, + } + if phases: + entry["phases"] = phases + + if failed: + counts: dict[tuple[str, str], int] = {} + order: list[tuple[str, str]] = [] + for r in failed: + key = (r.fail_phase, str(r.err)) + if key not in counts: + order.append(key) + counts[key] = 0 + counts[key] += 1 + entry["errors"] = [ + {"phase": ph, "count": counts[(ph, msg)], "message": msg} + for ph, msg in order + ] + + if detail is not None: + v: dict = {} + if detail.resolved_ip: + v["ip"] = detail.resolved_ip + if detail.tls_version: + v["tls_version"] = detail.tls_version + v["tls_cipher"] = detail.tls_cipher + v["tls_bits"] = detail.tls_bits + if detail.cert: + c = detail.cert + v["cert"] = { + "cn": c.cn, + "sans": c.sans, + "expiry": c.expiry, + "issuer_cn": c.issuer_cn, + "verified": c.verified, + } + if detail.headers: + v["headers"] = dict(detail.headers) + if v: + entry["verbose"] = v + + return entry + + +# ── entry point ─────────────────────────────────────────────────────────────── + + +def run(args: list[str], stdout: IO, stderr: IO) -> int: + parser = _Parser( + prog="latprobe", + description="measure per-phase HTTP request latency", + out=stdout, + err=stderr, + ) + parser.add_argument("urls", nargs="+", metavar="url") + parser.add_argument( + "-n", "--count", type=int, default=1, metavar="N", + help="number of requests per URL (default: 1)", + ) + parser.add_argument( + "-c", "--concurrency", type=int, default=0, metavar="N", + help="max parallel URLs, 0=auto (default: 0)", + ) + parser.add_argument( + "--timeout", default="10s", metavar="DURATION", + help="per-request timeout, e.g. 10s, 500ms (default: 10s)", + ) + parser.add_argument( + "--fail", action="store_true", + help="exit non-zero on HTTP status >= 400", + ) + parser.add_argument( + "--json", action="store_true", dest="json_out", + help="output results as JSON instead of text", + ) + parser.add_argument( + "-v", "--verbose", action="store_true", + help="show resolved IP, TLS version/cipher, certificate, and response headers", + ) + + try: + ns = parser.parse_args(args) + except _ArgExit as exc: + return exc.code + + try: + timeout_secs = parse_duration(ns.timeout) + except ValueError: + stderr.write(f"latprobe: error: invalid timeout: {ns.timeout!r}\n") + return EXIT_USAGE + + opts = Options(timeout=timeout_secs, verbose=ns.verbose) + urls = ns.urls + count = ns.count + + workers = ns.concurrency + if workers <= 0: + workers = min(len(urls), 8) + workers = max(1, min(workers, len(urls))) + + def _probe(url: str) -> tuple[list[Result], list[Result]]: + return _run_samples(url, count, opts) + + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex: + all_results = list(ex.map(_probe, urls)) + + worst = EXIT_OK + json_items = [] + + for i, (url, (succeeded, failed)) in enumerate(zip(urls, all_results)): + for r in failed: + c = _phase_code(r.fail_phase) + if c > worst: + worst = c + if ns.fail: + for r in succeeded: + if r.status_code >= 400: + worst = max(worst, EXIT_HTTP) + + last_detail = succeeded[-1].detail if succeeded else ( + failed[-1].detail if failed else None + ) + + if ns.json_out: + json_items.append(_build_json_entry(url, succeeded, failed, last_detail)) + continue + + if i > 0: + stdout.write("\n") + _print_url(url, succeeded, failed, count, ns.fail, stdout) + + if ns.json_out: + stdout.write(json.dumps(json_items, indent=2) + "\n") + + return worst diff --git a/python/latprobe/duration.py b/python/latprobe/duration.py new file mode 100644 index 0000000..dc8d020 --- /dev/null +++ b/python/latprobe/duration.py @@ -0,0 +1,14 @@ +def parse_duration(s: str) -> float: + """Parse a duration string to seconds. + + Suffixes: ms (milliseconds), s (seconds), m (minutes). + A bare number is treated as seconds. + """ + s = s.strip() + if s.endswith("ms"): + return float(s[:-2]) / 1000 + if s.endswith("s"): + return float(s[:-1]) + if s.endswith("m"): + return float(s[:-1]) * 60 + return float(s) diff --git a/python/latprobe/probe.py b/python/latprobe/probe.py new file mode 100644 index 0000000..e7e5748 --- /dev/null +++ b/python/latprobe/probe.py @@ -0,0 +1,306 @@ +from __future__ import annotations + +import datetime +import socket +import ssl +import time +import urllib.parse +from dataclasses import dataclass, field + +# ── verbose detail dataclasses ──────────────────────────────────────────────── + + +@dataclass +class CertInfo: + cn: str = "" + sans: list[str] = field(default_factory=list) + expiry: str = "" # ISO date "YYYY-MM-DD" + issuer_cn: str = "" + verified: bool = False # True = TLS handshake passed verification + + +@dataclass +class VerboseDetail: + resolved_ip: str = "" + tls_version: str = "" + tls_cipher: str = "" + tls_bits: int = 0 + cert: CertInfo | None = None + headers: dict[str, str] = field(default_factory=dict) # all parsed response headers + + +# ── core dataclasses ────────────────────────────────────────────────────────── + + +@dataclass +class Options: + timeout: float = 10.0 + verbose: bool = False + + +@dataclass +class Phase: + ms: float = 0.0 + present: bool = False + + +@dataclass +class Result: + url: str + dns: Phase = field(default_factory=Phase) + connect: Phase = field(default_factory=Phase) + tls: Phase = field(default_factory=Phase) + ttfb: Phase = field(default_factory=Phase) + transfer: Phase = field(default_factory=Phase) + total: Phase = field(default_factory=Phase) + status_code: int = 0 + fail_phase: str = "" + err: Exception | None = None + detail: VerboseDetail | None = None # populated only when opts.verbose=True + + +# ── internal helpers ────────────────────────────────────────────────────────── + + +def _p(start: float, end: float) -> Phase: + return Phase(ms=(end - start) * 1000, present=True) + + +def _parse_status(data: bytes) -> int: + eol = data.find(b"\r\n") + if eol == -1: + eol = data.find(b"\n") + if eol == -1: + return 0 + parts = data[:eol].decode("latin-1", errors="replace").split(None, 2) + if len(parts) >= 2: + try: + return int(parts[1]) + except ValueError: + pass + return 0 + + +def _parse_cert_date(s: str) -> str: + """Convert SSL cert date 'Jun 14 00:00:00 2025 GMT' → '2025-06-14'.""" + if not s: + return "" + s = " ".join(s.split()) # collapse double-spaces ("May 5 ..." → "May 5 ...") + for fmt in ("%b %d %H:%M:%S %Y %Z", "%b %d %H:%M:%S %Y"): + try: + return datetime.datetime.strptime(s, fmt).strftime("%Y-%m-%d") + except ValueError: + pass + return "" + + +def _parse_cert(peer: dict, *, verified: bool) -> CertInfo: + """Build CertInfo from the dict returned by SSLSocket.getpeercert().""" + def _attr(rdns, key: str) -> str: + for rdn in rdns: + for k, v in rdn: + if k == key: + return v + return "" + + issuer = peer.get("issuer", ()) + # Prefer organizationName for issuer (more human-readable than CA CN) + issuer_cn = _attr(issuer, "organizationName") or _attr(issuer, "commonName") + + return CertInfo( + cn=_attr(peer.get("subject", ()), "commonName"), + sans=[v for k, v in peer.get("subjectAltName", ()) if k == "DNS"], + expiry=_parse_cert_date(peer.get("notAfter", "")), + issuer_cn=issuer_cn, + verified=verified, + ) + + +def _parse_response_headers(buf: bytes) -> dict[str, str]: + """Parse all HTTP response headers from a raw buffer (up to \\r\\n\\r\\n).""" + header_section = buf.split(b"\r\n\r\n", 1)[0] + lines = header_section.decode("latin-1", errors="replace").splitlines() + headers: dict[str, str] = {} + for line in lines[1:]: # skip status line + if ":" in line: + name, _, value = line.partition(":") + headers[name.strip()] = value.strip() + return headers + + +# ── measure ─────────────────────────────────────────────────────────────────── + + +def measure(raw_url: str, opts: Options | None = None) -> Result: + """Probe raw_url and return a Result with per-phase timings. + + Partial phases are preserved when the request fails mid-flight. + Redirects are not followed; bodies are drained so Transfer timing is real. + When opts.verbose=True, Result.detail is populated with resolved IP, + TLS metadata, certificate info, and response headers. + """ + if opts is None: + opts = Options() + r = Result(url=raw_url) + + if opts.verbose: + r.detail = VerboseDetail() + + parsed = urllib.parse.urlparse(raw_url) + scheme = parsed.scheme.lower() + if scheme not in ("http", "https"): + r.fail_phase = "request" + r.err = ValueError(f"unsupported scheme: {scheme!r}") + return r + + host = parsed.hostname or "" + port = parsed.port or (443 if scheme == "https" else 80) + path = parsed.path or "/" + if parsed.query: + path = path + "?" + parsed.query + use_tls = scheme == "https" + + t_start = time.perf_counter() + + # ── DNS ────────────────────────────────────────────────────────────────── + + t0 = time.perf_counter() + try: + infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except socket.gaierror as exc: + r.fail_phase = "dns" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + r.dns = _p(t0, time.perf_counter()) + + if r.detail is not None: + r.detail.resolved_ip = str(infos[0][4][0]) + + # ── TCP connect ─────────────────────────────────────────────────────────── + + addr = infos[0][4] + family = infos[0][0] + sock = socket.socket(family, socket.SOCK_STREAM) + sock.settimeout(opts.timeout) + t0 = time.perf_counter() + try: + sock.connect(addr) + except socket.timeout as exc: + sock.close() + r.connect = _p(t0, time.perf_counter()) + r.fail_phase = "timeout" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + except OSError as exc: + sock.close() + r.connect = _p(t0, time.perf_counter()) + r.fail_phase = "connect" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + r.connect = _p(t0, time.perf_counter()) + + # ── TLS handshake (HTTPS only) ──────────────────────────────────────────── + + if use_tls: + ctx = ssl.create_default_context() + t0 = time.perf_counter() + try: + sock = ctx.wrap_socket(sock, server_hostname=host) + except socket.timeout as exc: + r.tls = _p(t0, time.perf_counter()) + r.fail_phase = "timeout" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + except (ssl.SSLError, OSError) as exc: + r.tls = _p(t0, time.perf_counter()) + r.fail_phase = "tls" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + r.tls = _p(t0, time.perf_counter()) + + if r.detail is not None: + r.detail.tls_version = sock.version() or "" + cipher_name, _, bits = sock.cipher() + r.detail.tls_cipher = cipher_name or "" + r.detail.tls_bits = bits or 0 + peer = sock.getpeercert() + if peer: + r.detail.cert = _parse_cert(peer, verified=True) + + # ── Send request ────────────────────────────────────────────────────────── + + request = ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {host}\r\n" + f"Connection: close\r\n" + f"User-Agent: latprobe/1.0\r\n" + f"\r\n" + ).encode() + try: + sock.sendall(request) + except OSError as exc: + sock.close() + r.fail_phase = "request" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + t_wrote = time.perf_counter() + + # ── TTFB — accumulate until end of headers (\r\n\r\n) ──────────────────── + # + # t_first_byte is stamped on the first recv() that returns data (unchanged + # semantics vs the old single-recv approach). We keep reading until the + # full header section is in buf so that response headers can be parsed when + # opts.verbose=True. + + try: + buf = b"" + t_first_byte: float | None = None + while b"\r\n\r\n" not in buf: + chunk = sock.recv(4096) + if not chunk: + raise OSError("server closed connection before headers complete") + if t_first_byte is None: + t_first_byte = time.perf_counter() + buf += chunk + except socket.timeout as exc: + sock.close() + r.fail_phase = "timeout" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + except OSError as exc: + sock.close() + r.fail_phase = "transfer" + r.err = exc + r.total = _p(t_start, time.perf_counter()) + return r + + t_first_byte = t_first_byte or time.perf_counter() + r.ttfb = _p(t_wrote, t_first_byte) + r.status_code = _parse_status(buf) + + if r.detail is not None: + r.detail.headers = _parse_response_headers(buf) + + # ── Transfer — drain remaining body ─────────────────────────────────────── + + try: + while True: + chunk = sock.recv(65536) + if not chunk: + break + except OSError: + pass + finally: + sock.close() + + t_end = time.perf_counter() + r.transfer = _p(t_first_byte, t_end) + r.total = _p(t_start, t_end) + return r diff --git a/python/tests/test_cli.py b/python/tests/test_cli.py new file mode 100644 index 0000000..cb22fe4 --- /dev/null +++ b/python/tests/test_cli.py @@ -0,0 +1,386 @@ +import http.server +import io +import json +import socket +import threading +import unittest + +from latprobe.cli import ( + EXIT_CONNECT, + EXIT_DNS, + EXIT_HTTP, + EXIT_OK, + EXIT_TIMEOUT, + EXIT_USAGE, + run, +) + + +class _OKHandler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.end_headers() + self.wfile.write(b"hello latprobe") + + def log_message(self, *args): + pass + + +class _NotFoundHandler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(404) + self.end_headers() + self.wfile.write(b"not found") + + 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: + 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 + + +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 TestCLIUsageErrors(unittest.TestCase): + + def test_no_args_returns_usage(self): + code, out, err = _invoke([]) + self.assertEqual(code, EXIT_USAGE) + + def test_help_returns_ok(self): + code, out, err = _invoke(["-h"]) + self.assertEqual(code, EXIT_OK) + self.assertIn("latprobe", out) + + def test_invalid_timeout_returns_usage(self): + code, out, err = _invoke(["--timeout", "bad", "http://example.com"]) + self.assertEqual(code, EXIT_USAGE) + self.assertIn("timeout", err) + + +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) + + @classmethod + def tearDownClass(cls): + cls.ok_server.shutdown() + cls.nf_server.shutdown() + + def _url(self, port=None): + return f"http://127.0.0.1:{port or self.ok_port}" + + def test_single_url_success(self): + code, out, err = _invoke([self._url()]) + self.assertEqual(code, EXIT_OK) + self.assertIn("200", out) + self.assertIn("Total", out) + self.assertIn("DNS lookup", out) + + def test_output_has_separator(self): + code, out, _ = _invoke([self._url()]) + self.assertIn("─", out) + + def test_count_shows_aggregate(self): + code, out, _ = _invoke(["-n", "3", self._url()]) + self.assertEqual(code, EXIT_OK) + self.assertIn("3 samples", out) + self.assertIn("min", out) + self.assertIn("avg", out) + self.assertIn("max", out) + + def test_multiple_urls_output_separated_by_blank_line(self): + url = self._url() + code, out, _ = _invoke([url, url]) + self.assertEqual(code, EXIT_OK) + self.assertIn("\n\n", out) + + def test_fail_flag_ok_on_200(self): + code, out, _ = _invoke(["--fail", self._url()]) + self.assertEqual(code, EXIT_OK) + self.assertNotIn("✗", 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) + + +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) + self.assertIn("✗", 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) + + def test_timeout(self): + port = _black_hole_port() + code, _, _ = _invoke([f"http://127.0.0.1:{port}", "--timeout", "200ms"]) + self.assertEqual(code, EXIT_TIMEOUT) + + def test_worst_code_across_urls(self): + server, port = _start_server( + type("_H", (http.server.BaseHTTPRequestHandler,), { + "do_GET": lambda self: (self.send_response(200), self.end_headers(), self.wfile.write(b"ok")), + "log_message": lambda *a: None, + }) + ) + try: + code, _, _ = _invoke([ + f"http://127.0.0.1:{port}", # exit 0 + "http://no.such.host.invalid", # exit 2 + ]) + self.assertEqual(code, EXIT_DNS) + finally: + server.shutdown() + + def test_all_failed_single_header(self): + code, out, _ = _invoke(["http://no.such.host.invalid"]) + self.assertIn("(FAILED)", out) + self.assertNotIn("0/1 succeeded", out) + + def test_all_failed_multi_header(self): + code, out, _ = _invoke([ + "-n", "3", + "http://no.such.host.invalid", + ]) + self.assertIn("0/3 succeeded", out) + + def test_mixed_aggregate_shows_samples(self): + port = _free_port() + server, ok_port = _start_server(_OKHandler) + try: + code, out, _ = _invoke([ + "-n", "1", + f"http://127.0.0.1:{ok_port}", + f"http://127.0.0.1:{port}", + ]) + self.assertEqual(code, EXIT_CONNECT) + self.assertIn("200", out) + finally: + server.shutdown() + + +class TestCLIJSON(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 _url(self, port=None): + return f"http://127.0.0.1:{port or self.ok_port}" + + def test_json_success_schema(self): + code, out, _ = _invoke(["--json", self._url()]) + self.assertEqual(code, EXIT_OK) + data = json.loads(out) + self.assertIsInstance(data, list) + self.assertEqual(len(data), 1) + entry = data[0] + self.assertEqual(entry["url"], self._url()) + self.assertEqual(entry["status"], 200) + self.assertEqual(entry["succeeded"], 1) + self.assertEqual(entry["failed"], 0) + self.assertIn("phases", entry) + self.assertIn("total", entry["phases"]) + self.assertNotIn("errors", entry) + + def test_json_phase_fields(self): + code, out, _ = _invoke(["--json", self._url()]) + data = json.loads(out) + total = data[0]["phases"]["total"] + for key in ("min_ms", "avg_ms", "max_ms"): + self.assertIn(key, total) + self.assertGreater(total[key], 0) + + def test_json_dns_failure_schema(self): + code, out, _ = _invoke(["--json", "http://no.such.host.invalid"]) + self.assertEqual(code, EXIT_DNS) + data = json.loads(out) + entry = data[0] + self.assertEqual(entry["succeeded"], 0) + self.assertEqual(entry["failed"], 1) + self.assertNotIn("phases", entry) + self.assertIn("errors", entry) + self.assertEqual(entry["errors"][0]["phase"], "dns") + + def test_json_sampling(self): + code, out, _ = _invoke(["--json", "-n", "3", self._url()]) + self.assertEqual(code, EXIT_OK) + data = json.loads(out) + self.assertEqual(data[0]["succeeded"], 3) + + def test_json_multiple_urls_ordered(self): + code, out, _ = _invoke([ + "--json", + self._url(), + "http://no.such.host.invalid", + ]) + data = json.loads(out) + self.assertEqual(len(data), 2) + self.assertEqual(data[0]["status"], 200) + self.assertEqual(data[1]["succeeded"], 0) + + def test_json_fail_flag_exit_code(self): + code, out, _ = _invoke(["--json", "--fail", self._url(self.nf_port)]) + self.assertEqual(code, EXIT_HTTP) + data = json.loads(out) + self.assertEqual(data[0]["status"], 404) + + def test_json_error_grouping(self): + code, out, _ = _invoke([ + "--json", "-n", "2", + "http://no.such.host.invalid", + ]) + data = json.loads(out) + self.assertEqual(data[0]["errors"][0]["count"], 2) + + +class TestCLIVerbose(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 _url(self, port=None): + return f"http://127.0.0.1:{port or self.ok_port}" + + def test_verbose_shows_ip(self): + code, out, _ = _invoke(["--verbose", self._url()]) + self.assertEqual(code, EXIT_OK) + self.assertIn("IP", out) + self.assertIn("127.0.0.1", out) + + def test_short_flag_v(self): + code, out, _ = _invoke(["-v", self._url()]) + self.assertEqual(code, EXIT_OK) + self.assertIn("127.0.0.1", out) + + def test_verbose_no_tls_for_http(self): + code, out, _ = _invoke(["--verbose", self._url()]) + self.assertNotIn("TLS", out) + self.assertNotIn("Cert", out) + + def test_verbose_shows_headers(self): + code, out, _ = _invoke(["--verbose", self._url()]) + # BaseHTTPServer sends Server and Content-Type + self.assertIn("Server", out) + + def test_verbose_absent_without_flag(self): + code, out, _ = _invoke([self._url()]) + # Without --verbose, there is no IP label row (the URL contains the + # IP but is not followed by a " IP" verbose block line) + self.assertNotIn("\n IP ", out) + + def test_verbose_aggregate_shows_ip(self): + code, out, _ = _invoke(["-v", "-n", "2", self._url()]) + self.assertEqual(code, EXIT_OK) + self.assertIn("127.0.0.1", out) + self.assertIn("2 samples", out) + + def test_verbose_on_connect_fail_shows_ip(self): + port = _free_port() + code, out, _ = _invoke(["--verbose", f"http://127.0.0.1:{port}"]) + self.assertEqual(code, EXIT_CONNECT) + self.assertIn("FAILED", out) + self.assertIn("127.0.0.1", out) + + def test_verbose_dns_fail_no_ip(self): + code, out, _ = _invoke(["--verbose", "http://no.such.host.invalid"]) + self.assertEqual(code, EXIT_DNS) + # IP is unknown for DNS failures — verbose block is empty so no IP row + self.assertNotIn("IP", out) + + def test_verbose_json_includes_ip(self): + code, out, _ = _invoke(["--verbose", "--json", self._url()]) + self.assertEqual(code, EXIT_OK) + data = json.loads(out) + self.assertIn("verbose", data[0]) + self.assertEqual(data[0]["verbose"]["ip"], "127.0.0.1") + + def test_verbose_json_includes_headers(self): + code, out, _ = _invoke(["--verbose", "--json", self._url()]) + data = json.loads(out) + self.assertIn("headers", data[0]["verbose"]) + self.assertIsInstance(data[0]["verbose"]["headers"], dict) + + def test_no_verbose_key_in_json_without_flag(self): + code, out, _ = _invoke(["--json", self._url()]) + data = json.loads(out) + self.assertNotIn("verbose", data[0]) + + def test_verbose_json_no_tls_for_http(self): + code, out, _ = _invoke(["--verbose", "--json", self._url()]) + data = json.loads(out) + self.assertNotIn("tls_version", data[0]["verbose"]) + self.assertNotIn("cert", data[0]["verbose"]) + + def test_verbose_json_connect_fail_has_ip(self): + port = _free_port() + code, out, _ = _invoke(["--verbose", "--json", f"http://127.0.0.1:{port}"]) + data = json.loads(out) + self.assertIn("verbose", data[0]) + self.assertEqual(data[0]["verbose"]["ip"], "127.0.0.1") + + def test_verbose_json_dns_fail_no_verbose_object(self): + code, out, _ = _invoke(["--verbose", "--json", "http://no.such.host.invalid"]) + data = json.loads(out) + # DNS failure: detail has no IP, so the verbose dict is empty → omitted + self.assertNotIn("verbose", data[0]) diff --git a/python/tests/test_integration.py b/python/tests/test_integration.py new file mode 100644 index 0000000..5773d48 --- /dev/null +++ b/python/tests/test_integration.py @@ -0,0 +1,430 @@ +"""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) diff --git a/python/tests/test_probe.py b/python/tests/test_probe.py new file mode 100644 index 0000000..eb0cc48 --- /dev/null +++ b/python/tests/test_probe.py @@ -0,0 +1,208 @@ +import http.server +import socket +import threading +import time +import unittest + +from latprobe.probe import Options, Phase, Result, VerboseDetail, measure + + +class _OKHandler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.end_headers() + self.wfile.write(b"hello latprobe") + + def log_message(self, *args): + pass + + +class _NotFoundHandler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(404) + self.end_headers() + self.wfile.write(b"not found") + + 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 TTFB 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_total_gte_sum_of_phases(self): + r = measure(f"http://127.0.0.1:{self.ok_port}") + phase_sum = r.dns.ms + r.connect.ms + r.ttfb.ms + r.transfer.ms + self.assertGreaterEqual(r.total.ms, phase_sum * 0.9) + + 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_timeout(self): + opts = Options() + r = measure(f"http://127.0.0.1:{self.ok_port}", opts) + self.assertIsNone(r.err) + self.assertEqual(r.status_code, 200) + + +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) + + def test_partial_phases_preserved_on_connect_fail(self): + port = _free_port() + r = measure(f"http://127.0.0.1:{port}") + self.assertTrue(r.dns.present, "dns should be recorded before connect") + self.assertTrue(r.connect.present, "connect duration recorded even on refusal") + self.assertGreater(r.dns.ms, 0) + self.assertGreater(r.connect.ms, 0) + + +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_no_tls_fields_for_http(self): + r = measure(self._url(), Options(verbose=True)) + self.assertEqual(r.detail.tls_version, "") + self.assertEqual(r.detail.tls_cipher, "") + self.assertEqual(r.detail.tls_bits, 0) + 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) + # BaseHTTPServer always sends Content-Type for 200 responses + 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") + self.assertEqual(r.detail.headers, {}) + + def test_detail_none_on_dns_failure_without_verbose(self): + r = measure("http://no.such.host.invalid") + self.assertIsNone(r.detail) + + def test_detail_empty_ip_on_dns_failure(self): + r = measure("http://no.such.host.invalid", Options(verbose=True)) + self.assertIsNotNone(r.detail) + self.assertEqual(r.detail.resolved_ip, "") + + def test_headers_not_populated_on_connect_fail(self): + port = _free_port() + r = measure(f"http://127.0.0.1:{port}", Options(verbose=True)) + self.assertEqual(r.detail.headers, {})