From 4e4956ceebf3b4086d7dfcf72d9d3ea2317ccfac Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 03:24:03 +0300 Subject: [PATCH] compare: diff identity, latency, speed and findings between two reports --- src/netcheck/compare.py | 140 +++++++++++++++++++++++++++++ tests/fixtures/reports/after.json | 80 +++++++++++++++++ tests/fixtures/reports/before.json | 69 ++++++++++++++ tests/test_compare.py | 119 ++++++++++++++++++++++++ 4 files changed, 408 insertions(+) create mode 100644 src/netcheck/compare.py create mode 100644 tests/fixtures/reports/after.json create mode 100644 tests/fixtures/reports/before.json create mode 100644 tests/test_compare.py diff --git a/src/netcheck/compare.py b/src/netcheck/compare.py new file mode 100644 index 0000000..ba8aa42 --- /dev/null +++ b/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" diff --git a/tests/fixtures/reports/after.json b/tests/fixtures/reports/after.json new file mode 100644 index 0000000..e142389 --- /dev/null +++ b/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": {} +} diff --git a/tests/fixtures/reports/before.json b/tests/fixtures/reports/before.json new file mode 100644 index 0000000..8ef6660 --- /dev/null +++ b/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": {} +} diff --git a/tests/test_compare.py b/tests/test_compare.py new file mode 100644 index 0000000..f1b0ec5 --- /dev/null +++ b/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)