config: pydantic-settings loader with env > .env > yaml precedence

src/netcheck/config.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, Field, SecretStr
+from pydantic_settings import (
+    BaseSettings,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+    YamlConfigSettingsSource,
+)
+
+DEFAULT_CONFIG_PATH = Path("config.yaml")
+DEFAULT_ENV_PATH = Path(".env")
+
+
+class Timeouts(BaseModel):
+    http_seconds: float = 8.0
+    module_seconds: float = 30.0
+    speedtest_seconds: float = 90.0
+    dns_seconds: float = 4.0
+    subprocess_seconds: float = 60.0
+
+
+class HostSpec(BaseModel):
+    label: str
+    host: str
+
+
+class Probing(BaseModel):
+    reference_hosts: list[HostSpec] = Field(
+        default_factory=lambda: [
+            HostSpec(label="cloudflare-dns", host="1.1.1.1"),
+            HostSpec(label="google-dns", host="8.8.8.8"),
+            HostSpec(label="quad9-dns", host="9.9.9.9"),
+        ]
+    )
+    service_hosts: list[HostSpec] = Field(
+        default_factory=lambda: [
+            HostSpec(label="cloudflare", host="cloudflare.com"),
+            HostSpec(label="google", host="google.com"),
+            HostSpec(label="github", host="github.com"),
+        ]
+    )
+    ping_count: int = 20
+    quick_ping_count: int = 5
+    ping_interval_seconds: float = 0.25
+    ping_timeout_seconds: float = 2.0
+    mtr_cycles: int = 10
+    quick_mtr_cycles: int = 1
+    max_hops: int = 30
+    trace_concurrency: int = 2
+
+
+class Speedtest(BaseModel):
+    enabled_tiers: list[str] = Field(default_factory=lambda: ["ookla_bin", "cloudflare", "fastcom"])
+    download_sizes_bytes: list[int] = Field(default_factory=lambda: [1_000_000, 10_000_000, 25_000_000])
+    upload_sizes_bytes: list[int] = Field(default_factory=lambda: [1_000_000, 5_000_000])
+    cloudflare_base_url: str = "https://speed.cloudflare.com"
+    fastcom_api_url: str = "https://api.fast.com/netflix/speedtest/v2"
+    ndt7_locate_url: str = "https://locate.measurementlab.net/v2/nearest/ndt/ndt7"
+    bufferbloat_probe_interval_seconds: float = 0.2
+
+
+class Providers(BaseModel):
+    cf_trace_url: str = "https://www.cloudflare.com/cdn-cgi/trace"
+    ip_api_url: str = "http://ip-api.com/json/"
+    freeipapi_url: str = "https://freeipapi.com/api/json/"
+    ipinfo_url: str = "https://ipinfo.io/"
+    ipwhois_url: str = "https://ipwho.is/"
+    ripestat_base_url: str = "https://stat.ripe.net/data"
+    asrank_url: str = "https://api.asrank.caida.org/v2/graphql"
+    peeringdb_base_url: str = "https://www.peeringdb.com/api"
+    internetdb_url: str = "https://internetdb.shodan.io/"
+    abuseipdb_url: str = "https://api.abuseipdb.com/api/v2/check"
+    cymru_origin_zone: str = "origin.asn.cymru.com"
+    cymru_asn_zone: str = "asn.cymru.com"
+    ripestat_max_rows: int = 200
+    ripestat_timeframe_days: int = 14
+    peeringdb_cache_hours: int = 24
+    firehol_refresh_hours: int = 24
+    firehol_netsets: list[str] = Field(default_factory=list)
+
+
+class Dnsbl(BaseModel):
+    zones: list[str] = Field(
+        default_factory=lambda: [
+            "zen.spamhaus.org",
+            "bl.spamcop.net",
+            "b.barracudacentral.org",
+            "dnsbl.dronebl.org",
+        ]
+    )
+
+
+class Band(BaseModel):
+    good: float
+    warn: float
+
+
+class BufferbloatBands(BaseModel):
+    a: float = 5.0
+    b: float = 30.0
+    c: float = 60.0
+    d: float = 200.0
+    e: float = 400.0
+
+
+class VpnBands(BaseModel):
+    likely: float = 0.40
+    confirmed: float = 0.75
+
+
+class Thresholds(BaseModel):
+    latency_ms: Band = Band(good=40.0, warn=100.0)
+    jitter_ms: Band = Band(good=5.0, warn=20.0)
+    loss_pct: Band = Band(good=0.0, warn=2.0)
+    bufferbloat_ms: BufferbloatBands = BufferbloatBands()
+    vpn_confidence: VpnBands = VpnBands()
+
+
+class Output(BaseModel):
+    logs_dir: str = "./logs"
+    cache_dir: str = "./.cache"
+    emoji: bool = True
+
+
+class Watch(BaseModel):
+    interval_seconds: int = 60
+    speedtest_every_n_cycles: int = 10
+    dashboard_refresh_hz: int = 4
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_prefix="NETCHECK_",
+        env_nested_delimiter="__",
+        env_file_encoding="utf-8",
+        extra="ignore",
+        yaml_file=None,
+    )
+
+    timeouts: Timeouts = Timeouts()
+    probing: Probing = Probing()
+    speedtest: Speedtest = Speedtest()
+    providers: Providers = Providers()
+    dnsbl: Dnsbl = Dnsbl()
+    thresholds: Thresholds = Thresholds()
+    output: Output = Output()
+    watch: Watch = Watch()
+
+    ipinfo_token: SecretStr | None = Field(default=None, alias="IPINFO_TOKEN")
+    peeringdb_api_key: SecretStr | None = Field(default=None, alias="PEERINGDB_API_KEY")
+    abuseipdb_api_key: SecretStr | None = Field(default=None, alias="ABUSEIPDB_API_KEY")
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        return (
+            init_settings,
+            env_settings,
+            dotenv_settings,
+            YamlConfigSettingsSource(settings_cls),
+            file_secret_settings,
+        )
+
+
+def load_settings(
+    config_path: Path | None = None,
+    env_file: Path | None = None,
+) -> Settings:
+    yaml_path = Path(config_path) if config_path is not None else DEFAULT_CONFIG_PATH
+    dotenv_path = Path(env_file) if env_file is not None else DEFAULT_ENV_PATH
+    Settings.model_config["yaml_file"] = yaml_path if yaml_path.exists() else None
+    kwargs: dict[str, Any] = {"_env_file": str(dotenv_path) if dotenv_path.exists() else None}
+    return Settings(**kwargs)

tests/test_config.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import textwrap
+
+import pytest
+
+from netcheck.config import load_settings
+
+
+@pytest.fixture()
+def yaml_file(tmp_path):
+    path = tmp_path / "config.yaml"
+    path.write_text(
+        textwrap.dedent(
+            """
+            timeouts:
+              http_seconds: 3.5
+            probing:
+              ping_count: 7
+              reference_hosts:
+                - { label: "yaml-host", host: "9.9.9.9" }
+            output:
+              logs_dir: "./yaml-logs"
+              emoji: false
+            thresholds:
+              latency_ms: { good: 11.0, warn: 22.0 }
+            """
+        ).strip(),
+        encoding="utf-8",
+    )
+    return path
+
+
+def test_defaults_apply_when_yaml_omits_a_field(yaml_file, tmp_path):
+    s = load_settings(config_path=yaml_file, env_file=tmp_path / "missing.env")
+    assert s.probing.max_hops == 30
+    assert s.timeouts.module_seconds == 30.0
+
+
+def test_yaml_overrides_defaults(yaml_file, tmp_path):
+    s = load_settings(config_path=yaml_file, env_file=tmp_path / "missing.env")
+    assert s.timeouts.http_seconds == 3.5
+    assert s.probing.ping_count == 7
+    assert s.output.emoji is False
+    assert s.thresholds.latency_ms.warn == 22.0
+    assert s.probing.reference_hosts[0].host == "9.9.9.9"
+
+
+def test_dotenv_overrides_yaml(yaml_file, tmp_path):
+    env_file = tmp_path / ".env"
+    env_file.write_text("NETCHECK_PROBING__PING_COUNT=13\n", encoding="utf-8")
+    s = load_settings(config_path=yaml_file, env_file=env_file)
+    assert s.probing.ping_count == 13
+
+
+def test_environment_overrides_dotenv(yaml_file, tmp_path, monkeypatch):
+    env_file = tmp_path / ".env"
+    env_file.write_text("NETCHECK_PROBING__PING_COUNT=13\n", encoding="utf-8")
+    monkeypatch.setenv("NETCHECK_PROBING__PING_COUNT", "99")
+    s = load_settings(config_path=yaml_file, env_file=env_file)
+    assert s.probing.ping_count == 99
+
+
+def test_secrets_default_to_none_so_the_tool_runs_unconfigured(tmp_path):
+    s = load_settings(config_path=tmp_path / "absent.yaml", env_file=tmp_path / "absent.env")
+    assert s.ipinfo_token is None
+    assert s.peeringdb_api_key is None
+    assert s.abuseipdb_api_key is None
+
+
+def test_secrets_load_from_dotenv_without_the_nested_prefix(tmp_path, monkeypatch):
+    monkeypatch.delenv("IPINFO_TOKEN", raising=False)
+    env_file = tmp_path / ".env"
+    env_file.write_text("IPINFO_TOKEN=abc123\n", encoding="utf-8")
+    s = load_settings(config_path=tmp_path / "absent.yaml", env_file=env_file)
+    assert s.ipinfo_token is not None
+    assert s.ipinfo_token.get_secret_value() == "abc123"
+    assert "abc123" not in repr(s)
+
+
+def test_missing_yaml_is_not_an_error(tmp_path):
+    s = load_settings(config_path=tmp_path / "nope.yaml", env_file=tmp_path / "nope.env")
+    assert s.probing.ping_count == 20