http-latency-prober/docs/py-latprobe-walkthrough.md

latprobe (Python) — code walkthrough

A guided tour of python/latprobe/ for anyone about to edit the code by hand. For CLI usage/flags, see docs/usage/py-latprobe.md.

Layout

python/
├── simple.py           step 1 (standalone) — reachability + total elapsed time
├── phases.py           step 2 (standalone) — per-phase timing, single file
├── sites.txt           sample config for simple.py / phases.py
├── configs/            fixture configs (all-ok, dns-failure, timeout, tls-errors, ...)
├── latprobe/           step 3 — the real package
│   ├── __main__.py
│   ├── probe.py
│   ├── aggregate.py
│   ├── duration.py
│   └── cli.py
└── tests/
    ├── test_probe.py
    ├── test_cli.py
    └── test_integration.py

simple.py and phases.py are earlier milestones, kept as reference (mirrors the Go step-by-step convention: simple → phases → full package). They are not imported by the package.

The package: python/latprobe/

probe.py — the engine

No argparse, no printing. measure(url, opts) -> Result does one HTTP(S) request over raw sockets (not requests/urllib) so it can time each phase itself:

• DNS (socket.getaddrinfo)
• TCP connect
• TLS handshake (ssl.wrap_socket, HTTPS only)
• TTFB (write request → first byte, reading until \r\n\r\n)
• Transfer (drain the rest of the body)

Each phase is timed with time.perf_counter() and wrapped by the _p() helper into a Phase(ms, present).

Every failure path (DNS error, connect timeout, TLS error, etc.) returns early with a partially-filled Result — fail_phase says where it died, err holds the exception, and whatever phases completed before the failure are preserved. This is why the CLI can still print e.g. "DNS: 12ms" even if TCP connect then failed.

Options(timeout, verbose) goes in, Result comes out. When verbose=True, Result.detail (VerboseDetail) is also filled in: resolved IP, TLS version/cipher, parsed certificate (CertInfo), and all response headers.

This is the only file with actual measurement logic — change how something is measured here (e.g. add a redirect-follow phase, redefine what counts as TTFB).

aggregate.py — pure math, no I/O

summarize(results: list[Result]) -> Aggregate takes a list of successful Results (from repeated -n sampling of the same URL) and computes min/avg/max per phase into PhaseStats. Aggregate.status_code uses the last result's status. Small and self-contained — safe to extend (e.g. add p50/p95) without touching anything else.

duration.py

One function: parse_duration("500ms" | "10s" | "2m") -> float seconds, used for --timeout.

cli.py — everything else

Argument parsing, orchestration, and both text/JSON rendering. The file you'll touch most for UX changes.

• _Parser subclasses argparse.ArgumentParser to redirect all output through injected stdout/stderr streams and raise _ArgExit instead of calling sys.exit — this is what makes run() fully testable without subprocess (tests just pass in io.StringIO()).
• Exit codes (EXIT_DNS=2, EXIT_CONNECT=3, etc.) are commented as mirroring the Go version. _phase_code() maps a fail_phase string to the matching code, and run() tracks the worst code across all URLs/samples.
• run(args, stdout, stderr) -> int is the entry point:
  1. parse args → build Options
  2. run _run_samples() per URL concurrently via ThreadPoolExecutor (--concurrency, defaulting to min(len(urls), 8))
  3. for each URL, pick one of four print paths in _print_url (_print_single / _print_aggregate / _print_all_failed), based on success/failure counts and whether -n > 1
  4. or, if --json, build dict entries via _build_json_entry and dump them all at the end.
• Verbose rendering (_print_verbose_block) is shared between the single and aggregate text paths; JSON verbose data is built separately in _build_json_entry.

__main__.py

Trivial shim: sys.exit(run(sys.argv[1:], sys.stdout, sys.stderr)), letting you run python -m latprobe <url>.

Tests (python/tests/)

• test_probe.py — unit tests against measure() directly (against the configs/*.txt failure-mode fixtures: DNS failure, connection refused, TLS errors, timeouts).
• test_cli.py — drives run() with injected io.StringIO streams, checking text/JSON output and exit codes.
• test_integration.py — end-to-end, against the configs/*.txt files (all-ok.txt, mixed.txt, http-errors.txt, etc.).

How it connects, end to end

__main__.py
  → cli.run()
      → parses flags
      → for each URL: probe.measure() (× count, across a thread pool)
      → collects Result objects
      → if count > 1: aggregate.summarize() → Aggregate
      → cli.py's _print_* / _build_json_entry render text or JSON
      → returns the worst exit code seen

Where to make changes

Change you want	File
Timing/measurement behavior	probe.py
min/avg/max or new stats	aggregate.py
Flags, output formatting, exit-code logic	cli.py
Duration string parsing	duration.py

The dataclasses (Result, Phase, Options, VerboseDetail, Aggregate, PhaseStats) are the contracts between these files — adding a field to Result in probe.py typically means threading it through aggregate.py (if it should be averaged) and cli.py (if it should be printed/JSON-encoded).