From f8f5fbc49c3f9f177a0a975e90444f2e2febadc1 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sat, 8 Aug 2026 21:28:58 +0300 Subject: [PATCH] models: ModuleResult, ProbeError, Finding, Signal envelopes --- src/netcheck/models.py | 102 +++++++++++++++++++++++++++++++++++++++++ tests/test_models.py | 99 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 src/netcheck/models.py create mode 100644 tests/test_models.py diff --git a/src/netcheck/models.py b/src/netcheck/models.py new file mode 100644 index 0000000..1004cc2 --- /dev/null +++ b/src/netcheck/models.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import math +from dataclasses import dataclass, field, fields, is_dataclass +from datetime import date, datetime +from enum import Enum +from pathlib import Path +from typing import Any + +ERROR_KINDS = ( + "timeout", + "http_error", + "rate_limited", + "blocked", + "parse_error", + "unavailable", + "no_privilege", + "not_applicable", +) +STATUSES = ("ok", "partial", "failed", "skipped") +SEVERITIES = ("ok", "info", "warn", "crit") +DIRECTIONS = ("vpn", "clean") + + +@dataclass +class ProbeError: + source: str + kind: str + message: str + retryable: bool = False + + def __post_init__(self) -> None: + if self.kind not in ERROR_KINDS: + raise ValueError(f"unknown ProbeError kind: {self.kind!r}") + + +@dataclass +class ModuleResult: + name: str + status: str + data: Any | None = None + errors: list[ProbeError] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + started_at: str = "" + duration_ms: int = 0 + + def __post_init__(self) -> None: + if self.status not in STATUSES: + raise ValueError(f"unknown ModuleResult status: {self.status!r}") + + +@dataclass +class Finding: + id: str + severity: str + title: str + detail: str + metric: str | None = None + value: float | str | None = None + threshold: float | str | None = None + advice: str | None = None + + def __post_init__(self) -> None: + if self.severity not in SEVERITIES: + raise ValueError(f"unknown Finding severity: {self.severity!r}") + + +@dataclass +class Signal: + name: str + observed: bool + weight: float + direction: str + note: str = "" + + def __post_init__(self) -> None: + if self.direction not in DIRECTIONS: + raise ValueError(f"unknown Signal direction: {self.direction!r}") + + +def to_jsonable(obj: Any) -> Any: + if obj is None or isinstance(obj, (str, bool, int)): + return obj + if isinstance(obj, float): + return obj if math.isfinite(obj) else None + if is_dataclass(obj) and not isinstance(obj, type): + return {f.name: to_jsonable(getattr(obj, f.name)) for f in fields(obj)} + if isinstance(obj, dict): + return {str(k): to_jsonable(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple)): + return [to_jsonable(v) for v in obj] + if isinstance(obj, (set, frozenset)): + return sorted(to_jsonable(v) for v in obj) + if isinstance(obj, Enum): + return to_jsonable(obj.value) + if isinstance(obj, (datetime, date)): + return obj.isoformat() + if isinstance(obj, Path): + return str(obj) + if isinstance(obj, bytes): + return obj.decode("utf-8", "replace") + return str(obj) diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..6b9cec3 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import json + +import pytest + +from netcheck.models import Finding, ModuleResult, ProbeError, Signal, to_jsonable + + +def test_probe_error_rejects_unknown_kind(): + with pytest.raises(ValueError, match="unknown ProbeError kind"): + ProbeError(source="ip-api", kind="exploded", message="boom") + + +@pytest.mark.parametrize( + "kind", + [ + "timeout", + "http_error", + "rate_limited", + "blocked", + "parse_error", + "unavailable", + "no_privilege", + "not_applicable", + ], +) +def test_probe_error_accepts_documented_kinds(kind): + assert ProbeError(source="s", kind=kind, message="m").kind == kind + + +def test_module_result_rejects_unknown_status(): + with pytest.raises(ValueError, match="unknown ModuleResult status"): + ModuleResult(name="bgp", status="borked") + + +def test_module_result_serializes_to_strict_json(): + result = ModuleResult( + name="reputation", + status="partial", + data={"firehol_hits": ["firehol_level1"]}, + errors=[ProbeError(source="internetdb", kind="timeout", message="8s", retryable=True)], + warnings=["abuseipdb key missing"], + started_at="2026-08-08T19:12:00Z", + duration_ms=1234, + ) + text = json.dumps(to_jsonable(result), allow_nan=False) + back = json.loads(text) + assert back["name"] == "reputation" + assert back["status"] == "partial" + assert back["data"]["firehol_hits"] == ["firehol_level1"] + assert back["errors"][0] == { + "source": "internetdb", + "kind": "timeout", + "message": "8s", + "retryable": True, + } + assert back["warnings"] == ["abuseipdb key missing"] + assert back["duration_ms"] == 1234 + + +def test_to_jsonable_coerces_non_finite_numbers_to_null(): + payload = {"a": float("inf"), "b": float("-inf"), "c": float("nan"), "d": 1.5} + out = to_jsonable(payload) + assert out == {"a": None, "b": None, "c": None, "d": 1.5} + json.dumps(out, allow_nan=False) + + +def test_to_jsonable_handles_nested_dataclasses_and_sets(): + finding = Finding( + id="latency.high", + severity="warn", + title="Latency above target", + detail="avg 130 ms to 1.1.1.1", + metric="avg_ms", + value=130.0, + threshold=100.0, + advice="Check for a saturated uplink.", + ) + signal = Signal(name="tunnel_iface", observed=True, weight=0.35, direction="vpn", note="wg0") + out = to_jsonable({"findings": [finding], "signals": (signal,), "tags": {"a"}}) + assert out["findings"][0]["severity"] == "warn" + assert out["signals"][0]["weight"] == 0.35 + assert out["tags"] == ["a"] + json.dumps(out, allow_nan=False) + + +def test_finding_rejects_unknown_severity(): + with pytest.raises(ValueError, match="unknown Finding severity"): + Finding( + id="x", + severity="apocalyptic", + title="t", + detail="d", + metric=None, + value=None, + threshold=None, + advice=None, + )