From 71b1fc95e3559c6f92028a81d42228aec8ee8fdf Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:42:22 +0300 Subject: [PATCH] exporter: report assembly, strict json and atomic writes --- src/netcheck/exporter.py | 97 +++++++++++++++++ tests/test_exporter.py | 217 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 314 insertions(+) create mode 100644 src/netcheck/exporter.py create mode 100644 tests/test_exporter.py diff --git a/src/netcheck/exporter.py b/src/netcheck/exporter.py new file mode 100644 index 0000000..f830abb --- /dev/null +++ b/src/netcheck/exporter.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from typing import Any + +from netcheck.interpret import overall_verdict +from netcheck.models import Finding, ModuleResult, to_jsonable + +SCHEMA_VERSION = 1 +SECTION_ORDER = ( + "connection", + "ip_geo", + "vpn_assessment", + "bgp", + "reputation", + "latency", + "path", + "speed", +) + +_UNSAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]+") + + +def sanitize_name(value: str | None, fallback: str = "unknown") -> str: + cleaned = _UNSAFE_NAME_RE.sub("_", (value or "").strip()).strip("._") + return cleaned[:32] or fallback + + +def compact_timestamp(iso: str) -> str: + # Windows forbids ':' in filenames, so the stamp is the compact ISO form. + return re.sub(r"[-:]", "", (iso or "").strip()) or "unknown" + + +def report_filename(asn: str | None, started_at: str, extension: str) -> str: + return f"report_{sanitize_name(asn)}_{compact_timestamp(started_at)}.{extension.lstrip('.')}" + + +def flatten_errors(modules: dict[str, ModuleResult]) -> list[dict[str, Any]]: + flat: list[dict[str, Any]] = [] + for section, result in modules.items(): + for error in result.errors: + entry = to_jsonable(error) + entry["module"] = result.name or section + flat.append(entry) + return flat + + +def build_report( + meta: dict[str, Any], + modules: dict[str, ModuleResult], + findings: list[Finding], + raw: dict[str, Any], +) -> dict[str, Any]: + status, score, summary = overall_verdict(findings) + report: dict[str, Any] = {"schema_version": SCHEMA_VERSION, "meta": to_jsonable(meta)} + for section in SECTION_ORDER: + result = modules.get(section) or ModuleResult(name=section, status="skipped") + report[section] = to_jsonable(result) + report["interpretation"] = { + "overall_status": status, + "overall_score": score, + "summary_text": summary, + "findings": to_jsonable(findings), + } + report["errors"] = flatten_errors(modules) + report["raw"] = to_jsonable(raw) + return report + + +def dump_json(report: dict[str, Any]) -> str: + return json.dumps(to_jsonable(report), allow_nan=False, ensure_ascii=False, indent=2) + + +def atomic_write(path: Path, text: str) -> Path: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(text, encoding="utf-8") + os.replace(tmp, path) + return path + + +def egress_asn(report: dict[str, Any]) -> str | None: + data = (report.get("ip_geo") or {}).get("data") or {} + return ((data.get("egress_v4") or {}) if isinstance(data, dict) else {}).get("asn") + + +def write_report(report: dict[str, Any], markdown: str, logs_dir: Path) -> tuple[Path, Path]: + started = (report.get("meta") or {}).get("started_at", "") + asn = egress_asn(report) + base = Path(logs_dir) + json_path = atomic_write(base / report_filename(asn, started, "json"), dump_json(report)) + md_path = atomic_write(base / report_filename(asn, started, "md"), markdown) + return json_path, md_path diff --git a/tests/test_exporter.py b/tests/test_exporter.py new file mode 100644 index 0000000..e100a3a --- /dev/null +++ b/tests/test_exporter.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import json +import math +from pathlib import Path + +import pytest + +from netcheck.exporter import ( + SCHEMA_VERSION, + SECTION_ORDER, + atomic_write, + build_report, + compact_timestamp, + dump_json, + egress_asn, + flatten_errors, + report_filename, + sanitize_name, + write_report, +) +from netcheck.models import ( + Finding, + IpGeo, + LocalNet, + ModuleResult, + PingResult, + ProbeError, + SpeedResult, +) + + +def meta() -> dict: + return { + "run_id": "b7f1", + "started_at": "2026-08-08T19:12:00Z", + "finished_at": "2026-08-08T19:13:04Z", + "mode": "auto", + "target": None, + "flags": {"quick": True, "full": False, "dnsbl": False, "ndt7": False, "tcp_trace": False}, + "host_os": "Windows", + "capabilities": {"os_name": "Windows", "chosen_latency_backend": "icmp_win"}, + } + + +def modules() -> dict[str, ModuleResult]: + return { + "connection": ModuleResult(name="connection", status="ok", data=LocalNet(iface_name="Wi-Fi 2")), + "ip_geo": ModuleResult( + name="ip_geo", + status="ok", + data={ + "egress_v4": IpGeo(ip="203.0.113.44", asn="AS64500", as_name="Example Telecom"), + "egress_v6": None, + "cf_trace": None, + "dual_stack_note": None, + }, + ), + "latency": ModuleResult( + name="latency", + status="partial", + data=[PingResult(label="cloudflare-dns", host="1.1.1.1", method="icmp_win", sent=5, received=5)], + errors=[ProbeError(source="quad9-dns", kind="timeout", message="2s", retryable=True)], + warnings=["quad9 did not answer"], + duration_ms=1500, + ), + "speed": ModuleResult(name="speed", status="skipped", data=None), + } + + +def test_report_carries_every_top_level_key_from_the_spec(): + report = build_report(meta(), modules(), [], {"ip-api": {"status": "success"}}) + expected = {"schema_version", "meta", "interpretation", "errors", "raw", *SECTION_ORDER} + assert expected.issubset(report.keys()) + assert report["schema_version"] == SCHEMA_VERSION + + +def test_missing_modules_are_rendered_as_skipped_sections_not_dropped(): + report = build_report(meta(), modules(), [], {}) + assert report["bgp"]["status"] == "skipped" + assert report["bgp"]["data"] is None + assert set(SECTION_ORDER) == { + "connection", + "ip_geo", + "vpn_assessment", + "bgp", + "reputation", + "latency", + "path", + "speed", + } + + +def test_interpretation_is_computed_from_the_findings(): + findings = [Finding(id="a", severity="warn", title="Jitter high", detail="d")] + report = build_report(meta(), modules(), findings, {}) + assert report["interpretation"]["overall_status"] == "warn" + assert report["interpretation"]["overall_score"] == 90 + assert report["interpretation"]["findings"][0]["id"] == "a" + + +def test_errors_are_flattened_to_the_top_level_with_their_module(): + flat = flatten_errors(modules()) + assert flat == [ + { + "source": "quad9-dns", + "kind": "timeout", + "message": "2s", + "retryable": True, + "module": "latency", + } + ] + + +def test_raw_payloads_are_stored_verbatim_under_their_source_key(): + raw = {"ip-api": {"status": "success", "as": "AS64500 Example"}, "cf-trace": "ip=203.0.113.44"} + report = build_report(meta(), modules(), [], raw) + assert report["raw"]["ip-api"]["as"] == "AS64500 Example" + assert report["raw"]["cf-trace"] == "ip=203.0.113.44" + + +def test_non_finite_numbers_that_reached_the_pipeline_serialize_as_null(): + # Zero-duration timing math produces exactly this on the failure paths the + # report most needs to show; allow_nan=False would otherwise abort the write. + broken = modules() + broken["speed"] = ModuleResult( + name="speed", + status="partial", + data=SpeedResult( + method="cloudflare", + download_mbps=float("inf"), + upload_mbps=float("nan"), + idle_rtt_ms=float("-inf"), + bufferbloat_down_ms=12.5, + ), + ) + report = build_report(meta(), broken, [], {"cloudflare": {"rate": float("inf")}}) + text = dump_json(report) + back = json.loads(text) + assert back["speed"]["data"]["download_mbps"] is None + assert back["speed"]["data"]["upload_mbps"] is None + assert back["speed"]["data"]["idle_rtt_ms"] is None + assert back["speed"]["data"]["bufferbloat_down_ms"] == 12.5 + assert back["raw"]["cloudflare"]["rate"] is None + assert "Infinity" not in text + assert "NaN" not in text + + +def test_dump_json_is_strict_and_round_trips(): + report = build_report(meta(), modules(), [], {}) + back = json.loads(dump_json(report)) + assert back["meta"]["run_id"] == "b7f1" + assert back["latency"]["data"][0]["label"] == "cloudflare-dns" + assert math.isfinite(back["latency"]["duration_ms"]) + + +def test_compact_timestamp_strips_the_characters_windows_forbids(): + assert compact_timestamp("2026-08-08T19:12:00Z") == "20260808T191200Z" + assert compact_timestamp("") == "unknown" + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("AS64500", "AS64500"), + ("AS64500 Example Telecom", "AS64500_Example_Telecom"), + ("AS64500/../etc", "AS64500_.._etc"), + ("AS:64500", "AS_64500"), + ("", "unknown"), + (None, "unknown"), + ("///", "unknown"), + ], +) +def test_sanitize_name_produces_windows_safe_fragments(value, expected): + assert sanitize_name(value) == expected + + +def test_report_filename_matches_the_documented_pattern(): + assert ( + report_filename("AS64500", "2026-08-08T19:12:00Z", "json") + == "report_AS64500_20260808T191200Z.json" + ) + assert ( + report_filename(None, "2026-08-08T19:12:00Z", "md") + == "report_unknown_20260808T191200Z.md" + ) + + +def test_egress_asn_is_read_out_of_the_assembled_report(): + report = build_report(meta(), modules(), [], {}) + assert egress_asn(report) == "AS64500" + assert egress_asn({"ip_geo": {"data": None}}) is None + assert egress_asn({}) is None + + +def test_atomic_write_creates_the_directory_and_leaves_no_temp_file(tmp_path: Path): + target = tmp_path / "logs" / "report.json" + atomic_write(target, '{"a": 1}') + assert target.read_text(encoding="utf-8") == '{"a": 1}' + assert list((tmp_path / "logs").iterdir()) == [target] + + +def test_atomic_write_replaces_an_existing_file(tmp_path: Path): + target = tmp_path / "report.json" + atomic_write(target, "old") + atomic_write(target, "new") + assert target.read_text(encoding="utf-8") == "new" + assert not list(tmp_path.glob("*.tmp")) + + +def test_write_report_emits_both_artifacts_with_matching_names(tmp_path: Path): + report = build_report(meta(), modules(), [], {}) + json_path, md_path = write_report(report, "# netcheck report\n", tmp_path) + assert json_path.name == "report_AS64500_20260808T191200Z.json" + assert md_path.name == "report_AS64500_20260808T191200Z.md" + assert json.loads(json_path.read_text(encoding="utf-8"))["schema_version"] == SCHEMA_VERSION + assert md_path.read_text(encoding="utf-8").startswith("# netcheck report")