compare: diff identity, latency, speed and findings between two reports

src/netcheck/compare.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from netcheck.exporter import badge
+
+
+@dataclass
+class Change:
+    label: str
+    before: Any = None
+    after: Any = None
+    delta: float | None = None
+
+
+@dataclass
+class ReportDiff:
+    identity: list[Change] = field(default_factory=list)
+    latency: list[Change] = field(default_factory=list)
+    speed: list[Change] = field(default_factory=list)
+    new_findings: list[dict] = field(default_factory=list)
+    resolved_findings: list[dict] = field(default_factory=list)
+
+
+def load_report(path: Path) -> dict:
+    return json.loads(Path(path).read_text(encoding="utf-8"))
+
+
+def _data(report: dict, section: str) -> Any:
+    return (report.get(section) or {}).get("data")
+
+
+def _geo(report: dict) -> dict:
+    bundle = _data(report, "ip_geo") or {}
+    return (bundle.get("egress_v4") or {}) if isinstance(bundle, dict) else {}
+
+
+def _delta(before: Any, after: Any) -> float | None:
+    if isinstance(before, (int, float)) and isinstance(after, (int, float)):
+        return round(float(after) - float(before), 3)
+    return None
+
+
+def identity_changes(before: dict, after: dict) -> list[Change]:
+    a, b = _geo(before), _geo(after)
+    vpn_a = _data(before, "vpn_assessment") or {}
+    vpn_b = _data(after, "vpn_assessment") or {}
+    pairs = [
+        ("Egress IP", a.get("ip"), b.get("ip")),
+        ("ASN", a.get("asn"), b.get("asn")),
+        ("Organisation", a.get("as_name") or a.get("org"), b.get("as_name") or b.get("org")),
+        ("Country", a.get("country_code"), b.get("country_code")),
+        ("Address type", a.get("ip_type"), b.get("ip_type")),
+        ("VPN verdict", vpn_a.get("verdict"), vpn_b.get("verdict")),
+        ("VPN confidence", vpn_a.get("confidence"), vpn_b.get("confidence")),
+    ]
+    return [Change(label, x, y, _delta(x, y)) for label, x, y in pairs if x != y]
+
+
+def latency_changes(before: dict, after: dict) -> list[Change]:
+    a = {p.get("label"): p for p in _data(before, "latency") or []}
+    b = {p.get("label"): p for p in _data(after, "latency") or []}
+    changes: list[Change] = []
+    for label in sorted(set(a) | set(b)):
+        pa, pb = a.get(label) or {}, b.get(label) or {}
+        for metric in ("avg_ms", "jitter_ms", "loss_pct"):
+            x, y = pa.get(metric), pb.get(metric)
+            if x != y:
+                changes.append(Change(f"{label} {metric}", x, y, _delta(x, y)))
+    return changes
+
+
+def speed_changes(before: dict, after: dict) -> list[Change]:
+    a, b = _data(before, "speed") or {}, _data(after, "speed") or {}
+    fields = (
+        ("Speedtest method", "method"),
+        ("Download Mbps", "download_mbps"),
+        ("Upload Mbps", "upload_mbps"),
+        ("Bufferbloat down ms", "bufferbloat_down_ms"),
+        ("Bufferbloat up ms", "bufferbloat_up_ms"),
+        ("Bufferbloat grade", "bufferbloat_grade"),
+    )
+    return [
+        Change(label, a.get(key), b.get(key), _delta(a.get(key), b.get(key)))
+        for label, key in fields
+        if a.get(key) != b.get(key)
+    ]
+
+
+def finding_changes(before: dict, after: dict) -> tuple[list[dict], list[dict]]:
+    a = {f.get("id"): f for f in (before.get("interpretation") or {}).get("findings") or []}
+    b = {f.get("id"): f for f in (after.get("interpretation") or {}).get("findings") or []}
+    return [b[i] for i in b if i not in a], [a[i] for i in a if i not in b]
+
+
+def diff_reports(before: dict, after: dict) -> ReportDiff:
+    new, resolved = finding_changes(before, after)
+    return ReportDiff(
+        identity=identity_changes(before, after),
+        latency=latency_changes(before, after),
+        speed=speed_changes(before, after),
+        new_findings=new,
+        resolved_findings=resolved,
+    )
+
+
+def _fmt(value: Any) -> str:
+    if value is None:
+        return "—"
+    return f"{value:g}" if isinstance(value, (int, float)) else str(value)
+
+
+def _block(title: str, changes: list[Change]) -> list[str]:
+    if not changes:
+        return [f"### {title}", "", "No change.", ""]
+    return (
+        [f"### {title}", "", "| Field | Before | After | Δ |", "|---|---|---|---|"]
+        + [f"| {c.label} | {_fmt(c.before)} | {_fmt(c.after)} | {_fmt(c.delta)} |" for c in changes]
+        + [""]
+    )
+
+
+def render_diff(diff: ReportDiff, emoji: bool = True) -> str:
+    lines = ["# netcheck compare", ""]
+    lines += _block("Identity", diff.identity)
+    lines += _block("Latency", diff.latency)
+    lines += _block("Speed", diff.speed)
+    lines += ["### Findings", ""]
+    if not diff.new_findings and not diff.resolved_findings:
+        lines += ["No findings appeared or cleared.", ""]
+    else:
+        lines += [
+            f"- new: {badge(f.get('severity', 'info'), emoji)} {f.get('title', '')}" for f in diff.new_findings
+        ]
+        lines += [f"- resolved: {badge('ok', emoji)} {f.get('title', '')}" for f in diff.resolved_findings]
+        lines += [""]
+    return "\n".join(lines).rstrip() + "\n"

tests/fixtures/reports/after.json

@@ -0,0 +1,80 @@
+{
+  "schema_version": 1,
+  "meta": { "run_id": "bbbb2222", "started_at": "2026-08-08T21:40:00Z", "mode": "auto", "target": null },
+  "connection": { "name": "connection", "status": "ok", "data": { "iface_name": "wg0" }, "errors": [], "warnings": [], "duration_ms": 41 },
+  "ip_geo": {
+    "name": "ip_geo",
+    "status": "ok",
+    "data": {
+      "egress_v4": {
+        "ip": "198.51.100.7",
+        "asn": "AS64777",
+        "as_name": "Example Hosting",
+        "org": "Example Hosting BV",
+        "country_code": "DE",
+        "ip_type": "hosting"
+      },
+      "egress_v6": null,
+      "cf_trace": null,
+      "dual_stack_note": null
+    },
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 950
+  },
+  "vpn_assessment": {
+    "name": "vpn_assessment",
+    "status": "ok",
+    "data": { "verdict": "confirmed", "confidence": 0.75, "signals": [], "tunnel_iface": "wg0", "dns_leak": null },
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 310
+  },
+  "bgp": { "name": "bgp", "status": "ok", "data": { "asn": "AS64777", "holder": "Example Hosting BV" }, "errors": [], "warnings": [], "duration_ms": 1300 },
+  "reputation": { "name": "reputation", "status": "ok", "data": { "captcha_risk": "medium", "firehol_hits": [] }, "errors": [], "warnings": [], "duration_ms": 820 },
+  "latency": {
+    "name": "latency",
+    "status": "ok",
+    "data": [
+      { "label": "cloudflare-dns", "host": "1.1.1.1", "method": "icmp_win", "avg_ms": 46.9, "jitter_ms": 1.9, "loss_pct": 0.0 },
+      { "label": "quad9-dns", "host": "9.9.9.9", "method": "icmp_win", "avg_ms": 48.0, "jitter_ms": 3.0, "loss_pct": 1.0 }
+    ],
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 5400
+  },
+  "path": { "name": "path", "status": "ok", "data": [], "errors": [], "warnings": [], "duration_ms": 4100 },
+  "speed": {
+    "name": "speed",
+    "status": "ok",
+    "data": {
+      "method": "cloudflare",
+      "download_mbps": 96.1,
+      "upload_mbps": 38.0,
+      "bufferbloat_down_ms": 74.0,
+      "bufferbloat_grade": "D"
+    },
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 31000
+  },
+  "interpretation": {
+    "overall_status": "crit",
+    "overall_score": 75,
+    "summary_text": "Bufferbloat under load (downstream): grade D",
+    "findings": [
+      {
+        "id": "speed.bufferbloat_down",
+        "severity": "crit",
+        "title": "Bufferbloat under load (downstream): grade D",
+        "detail": "Latency rose by 74.0 ms while saturating the downstream direction.",
+        "metric": "bufferbloat_down_ms",
+        "value": "D",
+        "threshold": 60.0,
+        "advice": "Enabling SQM/fq_codel on the router is the standard fix."
+      }
+    ]
+  },
+  "errors": [],
+  "raw": {}
+}

tests/fixtures/reports/before.json

@@ -0,0 +1,69 @@
+{
+  "schema_version": 1,
+  "meta": { "run_id": "aaaa1111", "started_at": "2026-08-08T19:12:00Z", "mode": "auto", "target": null },
+  "connection": { "name": "connection", "status": "ok", "data": { "iface_name": "Wi-Fi 2" }, "errors": [], "warnings": [], "duration_ms": 40 },
+  "ip_geo": {
+    "name": "ip_geo",
+    "status": "ok",
+    "data": {
+      "egress_v4": {
+        "ip": "203.0.113.44",
+        "asn": "AS64500",
+        "as_name": "Example Telecom",
+        "org": "Example Telecom BV",
+        "country_code": "NL",
+        "ip_type": "residential"
+      },
+      "egress_v6": null,
+      "cf_trace": null,
+      "dual_stack_note": null
+    },
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 900
+  },
+  "vpn_assessment": {
+    "name": "vpn_assessment",
+    "status": "ok",
+    "data": { "verdict": "none", "confidence": 0.1, "signals": [], "tunnel_iface": null, "dns_leak": null },
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 300
+  },
+  "bgp": { "name": "bgp", "status": "ok", "data": { "asn": "AS64500", "holder": "Example Telecom BV" }, "errors": [], "warnings": [], "duration_ms": 1200 },
+  "reputation": { "name": "reputation", "status": "ok", "data": { "captcha_risk": "low", "firehol_hits": [] }, "errors": [], "warnings": [], "duration_ms": 800 },
+  "latency": {
+    "name": "latency",
+    "status": "ok",
+    "data": [
+      { "label": "cloudflare-dns", "host": "1.1.1.1", "method": "icmp_win", "avg_ms": 12.4, "jitter_ms": 1.9, "loss_pct": 0.0 },
+      { "label": "google-dns", "host": "8.8.8.8", "method": "icmp_win", "avg_ms": 14.0, "jitter_ms": 2.0, "loss_pct": 0.0 }
+    ],
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 5200
+  },
+  "path": { "name": "path", "status": "ok", "data": [], "errors": [], "warnings": [], "duration_ms": 4000 },
+  "speed": {
+    "name": "speed",
+    "status": "ok",
+    "data": {
+      "method": "cloudflare",
+      "download_mbps": 284.3,
+      "upload_mbps": 41.7,
+      "bufferbloat_down_ms": 12.0,
+      "bufferbloat_grade": "B"
+    },
+    "errors": [],
+    "warnings": [],
+    "duration_ms": 30000
+  },
+  "interpretation": {
+    "overall_status": "ok",
+    "overall_score": 100,
+    "summary_text": "No problems found on this connection.",
+    "findings": []
+  },
+  "errors": [],
+  "raw": {}
+}

tests/test_compare.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import pytest
+
+from netcheck.compare import (
+    Change,
+    diff_reports,
+    finding_changes,
+    identity_changes,
+    latency_changes,
+    load_report,
+    render_diff,
+    speed_changes,
+)
+
+
+@pytest.fixture()
+def before(fixtures_dir):
+    return load_report(fixtures_dir / "reports" / "before.json")
+
+
+@pytest.fixture()
+def after(fixtures_dir):
+    return load_report(fixtures_dir / "reports" / "after.json")
+
+
+def by_label(changes: list[Change]) -> dict[str, Change]:
+    return {c.label: c for c in changes}
+
+
+def test_load_report_reads_a_saved_report(before):
+    assert before["schema_version"] == 1
+    assert before["meta"]["run_id"] == "aaaa1111"
+
+
+def test_identity_reports_the_egress_and_asn_change(before, after):
+    changes = by_label(identity_changes(before, after))
+    assert changes["Egress IP"].before == "203.0.113.44"
+    assert changes["Egress IP"].after == "198.51.100.7"
+    assert changes["ASN"].before == "AS64500"
+    assert changes["ASN"].after == "AS64777"
+    assert changes["Country"].after == "DE"
+    assert changes["Address type"].after == "hosting"
+
+
+def test_identity_reports_the_vpn_verdict_and_confidence_delta(before, after):
+    changes = by_label(identity_changes(before, after))
+    assert changes["VPN verdict"].before == "none"
+    assert changes["VPN verdict"].after == "confirmed"
+    assert changes["VPN confidence"].delta == pytest.approx(0.65)
+
+
+def test_identity_of_a_report_against_itself_is_empty(before):
+    assert identity_changes(before, before) == []
+
+
+def test_latency_deltas_are_signed_per_host_and_metric(before, after):
+    changes = by_label(latency_changes(before, after))
+    assert changes["cloudflare-dns avg_ms"].delta == pytest.approx(34.5)
+    assert changes["cloudflare-dns avg_ms"].before == 12.4
+    assert "cloudflare-dns jitter_ms" not in changes
+
+
+def test_a_host_present_in_only_one_report_is_still_reported(before, after):
+    changes = by_label(latency_changes(before, after))
+    assert changes["google-dns avg_ms"].after is None
+    assert changes["quad9-dns avg_ms"].before is None
+    assert changes["quad9-dns avg_ms"].delta is None
+
+
+def test_latency_of_a_report_against_itself_is_empty(before):
+    assert latency_changes(before, before) == []
+
+
+def test_speed_deltas_cover_throughput_and_the_bufferbloat_grade(before, after):
+    changes = by_label(speed_changes(before, after))
+    assert changes["Download Mbps"].delta == pytest.approx(-188.2)
+    assert changes["Upload Mbps"].delta == pytest.approx(-3.7)
+    assert changes["Bufferbloat grade"].before == "B"
+    assert changes["Bufferbloat grade"].after == "D"
+    assert "Speedtest method" not in changes
+
+
+def test_findings_are_split_into_new_and_resolved(before, after):
+    new, resolved = finding_changes(before, after)
+    assert [f["id"] for f in new] == ["speed.bufferbloat_down"]
+    assert resolved == []
+    new, resolved = finding_changes(after, before)
+    assert new == []
+    assert [f["id"] for f in resolved] == ["speed.bufferbloat_down"]
+
+
+def test_diff_reports_assembles_every_part(before, after):
+    diff = diff_reports(before, after)
+    assert diff.identity
+    assert diff.latency
+    assert diff.speed
+    assert len(diff.new_findings) == 1
+    assert diff.resolved_findings == []
+
+
+def test_render_diff_shows_before_and_after_values(before, after):
+    text = render_diff(diff_reports(before, after))
+    assert text.startswith("# netcheck compare")
+    assert "203.0.113.44" in text
+    assert "198.51.100.7" in text
+    assert "-188.2" in text
+    assert "Bufferbloat under load" in text
+
+
+def test_render_diff_of_two_identical_reports_says_so(before):
+    text = render_diff(diff_reports(before, before))
+    assert text.count("No change.") == 3
+    assert "No findings appeared or cleared." in text
+
+
+def test_render_diff_honours_the_emoji_setting(before, after):
+    assert "🔴" in render_diff(diff_reports(before, after), emoji=True)
+    assert "[crit]" in render_diff(diff_reports(before, after), emoji=False)