latency: ping summarization and unprivileged backend fan-out

src/netcheck/probes/latency.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+import socket
+import time
+
+from netcheck.models import Capabilities, PingResult
+from netcheck.netinfo import choose_latency_backend
+from netcheck.stats import rtt_stats
+
+
+def summarize_ping(
+    label: str,
+    host: str,
+    resolved_ip: str | None,
+    method: str,
+    samples: list[float | None],
+) -> PingResult:
+    s = rtt_stats(samples)
+    return PingResult(
+        label=label,
+        host=host,
+        resolved_ip=resolved_ip,
+        method=method,
+        sent=s.sent,
+        received=s.received,
+        loss_pct=s.loss_pct,
+        min_ms=s.min_ms,
+        avg_ms=s.avg_ms,
+        max_ms=s.max_ms,
+        mdev_ms=s.mdev_ms,
+        jitter_ms=s.jitter_ms,
+        samples=samples,
+    )
+
+
+async def tcp_connect_rtt(host: str, port: int = 443, timeout: float = 2.0) -> float | None:
+    began = time.perf_counter()
+    try:
+        _reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=timeout)
+    except (OSError, asyncio.TimeoutError):
+        return None
+    writer.close()
+    try:
+        await writer.wait_closed()
+    except OSError:
+        pass
+    return (time.perf_counter() - began) * 1000.0
+
+
+def _resolve(host: str) -> str | None:
+    try:
+        return socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)[0][4][0]
+    except (OSError, IndexError):
+        return None
+
+
+async def _icmp_samples(host: str, count: int, interval: float, timeout: float, backend: str) -> list[float | None]:
+    if backend == "icmp_win":
+        from netcheck.probes.icmp_win import ping_samples_win
+
+        return await asyncio.to_thread(ping_samples_win, host, count, interval, timeout)
+    from icmplib import async_ping
+
+    privileged = backend == "icmp_raw"
+    host_result = await async_ping(
+        host, count=count, interval=interval, timeout=timeout, privileged=privileged
+    )
+    samples: list[float | None] = list(host_result.rtts)
+    samples.extend([None] * (count - len(samples)))
+    return samples
+
+
+async def ping_host(
+    host: str,
+    label: str,
+    count: int,
+    interval: float,
+    timeout: float,
+    backend: str,
+) -> PingResult:
+    resolved = _resolve(host)
+    if backend in ("icmp_win", "icmp_dgram", "icmp_raw"):
+        try:
+            samples = await _icmp_samples(host, count, interval, timeout, backend)
+            return summarize_ping(label, host, resolved, backend, samples)
+        except Exception:
+            backend = "tcp"
+    samples = []
+    for index in range(count):
+        if index:
+            await asyncio.sleep(interval)
+        samples.append(await tcp_connect_rtt(host, timeout=timeout))
+    return summarize_ping(label, host, resolved, "tcp", samples)
+
+
+async def ping_fanout(
+    hosts: list[tuple[str, str]],
+    caps: Capabilities,
+    count: int,
+    interval: float,
+    timeout: float,
+) -> list[PingResult]:
+    backend = choose_latency_backend(caps)
+    return list(
+        await asyncio.gather(
+            *(ping_host(host, label, count, interval, timeout, backend) for label, host in hosts)
+        )
+    )

tests/test_prober_stats.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import pytest
+
+from netcheck.probes.latency import summarize_ping
+
+
+def test_summarize_builds_a_complete_ping_result():
+    result = summarize_ping(
+        label="cloudflare-dns",
+        host="1.1.1.1",
+        resolved_ip="1.1.1.1",
+        method="icmp_dgram",
+        samples=[10.0, 12.0, 14.0, 16.0],
+    )
+    assert result.label == "cloudflare-dns"
+    assert result.host == "1.1.1.1"
+    assert result.resolved_ip == "1.1.1.1"
+    assert result.method == "icmp_dgram"
+    assert result.sent == 4
+    assert result.received == 4
+    assert result.loss_pct == 0.0
+    assert result.min_ms == 10.0
+    assert result.avg_ms == 13.0
+    assert result.max_ms == 16.0
+    assert result.mdev_ms == pytest.approx(2.0)
+    assert result.jitter_ms == pytest.approx(2.0)
+    assert result.samples == [10.0, 12.0, 14.0, 16.0]
+
+
+def test_summarize_keeps_timeout_positions_in_the_sample_list():
+    result = summarize_ping("h", "example.test", None, "tcp", [None, 20.0, None, 24.0])
+    assert result.samples == [None, 20.0, None, 24.0]
+    assert result.sent == 4
+    assert result.received == 2
+    assert result.loss_pct == 50.0
+
+
+def test_summarize_of_an_all_timeout_run_is_not_an_error():
+    result = summarize_ping("h", "example.test", None, "icmp_win", [None, None, None])
+    assert result.received == 0
+    assert result.loss_pct == 100.0
+    assert result.avg_ms is None
+    assert result.jitter_ms is None
+
+
+def test_summarize_of_a_single_sample_reports_zero_jitter():
+    result = summarize_ping("h", "1.1.1.1", "1.1.1.1", "cfL4", [9.0])
+    assert result.min_ms == result.avg_ms == result.max_ms == 9.0
+    assert result.jitter_ms == 0.0
+    assert result.mdev_ms == 0.0
+
+
+def test_summarize_of_zero_probes_is_a_well_formed_empty_result():
+    result = summarize_ping("h", "1.1.1.1", None, "none", [])
+    assert result.sent == 0
+    assert result.received == 0
+    assert result.loss_pct == 0.0
+    assert result.samples == []
+
+
+def test_the_method_tag_is_preserved_verbatim():
+    for method in ("icmp_win", "icmp_dgram", "icmp_raw", "tcp", "cfL4"):
+        assert summarize_ping("h", "x", None, method, [1.0]).method == method