diff --git a/src/netcheck/probes/latency.py b/src/netcheck/probes/latency.py new file mode 100644 index 0000000..c719e0b --- /dev/null +++ b/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) + ) + ) diff --git a/tests/test_prober_stats.py b/tests/test_prober_stats.py new file mode 100644 index 0000000..2a1bf09 --- /dev/null +++ b/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