diff --git a/src/netcheck/interpret.py b/src/netcheck/interpret.py index b015a51..54efdc8 100644 --- a/src/netcheck/interpret.py +++ b/src/netcheck/interpret.py @@ -2,8 +2,19 @@ from __future__ import annotations from typing import Iterable -from netcheck.config import Band, Thresholds, VpnBands -from netcheck.models import CfTrace, DnsLeak, Finding, IpGeo, LocalNet, PingResult, Signal, TraceResult, VpnAssessment +from netcheck.config import Band, BufferbloatBands, Thresholds, VpnBands +from netcheck.models import ( + CfTrace, + DnsLeak, + Finding, + IpGeo, + LocalNet, + PingResult, + Signal, + SpeedResult, + TraceResult, + VpnAssessment, +) from netcheck.netinfo import is_tunnel_iface, mtu_anomaly _SEVERITY_ORDER = {"ok": 0, "info": 1, "warn": 2, "crit": 3} @@ -234,3 +245,78 @@ def assess_vpn( tunnel_iface=tunnel_iface, dns_leak=dns_leak, ) + + +_CONSEQUENCES = { + "A": "Video calls and games stay smooth while the line is fully loaded.", + "B": "Barely noticeable; a large upload may add a beat to a video call.", + "C": "Calls and games get choppy whenever something else is downloading.", + "D": "Any big transfer makes calls stutter and pages feel stuck.", + "E": "The connection feels broken while it is busy, even though bandwidth is fine.", + "F": "A single download makes a call unusable; this is queue bloat, not a slow line.", + "?": "Not measured — the speedtest tier that measures it did not run.", +} +_GRADE_SEVERITY = {"A": "ok", "B": "ok", "C": "warn", "D": "crit", "E": "crit", "F": "crit", "?": "info"} + + +def grade_bufferbloat(delta_ms: float | None, bands: BufferbloatBands) -> str: + if delta_ms is None: + return "?" + delta = max(0.0, delta_ms) + for grade, ceiling in (("A", bands.a), ("B", bands.b), ("C", bands.c), ("D", bands.d), ("E", bands.e)): + if delta <= ceiling: + return grade + return "F" + + +def bufferbloat_consequence(grade: str) -> str: + return _CONSEQUENCES.get(grade, _CONSEQUENCES["?"]) + + +def speed_findings(speed: SpeedResult, bands: BufferbloatBands) -> list[Finding]: + if speed.method == "none": + tried = ", ".join(a.tier for a in speed.tier_attempts) or "none" + return [ + Finding( + id="speed.unavailable", + severity="info", + title="No bandwidth measurement", + detail=f"Every speedtest tier failed or was disabled (tried: {tried}).", + advice="Install the Ookla speedtest binary, or rerun without --quick.", + ) + ] + findings: list[Finding] = [] + for direction, delta in (("down", speed.bufferbloat_down_ms), ("up", speed.bufferbloat_up_ms)): + grade = grade_bufferbloat(delta, bands) + severity = _GRADE_SEVERITY[grade] + if severity in ("ok", "info"): + continue + findings.append( + Finding( + id=f"speed.bufferbloat_{direction}", + severity=severity, + title=f"Bufferbloat under load ({direction}stream): grade {grade}", + detail=f"Latency rose by {delta} ms while saturating the {direction}stream direction.", + metric=f"bufferbloat_{direction}_ms", + value=grade, + threshold=bands.c, + advice=bufferbloat_consequence(grade) + + " Enabling SQM/fq_codel on the router is the standard fix.", + ) + ) + return findings + + +_SCORE_PENALTY = {"ok": 0, "info": 3, "warn": 10, "crit": 25} + + +def overall_verdict(findings: list[Finding]) -> tuple[str, int, str]: + score = 100 + for f in findings: + score -= _SCORE_PENALTY[f.severity] + score = max(0, score) + status = worst(f.severity for f in findings) + if status in ("ok", "info"): + return "ok", score, "No problems found on this connection." + headline = [f.title for f in findings if f.severity == status] + return status, score, "; ".join(headline[:3]) diff --git a/tests/test_interpret.py b/tests/test_interpret.py index fa40fb6..73f540e 100644 --- a/tests/test_interpret.py +++ b/tests/test_interpret.py @@ -283,3 +283,93 @@ def test_assess_vpn_returns_a_complete_assessment(): assert assessment.tunnel_iface == "wg0" assert len(assessment.signals) == 2 assert assessment.confidence >= 0.75 + + +from netcheck.config import BufferbloatBands +from netcheck.models import Finding, SpeedResult +from netcheck.interpret import bufferbloat_consequence, grade_bufferbloat, overall_verdict, speed_findings + + +def test_bufferbloat_grades_follow_the_configured_bands(): + b = BufferbloatBands() + assert grade_bufferbloat(0.0, b) == "A" + assert grade_bufferbloat(5.0, b) == "A" + assert grade_bufferbloat(5.1, b) == "B" + assert grade_bufferbloat(30.0, b) == "B" + assert grade_bufferbloat(60.0, b) == "C" + assert grade_bufferbloat(200.0, b) == "D" + assert grade_bufferbloat(400.0, b) == "E" + assert grade_bufferbloat(400.1, b) == "F" + + +def test_bufferbloat_grade_of_an_unmeasured_delta_is_unknown(): + assert grade_bufferbloat(None, BufferbloatBands()) == "?" + + +def test_negative_delta_is_graded_a_not_crashed(): + assert grade_bufferbloat(-3.0, BufferbloatBands()) == "A" + + +def test_each_grade_has_a_plain_language_consequence(): + for grade in "ABCDEF?": + assert bufferbloat_consequence(grade) + assert "call" in bufferbloat_consequence("F").lower() + + +def test_speed_findings_report_a_bad_bufferbloat_grade(): + speed = SpeedResult(method="cloudflare", download_mbps=300.0, upload_mbps=40.0, bufferbloat_down_ms=150.0) + findings = speed_findings(speed, BufferbloatBands()) + bloat = [f for f in findings if f.id == "speed.bufferbloat_down"][0] + assert bloat.severity == "crit" + assert bloat.value == "D" + assert "call" in bloat.advice.lower() or "queue" in bloat.advice.lower() + + +def test_speed_findings_are_silent_on_a_clean_line(): + speed = SpeedResult( + method="ookla_bin", + download_mbps=300.0, + upload_mbps=40.0, + bufferbloat_down_ms=3.0, + bufferbloat_up_ms=4.0, + ) + assert speed_findings(speed, BufferbloatBands()) == [] + + +def test_speed_findings_report_an_exhausted_cascade(): + speed = SpeedResult(method="none") + findings = speed_findings(speed, BufferbloatBands()) + assert [f.id for f in findings] == ["speed.unavailable"] + assert findings[0].severity == "info" + + +def test_overall_verdict_is_healthy_with_no_findings(): + status, score, summary = overall_verdict([]) + assert status == "ok" + assert score == 100 + assert "no problems" in summary.lower() + + +def test_overall_verdict_degrades_with_severity(): + warn = Finding(id="a", severity="warn", title="Jitter high", detail="") + crit = Finding(id="b", severity="crit", title="Loss on path", detail="") + info = Finding(id="c", severity="info", title="No speed data", detail="") + + status, score, summary = overall_verdict([info]) + assert (status, score) == ("ok", 97) + + status, score, summary = overall_verdict([warn]) + assert status == "warn" + assert score == 90 + + status, score, summary = overall_verdict([warn, crit]) + assert status == "crit" + assert score == 65 + assert "Loss on path" in summary + + +def test_overall_score_never_goes_below_zero(): + findings = [Finding(id=f"f{i}", severity="crit", title="x", detail="") for i in range(20)] + status, score, _ = overall_verdict(findings) + assert status == "crit" + assert score == 0