diff --git a/src/netsleuth/alerting.py b/src/netsleuth/alerting.py index 22c52cb..a19586b 100644 --- a/src/netsleuth/alerting.py +++ b/src/netsleuth/alerting.py @@ -2,6 +2,8 @@ from __future__ import annotations import httpx +from netsleuth.config import Watch + _BAD_STATUSES = {"warn", "crit"} _GOOD_STATUSES = {"ok", "info"} @@ -11,8 +13,7 @@ def should_fire( current: str, last_fired_at: float | None, now: float, - min_interval_s: float, - fire_on: set[str], + watch: Watch, ) -> bool: if previous is None or previous == current: return False @@ -22,14 +23,19 @@ def should_fire( event = "recovered" else: return False - if event not in fire_on: + if event not in set(watch.webhook_on): return False - if last_fired_at is not None and (now - last_fired_at) < min_interval_s: + if ( + last_fired_at is not None + and (now - last_fired_at) < watch.webhook_min_interval_seconds + ): return False return True -def build_payload(session_meta: dict, cycle_summary: dict, previous: str | None, current: str) -> dict: +def build_payload( + session_meta: dict, cycle_summary: dict, previous: str | None, current: str +) -> dict: return { "tool": "netsleuth", "asn": session_meta.get("asn"), @@ -42,10 +48,12 @@ def build_payload(session_meta: dict, cycle_summary: dict, previous: str | None, } -async def post_webhook(client: httpx.AsyncClient, url: str, payload: dict, timeout: float) -> bool: +async def post_webhook( + client: httpx.AsyncClient, url: str, payload: dict, timeout: float +) -> bool: try: response = await client.post(url, json=payload, timeout=timeout) response.raise_for_status() return True - except (httpx.HTTPError, OSError): + except httpx.HTTPError, OSError: return False diff --git a/src/netsleuth/watch.py b/src/netsleuth/watch.py index 86bddee..8da15bb 100644 --- a/src/netsleuth/watch.py +++ b/src/netsleuth/watch.py @@ -99,7 +99,9 @@ class WatchSession: self.cycles.append(summary) def history(self, label: str) -> list[float | None]: - return [(cycle["hosts"].get(label) or {}).get("avg_ms") for cycle in self.cycles] + return [ + (cycle["hosts"].get(label) or {}).get("avg_ms") for cycle in self.cycles + ] def labels(self) -> list[str]: seen: list[str] = [] @@ -137,7 +139,9 @@ def write_session(session: WatchSession, logs_dir: Path) -> Path: def render_dashboard(session: WatchSession) -> Table: - table = Table(title=f"netsleuth --watch · {session.asn or 'unknown ASN'} · cycle {len(session.cycles)}") + table = Table( + title=f"netsleuth --watch · {session.asn or 'unknown ASN'} · cycle {len(session.cycles)}" + ) for column in ("Host", "Avg ms", "Jitter", "Loss", "Trend"): table.add_column(column) latest = session.cycles[-1]["hosts"] if session.cycles else {} @@ -153,7 +157,9 @@ def render_dashboard(session: WatchSession) -> Table: caption_parts: list[str] = [] if session.cycles: last = session.cycles[-1] - caption_parts.append(f"verdict: {last.get('status', '?')} ({last.get('score', '?')}/100)") + caption_parts.append( + f"verdict: {last.get('status', '?')} ({last.get('score', '?')}/100)" + ) speed = next((c["speed"] for c in reversed(session.cycles) if c["speed"]), None) if speed: caption_parts.append( @@ -165,7 +171,6 @@ def render_dashboard(session: WatchSession) -> Table: return table - async def _measure_cycle( client: httpx.AsyncClient, settings: Settings, @@ -186,7 +191,10 @@ async def _measure_cycle( source_ip=bind.ipv4 if bind else None, ) speed = None - if is_speedtest_cycle(cycle, session.speedtest_every_n_cycles) and not options.quick: + if ( + is_speedtest_cycle(cycle, session.speedtest_every_n_cycles) + and not options.quick + ): result = await run_module( "speed", _speed_section(client, settings, options, None), @@ -199,7 +207,6 @@ async def _measure_cycle( async def _maybe_fire_webhook( client: httpx.AsyncClient, webhook_url: str | None, - fire_on: set[str], settings: Settings, session: WatchSession, cycle_summary: dict[str, Any], @@ -215,8 +222,7 @@ async def _maybe_fire_webhook( current_status, last_fired_at, time.perf_counter(), - settings.watch.webhook_min_interval_seconds, - fire_on, + settings.watch, ): payload = build_payload( {"asn": session.asn, "interface": session.interface}, @@ -224,11 +230,14 @@ async def _maybe_fire_webhook( previous_status, current_status, ) - if await post_webhook(client, webhook_url, payload, settings.timeouts.http_seconds): + if await post_webhook( + client, webhook_url, payload, settings.timeouts.http_seconds + ): session.alerts_fired += 1 return time.perf_counter() return last_fired_at + async def run_watch(settings: Settings, options) -> Path: # Imported here, not at module import time: cli imports watch lazily for --watch, # and a module-level import back into cli would close the cycle. @@ -247,14 +256,15 @@ async def run_watch(settings: Settings, options) -> Path: if options.extra_host: hosts.append(("target-host", options.extra_host)) console = Console() - console.print(f"netsleuth {__version__} · watching every {session.interval_seconds}s · Ctrl-C to stop") + console.print( + f"netsleuth {__version__} · watching every {session.interval_seconds}s · Ctrl-C to stop" + ) webhook_url = options.webhook or settings.watch.webhook_url - fire_on = set(settings.watch.webhook_on) if webhook_url: console.print( f"[yellow]--watch will POST your ASN and finding titles to {webhook_url} " - f"on a verdict transition into {sorted(fire_on)}.[/yellow]" + f"on a verdict transition into {sorted(set(settings.watch.webhook_on))}.[/yellow]" ) previous_status: str | None = None last_fired_at: float | None = None @@ -269,27 +279,49 @@ async def run_watch(settings: Settings, options) -> Path: try: geo, _cf, _flags, _raw = await gather_identity(client, settings.providers) session.asn = geo.asn - except (httpx.HTTPError, OSError): + except httpx.HTTPError, OSError: session.asn = None cycle = 1 try: - with Live(render_dashboard(session), console=console, refresh_per_second=settings.watch.dashboard_refresh_hz) as live: + with Live( + render_dashboard(session), + console=console, + refresh_per_second=settings.watch.dashboard_refresh_hz, + ) as live: while True: began = time.perf_counter() pings, speed = await _measure_cycle( - client, settings, options, session, cycle, hosts, caps, bind, _speed_section + client, + settings, + options, + session, + cycle, + hosts, + caps, + bind, + _speed_section, + ) + cycle_summary = summarize_cycle( + cycle, utc_now_iso(), pings, speed, settings.thresholds ) - cycle_summary = summarize_cycle(cycle, utc_now_iso(), pings, speed, settings.thresholds) session.add(cycle_summary) last_fired_at = await _maybe_fire_webhook( - client, webhook_url, fire_on, settings, session, cycle_summary, previous_status, last_fired_at + client, + webhook_url, + settings, + session, + cycle_summary, + previous_status, + last_fired_at, ) previous_status = cycle_summary["status"] live.update(render_dashboard(session)) - await asyncio.sleep(next_delay(began, time.perf_counter(), session.interval_seconds)) + await asyncio.sleep( + next_delay(began, time.perf_counter(), session.interval_seconds) + ) cycle += 1 - except (KeyboardInterrupt, asyncio.CancelledError): + except KeyboardInterrupt, asyncio.CancelledError: pass path = write_session(session, Path(settings.output.logs_dir)) diff --git a/tests/test_alerting.py b/tests/test_alerting.py index d689d7c..d695547 100644 --- a/tests/test_alerting.py +++ b/tests/test_alerting.py @@ -3,40 +3,50 @@ from __future__ import annotations import httpx from netsleuth.alerting import build_payload, post_webhook, should_fire +from netsleuth.config import Watch from netsleuth.exporter import dump_json def test_should_fire_is_silent_on_the_first_ever_tick(): - assert should_fire(None, "crit", None, 100.0, 300.0, {"crit"}) is False + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire(None, "crit", None, 100.0, watch) is False def test_should_fire_is_silent_when_there_is_no_transition(): - assert should_fire("crit", "crit", None, 100.0, 300.0, {"crit"}) is False + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire("crit", "crit", None, 100.0, watch) is False def test_should_fire_is_silent_when_the_new_state_is_not_in_fire_on(): - assert should_fire("ok", "warn", None, 100.0, 300.0, {"crit"}) is False + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire("ok", "warn", None, 100.0, watch) is False def test_should_fire_respects_the_minimum_interval(): - assert should_fire("ok", "crit", 100.0, 150.0, 300.0, {"crit"}) is False + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire("ok", "crit", 100.0, 150.0, watch) is False def test_should_fire_fires_after_the_interval_elapses(): - assert should_fire("ok", "crit", 100.0, 450.0, 300.0, {"crit"}) is True + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire("ok", "crit", 100.0, 450.0, watch) is True def test_should_fire_fires_on_first_transition_with_no_prior_firing(): - assert should_fire("ok", "crit", None, 100.0, 300.0, {"crit"}) is True + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire("ok", "crit", None, 100.0, watch) is True def test_should_fire_treats_recovery_as_a_separate_event(): - assert should_fire("crit", "ok", None, 100.0, 300.0, {"crit"}) is False - assert should_fire("crit", "ok", None, 100.0, 300.0, {"crit", "recovered"}) is True + watch1 = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"]) + assert should_fire("crit", "ok", None, 100.0, watch1) is False + watch2 = Watch(webhook_min_interval_seconds=300, webhook_on=["crit", "recovered"]) + assert should_fire("crit", "ok", None, 100.0, watch2) is True def test_should_fire_does_not_treat_info_to_ok_as_a_recovery(): - assert should_fire("info", "ok", None, 100.0, 300.0, {"recovered"}) is False + watch = Watch(webhook_min_interval_seconds=300, webhook_on=["recovered"]) + assert should_fire("info", "ok", None, 100.0, watch) is False def test_build_payload_shape_and_json_serializable(): @@ -58,12 +68,16 @@ def test_build_payload_shape_and_json_serializable(): async def test_post_webhook_returns_true_on_success(httpx_mock): httpx_mock.add_response(url="https://example.com/hook", status_code=200) async with httpx.AsyncClient() as client: - ok = await post_webhook(client, "https://example.com/hook", {"a": 1}, timeout=5.0) + ok = await post_webhook( + client, "https://example.com/hook", {"a": 1}, timeout=5.0 + ) assert ok is True async def test_post_webhook_returns_false_and_never_raises_on_failure(httpx_mock): httpx_mock.add_response(url="https://example.com/hook", status_code=500) async with httpx.AsyncClient() as client: - ok = await post_webhook(client, "https://example.com/hook", {"a": 1}, timeout=5.0) + ok = await post_webhook( + client, "https://example.com/hook", {"a": 1}, timeout=5.0 + ) assert ok is False