Jan Novak
24ea9c9e71
Adds the complete Python port of the Go latprobe tool plus diagnostic verbose output that the Go version does not yet have. Package (python/latprobe/): - probe.py: raw-socket HTTP timing with CertInfo/VerboseDetail dataclasses; TTFB loop accumulates to \r\n\r\n so headers are parseable without changing t_first_byte semantics; captures resolved IP, TLS version/cipher/bits, verified cert (via getpeercert()), and all response headers when verbose=True - aggregate.py: summarize() → per-phase min/avg/max PhaseStats - cli.py: injectable run(args,stdout,stderr)->int; argparse with injected streams; ThreadPoolExecutor concurrency across URLs; four text-rendering branches; JSON output; worst-exit-code accumulation; -v/--verbose flag appends IP/TLS/cert/header block after every timing table; JSON extended with "verbose" object (omitted when flag absent) - duration.py: parse Go-style duration strings (10s, 500ms, 2m, bare seconds) - Exit codes: 0 ok, 1 usage, 2 dns, 3 connect, 4 timeout, 5 tls, 6 http≥400 Tests: - test_probe.py: 18 hermetic tests (success, 404, DNS/connect/timeout/scheme failures, partial-phase invariants, verbose detail fields) - test_cli.py: 38 hermetic tests (usage errors, single/aggregate/failure text, worst-code, --fail, JSON schema/ordering/grouping, verbose text and JSON) - test_integration.py: 45 tests against live internet services (badssl.com for TLS errors, real cert/IP/header validation in verbose mode) Makefile: PYTHONPATH fix, fnmatch pattern test_[!i]*.py to exclude integration tests from py-test, new py-test-integration target Docs: docs/usage/py-latprobe.md, python/configs/usage-latprobe.md (runnable reference with captured output), plans for both the package and verbose feature Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
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
|