From e4f9c3b5b7ca4e5a64b9edc4cac46abc01cb81a2 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:46:53 +0300 Subject: [PATCH] exporter: explicit placeholders so a failed section never reads as a healthy one --- src/netcheck/exporter.py | 45 +++++++++++++++++++++++++ tests/test_exporter.py | 72 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+) diff --git a/src/netcheck/exporter.py b/src/netcheck/exporter.py index 4e6a54b..898d81b 100644 --- a/src/netcheck/exporter.py +++ b/src/netcheck/exporter.py @@ -190,7 +190,34 @@ def _tldr(report: dict, emoji: bool) -> list[str]: ] + [""] +_SECTION_TITLES = { + "connection": "## Connection & identity", + "ip_geo": "## Connection & identity", + "vpn_assessment": "## VPN / proxy assessment", + "bgp": "## ASN & BGP intelligence", + "reputation": "## Reputation", + "latency": "## Latency", + "path": "## Path", + "speed": "## Speed", +} + + +def unavailable(report: dict[str, Any], section: str) -> str | None: + module = _module(report, section) + if module.get("status") in ("ok", "partial") and module.get("data") is not None: + return None + detail = "; ".join(f"{e.get('source')}: {e.get('message')}" for e in module.get("errors") or []) + if not detail: + detail = "no data was collected for this section" + return f"_Not available — {module.get('status', 'skipped')}: {detail}._" + + def _connection(report: dict) -> list[str]: + conn_note = unavailable(report, "connection") + geo_note = unavailable(report, "ip_geo") + if conn_note or geo_note: + notes = [n for n in (conn_note, geo_note) if n] + return [_SECTION_TITLES["connection"], "", *notes, ""] local = _data(report, "connection") or {} geo_bundle = _data(report, "ip_geo") or {} geo = geo_bundle.get("egress_v4") or {} @@ -214,6 +241,9 @@ def _connection(report: dict) -> list[str]: def _vpn(report: dict) -> list[str]: + note = unavailable(report, "vpn_assessment") + if note: + return [_SECTION_TITLES["vpn_assessment"], "", note, ""] vpn = _data(report, "vpn_assessment") or {} lines = [ "## VPN / proxy assessment", @@ -256,6 +286,9 @@ def _vpn(report: dict) -> list[str]: def _bgp(report: dict) -> list[str]: + note = unavailable(report, "bgp") + if note: + return [_SECTION_TITLES["bgp"], "", note, ""] bgp = _data(report, "bgp") or {} rows = [ ["ASN", f"{bgp.get('asn') or '?'} — {bgp.get('holder') or '?'}"], @@ -281,6 +314,9 @@ def _bgp(report: dict) -> list[str]: def _reputation(report: dict) -> list[str]: + note = unavailable(report, "reputation") + if note: + return [_SECTION_TITLES["reputation"], "", note, ""] rep = _data(report, "reputation") or {} idb = rep.get("internetdb") or {} rows = [ @@ -307,6 +343,9 @@ def _reputation(report: dict) -> list[str]: def _latency(report: dict) -> list[str]: + note = unavailable(report, "latency") + if note: + return [_SECTION_TITLES["latency"], "", note, ""] pings = _data(report, "latency") or [] lines = ["## Latency", ""] + _table( ["Host", "Address", "Method", "Avg ms", "Min", "Max", "Jitter", "Loss"], @@ -338,6 +377,9 @@ def _bar(value: Any) -> str: def _path(report: dict) -> list[str]: + note = unavailable(report, "path") + if note: + return [_SECTION_TITLES["path"], "", note, ""] traces = _data(report, "path") or [] lines = ["## Path", ""] for trace in traces: @@ -360,6 +402,9 @@ def _path(report: dict) -> list[str]: def _speed(report: dict) -> list[str]: + note = unavailable(report, "speed") + if note: + return [_SECTION_TITLES["speed"], "", note, ""] speed = _data(report, "speed") or {} grade = speed.get("bufferbloat_grade") or "?" rows = [ diff --git a/tests/test_exporter.py b/tests/test_exporter.py index 36e8348..f8df2eb 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -526,3 +526,75 @@ def test_markdown_without_emoji_uses_text_badges(): text = render_markdown(full_report(), emoji=False) assert "[warn]" in text assert "🟡" not in text + + +from netcheck.exporter import unavailable + + +def dead_modules() -> dict[str, ModuleResult]: + return { + section: ModuleResult( + name=section, + status="failed", + data=None, + errors=[ProbeError(source=section, kind="unavailable", message="network unreachable", retryable=True)], + duration_ms=8000, + ) + for section in SECTION_ORDER + } + + +def dead_report() -> dict: + return build_report(meta(), dead_modules(), [], {}) + + +def test_unavailable_describes_a_failed_module_and_stays_quiet_on_a_healthy_one(): + assert unavailable(full_report(), "bgp") is None + note = unavailable(dead_report(), "bgp") + assert note is not None + assert "failed" in note + assert "network unreachable" in note + + +def test_unavailable_explains_a_skipped_module_without_inventing_an_error(): + report = build_report(meta(), {"speed": ModuleResult(name="speed", status="skipped")}, [], {}) + note = unavailable(report, "speed") + assert note is not None + assert "skipped" in note + + +def test_an_all_modules_failed_run_still_renders_every_section(): + text = render_markdown(dead_report()) + headings = [line for line in text.splitlines() if line.startswith("## ")] + assert len(headings) == 10 + assert text.count("Not available") == 8 + + +def test_an_all_modules_failed_report_is_still_valid_strict_json(): + report = dead_report() + back = json.loads(dump_json(report)) + assert back["schema_version"] == SCHEMA_VERSION + assert len(back["errors"]) == len(SECTION_ORDER) + assert all(section in back for section in SECTION_ORDER) + assert back["interpretation"]["overall_status"] == "ok" + assert back["raw"] == {} + + +def test_an_all_modules_failed_run_writes_both_artifacts(tmp_path: Path): + report = dead_report() + json_path, md_path = write_report(report, render_markdown(report), tmp_path) + assert json_path.name.startswith("report_unknown_") + assert md_path.name.startswith("report_unknown_") + assert "## Run diagnostics" in md_path.read_text(encoding="utf-8") + + +def test_the_diagnostics_table_still_names_every_failed_module(): + block = render_markdown(dead_report()).split("## Run diagnostics")[1] + assert block.count("failed") == len(SECTION_ORDER) + assert "unavailable" in block + + +def test_a_partial_module_with_data_is_rendered_normally_not_as_a_placeholder(): + report = build_report(meta(), modules(), [], {}) + assert unavailable(report, "latency") is None + assert "cloudflare-dns" in render_markdown(report)