From f99de6b7dcd97cdf6011519c7e23a938eb165224 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sat, 8 Aug 2026 22:07:07 +0300 Subject: [PATCH] interpret: threshold bands, latency and path findings --- src/netcheck/interpret.py | 134 ++++++++++++++++++++++++++++++++++++ tests/test_interpret.py | 138 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 272 insertions(+) create mode 100644 src/netcheck/interpret.py create mode 100644 tests/test_interpret.py diff --git a/src/netcheck/interpret.py b/src/netcheck/interpret.py new file mode 100644 index 0000000..53fad16 --- /dev/null +++ b/src/netcheck/interpret.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from typing import Iterable + +from netcheck.config import Band, Thresholds +from netcheck.models import Finding, PingResult, TraceResult + +_SEVERITY_ORDER = {"ok": 0, "info": 1, "warn": 2, "crit": 3} + + +def worst(severities: Iterable[str]) -> str: + best = "ok" + for severity in severities: + if _SEVERITY_ORDER[severity] > _SEVERITY_ORDER[best]: + best = severity + return best + + +def severity_for(value: float | None, band: Band, higher_is_worse: bool = True) -> str: + if value is None: + return "info" + if higher_is_worse: + if value <= band.good: + return "ok" + return "warn" if value <= band.warn else "crit" + if value >= band.good: + return "ok" + return "warn" if value >= band.warn else "crit" + + +def latency_findings(pings: list[PingResult], t: Thresholds) -> list[Finding]: + findings: list[Finding] = [] + for p in pings: + if p.received == 0: + findings.append( + Finding( + id=f"latency.unreachable.{p.label}", + severity="crit", + title=f"{p.host} did not answer", + detail=f"{p.sent} probes sent via {p.method}, none returned.", + metric="received", + value=0, + threshold=1, + advice="If every reference host is unreachable the link is down or filtering the probe method.", + ) + ) + continue + severity = severity_for(p.avg_ms, t.latency_ms) + if severity not in ("ok", "info"): + findings.append( + Finding( + id=f"latency.avg.{p.label}", + severity=severity, + title=f"Latency to {p.host} above target", + detail=f"Average {p.avg_ms} ms over {p.received} probes via {p.method}.", + metric="avg_ms", + value=p.avg_ms, + threshold=t.latency_ms.warn, + advice="Check for a saturated uplink, a distant route, or a congested Wi-Fi link.", + ) + ) + severity = severity_for(p.jitter_ms, t.jitter_ms) + if severity not in ("ok", "info"): + findings.append( + Finding( + id=f"latency.jitter.{p.label}", + severity=severity, + title=f"Jitter to {p.host} above target", + detail=f"Jitter {p.jitter_ms} ms over {p.received} probes via {p.method}.", + metric="jitter_ms", + value=p.jitter_ms, + threshold=t.jitter_ms.warn, + advice="Unstable latency hurts calls and games more than raw latency does.", + ) + ) + severity = severity_for(p.loss_pct, t.loss_pct) + if severity not in ("ok", "info"): + kind = "connection failures" if p.method == "tcp" else "packet loss" + findings.append( + Finding( + id=f"latency.loss.{p.label}", + severity=severity, + title=f"Loss to {p.host}", + detail=f"{p.loss_pct}% {kind} over {p.sent} probes via {p.method}.", + metric="loss_pct", + value=p.loss_pct, + threshold=t.loss_pct.warn, + advice="Sustained loss on every host points at the local link or the first upstream hop.", + ) + ) + return findings + + +def path_findings(trace: TraceResult) -> list[Finding]: + if not trace.hops: + return [ + Finding( + id="path.incomplete", + severity="info", + title="No path data", + detail=f"The traceroute to {trace.target} returned no hops (backend {trace.backend}).", + advice="ICMP may be filtered end to end; try --tcp-trace.", + ) + ] + findings: list[Finding] = [] + for index, hop in enumerate(trace.hops[:-1]): + following = trace.hops[index + 1] + # A single lossy hop with clean hops after it is ICMP rate limiting on that + # router, not a real problem; loss that persists to the next hop is real. + if hop.loss_pct >= 20.0 and following.loss_pct >= 20.0: + findings.append( + Finding( + id="path.loss_jump", + severity="crit" if hop.loss_pct >= 50.0 else "warn", + title="Sustained loss starts mid-path", + detail=f"Loss appears at hop {hop.ttl} ({hop.ip}) at {hop.loss_pct}% and persists downstream.", + metric="loss_pct", + value=hop.loss_pct, + threshold=20.0, + advice="This hop and everything after it share the problem; the hop before it is the last clean one.", + ) + ) + break + if not trace.completed: + findings.append( + Finding( + id="path.incomplete", + severity="info", + title="Path did not reach the target", + detail=f"The trace to {trace.target} stopped at hop {trace.hops[-1].ttl}.", + advice="Many networks drop the final ICMP reply; this alone is not a fault.", + ) + ) + return findings diff --git a/tests/test_interpret.py b/tests/test_interpret.py new file mode 100644 index 0000000..0053323 --- /dev/null +++ b/tests/test_interpret.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import pytest + +from netcheck.config import Band, Thresholds +from netcheck.models import PingResult, TraceHop, TraceResult +from netcheck.interpret import latency_findings, path_findings, severity_for, worst + + +@pytest.fixture() +def thresholds() -> Thresholds: + return Thresholds() + + +def test_severity_bands_are_inclusive_at_the_good_edge(): + band = Band(good=40.0, warn=100.0) + assert severity_for(10.0, band) == "ok" + assert severity_for(40.0, band) == "ok" + assert severity_for(40.1, band) == "warn" + assert severity_for(100.0, band) == "warn" + assert severity_for(100.1, band) == "crit" + + +def test_severity_of_a_missing_value_is_info_not_ok(): + assert severity_for(None, Band(good=40.0, warn=100.0)) == "info" + + +def test_severity_can_be_inverted_for_metrics_where_higher_is_better(): + band = Band(good=50.0, warn=10.0) + assert severity_for(100.0, band, higher_is_worse=False) == "ok" + assert severity_for(50.0, band, higher_is_worse=False) == "ok" + assert severity_for(30.0, band, higher_is_worse=False) == "warn" + assert severity_for(5.0, band, higher_is_worse=False) == "crit" + + +def test_worst_picks_the_highest_severity(): + assert worst(["ok", "info", "warn"]) == "warn" + assert worst(["ok", "crit", "warn"]) == "crit" + assert worst([]) == "ok" + assert worst(["ok", "ok"]) == "ok" + + +def ping(**kw) -> PingResult: + base = dict( + label="cloudflare-dns", + host="1.1.1.1", + resolved_ip="1.1.1.1", + method="icmp_dgram", + sent=20, + received=20, + loss_pct=0.0, + min_ms=10.0, + avg_ms=12.0, + max_ms=15.0, + mdev_ms=1.0, + jitter_ms=1.0, + samples=[], + ) + base.update(kw) + return PingResult(**base) + + +def test_a_healthy_ping_produces_no_findings(thresholds): + assert latency_findings([ping()], thresholds) == [] + + +def test_high_latency_produces_a_warn_finding_with_metric_and_threshold(thresholds): + findings = latency_findings([ping(avg_ms=130.0)], thresholds) + assert len(findings) == 1 + f = findings[0] + assert f.id == "latency.avg.cloudflare-dns" + assert f.severity == "crit" + assert f.metric == "avg_ms" + assert f.value == 130.0 + assert f.threshold == 100.0 + assert f.advice + + +def test_high_jitter_and_loss_each_produce_their_own_finding(thresholds): + findings = latency_findings([ping(jitter_ms=25.0, loss_pct=5.0, received=19)], thresholds) + ids = sorted(f.id for f in findings) + assert ids == ["latency.jitter.cloudflare-dns", "latency.loss.cloudflare-dns"] + + +def test_tcp_measured_loss_is_labelled_as_connection_failures_not_packet_loss(thresholds): + findings = latency_findings([ping(method="tcp", loss_pct=10.0)], thresholds) + loss = [f for f in findings if f.id.startswith("latency.loss")][0] + assert "connection failure" in loss.detail.lower() + assert "packet loss" not in loss.detail.lower() + + +def test_a_fully_dead_host_is_critical_regardless_of_bands(thresholds): + findings = latency_findings( + [ping(received=0, loss_pct=100.0, avg_ms=None, min_ms=None, max_ms=None, jitter_ms=None)], + thresholds, + ) + assert [f.severity for f in findings] == ["crit"] + assert findings[0].id == "latency.unreachable.cloudflare-dns" + + +def test_path_findings_highlight_the_first_sustained_loss_jump(): + trace = TraceResult( + target="1.1.1.1", + resolved_ip="1.1.1.1", + backend="mtr_json", + hops=[ + TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0), + TraceHop(ttl=2, ip="10.64.0.1", loss_pct=0.0, avg_ms=9.0), + TraceHop(ttl=3, ip="198.51.100.7", loss_pct=60.0, avg_ms=40.0), + TraceHop(ttl=4, ip="1.1.1.1", loss_pct=55.0, avg_ms=41.0), + ], + completed=True, + ) + findings = path_findings(trace) + loss = [f for f in findings if f.id == "path.loss_jump"][0] + assert loss.severity == "crit" + assert "hop 3" in loss.detail + assert "198.51.100.7" in loss.detail + + +def test_path_findings_ignore_a_single_hop_that_rate_limits_icmp(): + trace = TraceResult( + target="1.1.1.1", + resolved_ip="1.1.1.1", + backend="mtr_json", + hops=[ + TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0), + TraceHop(ttl=2, ip="10.64.0.1", loss_pct=100.0, avg_ms=None), + TraceHop(ttl=3, ip="1.1.1.1", loss_pct=0.0, avg_ms=12.0), + ], + completed=True, + ) + assert [f.id for f in path_findings(trace)] == [] + + +def test_path_findings_report_an_incomplete_trace(): + trace = TraceResult(target="1.1.1.1", resolved_ip="1.1.1.1", backend="icmplib", hops=[], completed=False) + assert [f.id for f in path_findings(trace)] == ["path.incomplete"]