From 005ad5a2ebf09be080be26e60dabdc7e2aa0c3c7 Mon Sep 17 00:00:00 2001 From: admin Date: Sun, 27 Sep 2026 14:20:00 -0500 Subject: [PATCH] Add Windows ping spike monitor and traceroute capture --- .gitignore | 13 ++ README.md | 57 ++++++ Start Monitor.cmd | 4 + network_monitor.py | 386 ++++++++++++++++++++++++++++++++++++++++ test_network_monitor.py | 209 ++++++++++++++++++++++ 5 files changed, 669 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 Start Monitor.cmd create mode 100644 network_monitor.py create mode 100644 test_network_monitor.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8020a19 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +__pycache__/ +logs/ +reports/ + +# Local, account-specific reporting and email preparation scripts. +final_report.py +create_email_draft.py + +# Local credentials and environments. +.env +.env.* +.venv/ +venv/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..01ed5b4 --- /dev/null +++ b/README.md @@ -0,0 +1,57 @@ +# Network spike monitor + +Windows 10/11, Python 3.10 or newer. No packages or administrator rights required. + +Double-click **Start Monitor.cmd**, or run: + +```powershell +python network_monitor.py +``` + +Pings `1.1.1.1` once per second. A spike is a reply of at least 100 ms, or (after 10 healthy samples) at least 3 times the healthy median AND 40 ms above it. Missing replies also count. Windows `tracert` runs after **at least three bad probes in each of two consecutive one-minute periods**. Periods are non-overlapping and start when monitoring starts; the rule is evaluated at the first probe after each period ends. For example, three spikes in minute one and three in minute two trigger a trace at about two minutes. Six spikes in one minute followed by a quiet minute do not trigger it. A period with fewer than three bad probes resets the streak; gaps from computer sleep also break it. Monitoring continues while tracing runs. An incident runs tracert repeatedly, one at a time, with a 2-second pause between attempts (`--trace-retry`). It continues even if pings recover, until a trace captures a spike or you stop the monitor. There is no attempt limit. After capture, ping monitoring continues; another qualifying period can start a new incident after the 120-second cooldown (`--cooldown`). Ctrl+C stops monitoring and cancels an active trace. + +Customize the target and sensitivity: + +```powershell +python network_monitor.py --target 8.8.8.8 --threshold 80 --required 3 --period 60 --consecutive 2 --cooldown 60 +``` + +Run one diagnostic immediately, then monitor normally (automatic incidents still repeat until capture): + +```powershell +python network_monitor.py --trace-now +``` + +Use `python network_monitor.py --help` for all options. Hostnames resolve to a single IPv4 address at startup so ping and tracert use the same destination. This measures the path to the selected target; it does not measure every Internet service. Run separate instances with separate `--log-dir` directories to compare destinations. + +## Preserved evidence + +Reverse DNS is enabled: raw logs show hop hostnames alongside their IP addresses when available, for example `hop1.subsystem2.spectrum.com [192.0.2.1]`. JSON hop records include both `hostname` and `ip`. Routers without reverse DNS show only an IP; routers that do not reply may show only stars. DNS lookup can make each trace take longer. + +All logs go to `logs` beside the program by default, or the directory supplied with `--log-dir`: + +- Only traces that capture a spike are retained. Each has a timestamped `trace_YYYYMMDD_HHMMSS_microseconds.txt` containing the **entire, unmodified stdout and stderr**, plus a matching `.json` with hop timings, command, incident ID, attempt number, and assessment. +- While a trace runs, output streams to `.pending.txt`. If it captures a spike, this file becomes the saved `.txt`; otherwise it is deleted, including on normal cancellation without measured spikes. A forced process kill or power failure may leave a temporary `.pending.txt`. +- `events.jsonl` appends individual spikes/lost replies, period summaries, and brief incident/attempt results between runs. Healthy pings appear only in the console and are not written to disk. Discarded attempts do not retain their full output or hop arrays here. +- Existing logs from earlier versions are left in place. Raw output uses the Windows command's native encoding. + +No logs are automatically deleted or rotated. A long-running session can grow large; archive logs after stopping the monitor. Probe intervals are best effort; timeouts may make a cycle longer than the requested interval. Values below 1 ms are represented as 0.5 ms for calculations. Parsing requires the Windows ping reply's `TTL` field and `ms` timings; unrecognized successful replies produce an explicit error. + +## Interpreting reports + +A trace is retained when **any individual timed probe at any hop** meets the spike threshold: 100 ms by default, or the relative ping threshold based on the healthy baseline frozen when the incident starts. A single slow probe counts even if the other two replies are fast, downstream hops recover, or the destination never replies. Partial/cancelled traces are also retained if they contain measured slow replies; timeouts alone do not qualify. This captures evidence without claiming the responding router is faulty. + +Saved JSON includes the individual `spike_probes` with hop, hostname, IP, and RTT. Compact event summaries include `peak_probe_ms`, `spike_limit_ms`, and `spike_probe_count` so you can see why an attempt was kept or discarded. Hop medians and `--hop-jump` are used only for the separate tentative segment assessment, not to decide whether to save a trace. + +The report looks for an RTT jump that remains elevated through subsequent responding hops, including the destination. It identifies a possible segment, not a proven faulty router. These are round-trip measurements taken at different times, not per-link transit times. Return paths and ICMP prioritization can differ. A spike can end while tracert is still running. Missing hops are not evidence of packet loss in forwarded traffic. + +Microsoft documents that some routers do not answer traceroute probes: [TRACERT troubleshooting](https://learn.microsoft.com/en-us/troubleshoot/windows-server/networking/trace-route-troubleshoot-tcp-ip-problems). + +## Verification + +```powershell +python -m unittest -v +python network_monitor.py --target 127.0.0.1 --count 3 --trace-now +``` + +`--count` ends monitoring after the requested probe count and cancels an active incident capture. A healthy loopback diagnostic should leave no saved trace files. diff --git a/Start Monitor.cmd b/Start Monitor.cmd new file mode 100644 index 0000000..efb54d1 --- /dev/null +++ b/Start Monitor.cmd @@ -0,0 +1,4 @@ +@echo off +cd /d "%~dp0" +python network_monitor.py %* +pause diff --git a/network_monitor.py b/network_monitor.py new file mode 100644 index 0000000..45ba3bc --- /dev/null +++ b/network_monitor.py @@ -0,0 +1,386 @@ +"""Windows ping spike monitor. Python 3.10+, standard library only.""" +from __future__ import annotations + +import argparse +from collections import deque +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +import ipaddress +import json +from pathlib import Path +import re +import socket +import statistics +import subprocess +import sys +import threading +import time + + +RTT = re.compile(r"([<=]?)\s*(\d+(?:[.,]\d+)?)\s*ms\b", re.I) +IP = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b") + + +def timestamp(): + return datetime.now().astimezone().isoformat(timespec="milliseconds") + + +def run_command(command, timeout, stop, capture_path=None): + """Capture output without blocking shutdown or filling a pipe.""" + if capture_path is not None: + # Send stdout/stderr straight to disk, preserving the complete original + # output even if this monitor is interrupted before tracing finishes. + with capture_path.open("wb", buffering=0) as capture: + with subprocess.Popen(command, stdout=capture, stderr=subprocess.STDOUT, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) as process: + deadline = time.monotonic() + timeout + try: + while process.poll() is None: + if stop.wait(0.2) or time.monotonic() >= deadline: + process.kill() + process.wait() + return -1, capture_path.read_bytes().decode(errors="replace") + except BaseException: + process.kill() + process.wait() + raise + return process.returncode, capture_path.read_bytes().decode(errors="replace") + with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) as process: + deadline = time.monotonic() + timeout + while True: + try: + output, _ = process.communicate(timeout=0.2) + return process.returncode, output.decode(errors="replace") + except subprocess.TimeoutExpired: + if stop.is_set() or time.monotonic() >= deadline: + process.kill() + output, _ = process.communicate() + return -1, output.decode(errors="replace") + "\n[Cancelled or command timed out]\n" + + +def parse_ping(output): + # TTL identifies an echo reply, excluding summaries and unreachable messages. + for line in output.splitlines(): + if re.search(r"TTL\s*=\s*\d+", line, re.I): + match = RTT.search(line) + if match: + return 0.5 if match[1] == "<" else float(match[2].replace(",", ".")) + raise ValueError("Echo reply received, but its latency could not be parsed: " + line) + return None + + +class Detector: + def __init__(self, threshold=100, multiplier=3, increase=40, window=60, + required=3, period=60, consecutive=2, start=None): + self.threshold = threshold + self.multiplier = multiplier + self.increase = increase + self.baseline = deque(maxlen=window) + self.required = required + self.period = period + self.consecutive = consecutive + self.period_start = time.monotonic() if start is None else start + self.period_bad = 0 + self.streak = 0 + + def observe(self, latency, now=None): + now = time.monotonic() if now is None else now + elapsed_periods = int((now - self.period_start) // self.period) + completed_bad = None + trigger = False + if elapsed_periods: + completed_bad = self.period_bad + self.streak = self.streak + 1 if completed_bad >= self.required else 0 + # Unobserved periods (e.g. computer sleep) break the streak. + if elapsed_periods > 1: + self.streak = 0 + trigger = self.streak >= self.consecutive + self.period_start += elapsed_periods * self.period + self.period_bad = 0 + baseline = statistics.median(self.baseline) if self.baseline else None + relative = (baseline is not None and len(self.baseline) >= 10 + and latency is not None + and latency >= max(baseline * self.multiplier, baseline + self.increase)) + bad = latency is None or latency >= self.threshold or relative + self.period_bad += int(bad) + # Freeze the healthy baseline during incidents so persistent lag stays visible. + if latency is not None and not bad: + self.baseline.append(latency) + return {"bad": bool(bad), "baseline_ms": baseline, + "trigger": trigger, "period_bad": self.period_bad, + "completed_period_bad": completed_bad, "consecutive_bad_periods": self.streak} + + +def parse_trace(output): + hops = [] + for line in output.splitlines(): + number = re.match(r"^\s*(\d+)\s+", line) + if not number: + continue + remainder = line[number.end():] + matches = list(RTT.finditer(remainder)) + times = [0.5 if m[1] == "<" else float(m[2].replace(",", ".")) for m in matches] + addresses = IP.findall(remainder) + hostname_match = re.search(r"([^\s\[\]]+)\s+\[(?:\d{1,3}\.){3}\d{1,3}\]", remainder) + hops.append({"hop": int(number[1]), "ip": addresses[-1] if addresses else None, + "hostname": hostname_match[1] if hostname_match else None, + "rtt_ms": times, "median_ms": statistics.median(times) if times else None, + "timeouts": remainder.count("*")}) + return hops + + +def sustained_jump(hops, destination, jump_ms): + """Return structured evidence only when a jump persists to the destination.""" + responding = [h for h in hops if h["median_ms"] is not None] + if not responding or responding[-1]["ip"] != destination: + return None + for index, hop in enumerate(responding): + previous = responding[index - 1] if index else None + floor = previous["median_ms"] if previous else 0 + downstream = responding[index + 1:] + if (hop["median_ms"] - floor >= jump_ms and downstream + and all(h["median_ms"] - floor >= jump_ms for h in downstream)): + return {"previous": previous, "hop": hop, "floor_ms": floor, + "destination_ms": responding[-1]["median_ms"]} + return None + + +def analyze_trace(hops, destination, jump_ms): + responding = [h for h in hops if h["median_ms"] is not None] + if not responding: + return "Inconclusive: no hops returned timed replies." + if responding[-1]["ip"] != destination: + return "Inconclusive: destination did not return timed replies; cannot locate the end-to-end delay." + evidence = sustained_jump(hops, destination, jump_ms) + if evidence: + previous, hop, floor = evidence["previous"], evidence["hop"], evidence["floor_ms"] + start = f"hop {previous['hop']} ({previous['ip']})" if previous else "this computer" + return (f"Possible delay begins between {start} and hop {hop['hop']} ({hop['ip']}); " + f"RTT rises from {floor:.1f} to {hop['median_ms']:.1f} ms and stays elevated " + "through the destination. This is a clue, not proof of a faulty router.") + if len(responding) >= 2 and responding[-1]["median_ms"] - responding[-2]["median_ms"] >= jump_ms: + return "Delay appears only at the destination/final segment; the destination may also deprioritize ICMP." + return ("No sustained downstream latency jump found. Isolated slow or silent hops do not identify " + "a forwarding problem; the spike may have ended before tracing reached it.") + + +class Log: + def __init__(self, directory): + directory.mkdir(parents=True, exist_ok=True) + self.directory = directory + self.file = (directory / "events.jsonl").open("a", encoding="utf-8") + self.lock = threading.Lock() + + def write(self, event, message=None, **fields): + with self.lock: + now = timestamp() + self.file.write(json.dumps({"time": now, "event": event, **fields}) + "\n") + self.file.flush() + if message: + print(f"[{now}] {message}", flush=True) + + def ping(self, target, latency, state): + label = "NO REPLY" if latency is None else f"{latency:.1f} ms" + message = f"{target}: {label}" + if state["bad"]: + self.write("ping", message + " [SPIKE/LOSS]", + target=target, latency_ms=latency, **state) + else: + with self.lock: + print(f"[{timestamp()}] {message}", flush=True) + + +def trace(target, args, log, stop, baseline=None, incident_id=None, attempt=1): + name = datetime.now().strftime("trace_%Y%m%d_%H%M%S_%f") + command = ["tracert", "-4", "-h", str(args.max_hops), "-w", str(args.trace_timeout), target] + raw_path = log.directory / (name + ".pending.txt") + log.write("trace_start", f"Running tracert to {target} (attempt {attempt}); keeping output only if a spike is captured.", + command=command, incident_id=incident_id, attempt=attempt) + try: + code, output = run_command(command, args.max_hops * 3 * args.trace_timeout / 1000 + 30, + stop, capture_path=raw_path) + except BaseException: + raw_path.unlink(missing_ok=True) + raise + hops = parse_trace(output) + assessment = analyze_trace(hops, target, args.hop_jump) + evidence = sustained_jump(hops, target, args.hop_jump) + spike_limit = args.threshold + if baseline is not None: + spike_limit = min(spike_limit, max(baseline * args.multiplier, baseline + args.increase)) + # Preserve transient samples even when medians hide them or later hops recover. + # Capturing a slow diagnostic reply is separate from locating a forwarding fault. + spike_probes = [{"hop": hop["hop"], "ip": hop["ip"], "hostname": hop["hostname"], + "rtt_ms": rtt} + for hop in hops for rtt in hop["rtt_ms"] if rtt >= spike_limit] + captured = bool(spike_probes) + peak_ms = max((rtt for hop in hops for rtt in hop["rtt_ms"]), default=None) + if captured and not evidence: + assessment = "Individual slow trace probe captured; this alone does not locate a faulty hop. " + assessment + if code == -1: + assessment = "Trace cancelled or timed out. " + assessment + if captured: + saved_path = log.directory / (name + ".txt") + raw_path.replace(saved_path) + else: + raw_path.unlink(missing_ok=True) + saved_path = None + report = {"time": timestamp(), "target": target, "returncode": code, "command": command, + "assessment": assessment, "hops": hops, "raw_file": str(saved_path) if saved_path else None, + "spike_captured": captured, "spike_limit_ms": spike_limit, + "spike_probes": spike_probes, "spike_probe_count": len(spike_probes), + "peak_probe_ms": peak_ms, "sustained_jump_found": evidence is not None, + "incident_id": incident_id, "attempt": attempt} + if captured: + (log.directory / (name + ".json")).write_text(json.dumps(report, indent=2), encoding="utf-8") + # Keep only the short result in the append-only event log for discarded attempts. + summary = {key: value for key, value in report.items() if key not in ("hops", "spike_probes")} + suffix = f" Saved: {saved_path}" if captured else " Output discarded (no captured spike)." + log.write("trace_complete", assessment + suffix, **summary) + return report + + +def trace_incident(target, args, log, stop, baseline): + incident_id = datetime.now().strftime("incident_%Y%m%d_%H%M%S_%f") + attempt = 0 + log.write("incident_start", "Incident detected: repeating tracert until a spike is captured.", + incident_id=incident_id, baseline_ms=baseline) + while not stop.is_set(): + attempt += 1 + report = trace(target, args, log, stop, baseline, incident_id, attempt) + if report["spike_captured"]: + log.write("incident_captured", f"Spike captured on trace {attempt}. Returning to ping monitoring.", + incident_id=incident_id, attempts=attempt, raw_file=report["raw_file"]) + return + if stop.is_set(): + break + log.write("trace_retry", f"No trace probe crossed the spike threshold; next trace in {args.trace_retry:g}s.", + incident_id=incident_id, attempt=attempt) + if stop.wait(args.trace_retry): + break + log.write("incident_cancelled", "Trace capture stopped before a spike was captured.", + incident_id=incident_id, attempts=attempt) + + +def positive_float(value): + result = float(value) + if not 0 < result < float("inf"): + raise argparse.ArgumentTypeError("must be a finite number greater than zero") + return result + + +def positive_int(value): + result = int(value) + if result <= 0: + raise argparse.ArgumentTypeError("must be greater than zero") + return result + + +def arguments(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--target", default="1.1.1.1", help="IPv4 address or hostname (default: 1.1.1.1)") + parser.add_argument("--interval", type=positive_float, default=1, help="seconds between probes") + parser.add_argument("--threshold", type=positive_float, default=100, help="absolute spike threshold in ms") + parser.add_argument("--multiplier", type=positive_float, default=3, help="relative baseline multiplier") + parser.add_argument("--increase", type=positive_float, default=40, help="minimum relative increase in ms") + parser.add_argument("--required", type=positive_int, default=3, help="bad probes needed per period (default: 3)") + parser.add_argument("--period", type=positive_float, default=60, help="period duration in seconds (default: 60)") + parser.add_argument("--consecutive", type=positive_int, default=2, help="consecutive bad periods needed (default: 2)") + parser.add_argument("--baseline-window", type=positive_int, default=60) + parser.add_argument("--cooldown", type=positive_float, default=120, help="seconds after capture before another incident can start") + parser.add_argument("--trace-retry", type=positive_float, default=2, help="seconds between incident trace attempts (default: 2)") + parser.add_argument("--ping-timeout", type=positive_int, default=1000, help="ping timeout in ms") + parser.add_argument("--trace-timeout", type=positive_int, default=500, help="per-hop probe timeout in ms") + parser.add_argument("--max-hops", type=positive_int, default=30) + parser.add_argument("--hop-jump", type=positive_float, default=30, help="sustained hop RTT increase in ms") + parser.add_argument("--log-dir", type=Path, default=Path(__file__).resolve().parent / "logs") + parser.add_argument("--count", type=positive_int, help="stop after this many probes (default: continuous)") + parser.add_argument("--trace-now", action="store_true", help="also run a diagnostic immediately") + args = parser.parse_args() + if args.max_hops > 255: + parser.error("--max-hops cannot exceed 255") + if args.baseline_window < 10: + parser.error("--baseline-window must be at least 10") + return args + + +def main(): + args = arguments() + if sys.platform != "win32": + print("This program requires Windows (ping.exe and tracert.exe).", file=sys.stderr) + return 1 + try: + target = socket.gethostbyname(args.target) + ipaddress.IPv4Address(target) + except (OSError, ValueError) as error: + print(f"Cannot resolve target: {error}", file=sys.stderr) + return 1 + stop = threading.Event() + log = Log(args.log_dir.resolve()) + detector = Detector(args.threshold, args.multiplier, args.increase, + args.baseline_window, args.required, args.period, args.consecutive) + executor = ThreadPoolExecutor(max_workers=1) + pending = None + next_incident = float("-inf") + pending_incident = False + samples = 0 + failed = False + try: + log.write("start", f"Monitoring {args.target} ({target}). Ctrl+C stops. Logs: {log.directory}", + target=target, settings={k: str(v) if isinstance(v, Path) else v for k, v in vars(args).items()}) + log.write("trigger_rule", f"Trace after {args.required}+ spikes/losses per {args.period:g}s " + f"period for {args.consecutive} consecutive periods.") + if args.trace_now: + pending = executor.submit(trace, target, args, log, stop) + while not stop.is_set() and (args.count is None or samples < args.count): + started = time.monotonic() + if pending is not None and pending.done(): + pending.result() + pending = None + if pending_incident: + next_incident = time.monotonic() + args.cooldown + pending_incident = False + code, output = run_command(["ping", "-4", "-n", "1", "-w", str(args.ping_timeout), target], + args.ping_timeout / 1000 + 5, stop) + if stop.is_set(): + break + latency = parse_ping(output) + if code == 0 and latency is None: + raise ValueError("ping reported success but no echo reply could be parsed: " + output) + state = detector.observe(latency) + if state["completed_period_bad"] is not None: + log.write("period_complete", f"Period: {state['completed_period_bad']} spikes/losses; " + f"consecutive qualifying periods: {state['consecutive_bad_periods']}.", + bad_count=state["completed_period_bad"], + streak=state["consecutive_bad_periods"]) + samples += 1 + log.ping(target, latency, state) + if state["trigger"] and pending is None and time.monotonic() >= next_incident: + baseline = state["baseline_ms"] if len(detector.baseline) >= 10 else None + pending_incident = True + pending = executor.submit(trace_incident, target, args, log, stop, baseline) + if args.count is None or samples < args.count: + stop.wait(max(0, args.interval - (time.monotonic() - started))) + if pending: + if pending_incident: + stop.set() # --count remains a bounded monitoring run. + log.write("waiting", "Finishing diagnostic; Ctrl+C cancels.") + while not pending.done(): + time.sleep(0.2) + pending.result() + except KeyboardInterrupt: + print("\nStopping...", flush=True) + except Exception as error: + failed = True + log.write("error", f"Error: {error}", error=str(error)) + finally: + stop.set() + executor.shutdown(wait=True, cancel_futures=True) + log.write("stop", "Monitor stopped.", samples=samples) + log.file.close() + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test_network_monitor.py b/test_network_monitor.py new file mode 100644 index 0000000..e893320 --- /dev/null +++ b/test_network_monitor.py @@ -0,0 +1,209 @@ +import pathlib +import json +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch, Mock +from types import SimpleNamespace + +from network_monitor import Detector, analyze_trace, parse_ping, parse_trace, run_command, trace, trace_incident, Log + + +class MonitorTests(unittest.TestCase): + def test_only_bad_pings_are_logged(self): + with tempfile.TemporaryDirectory() as directory: + log = Log(pathlib.Path(directory)) + detector = Detector(start=0) + try: + with patch("builtins.print") as display: + for second in range(100): + log.ping("1.1.1.1", 10, detector.observe(10, now=second)) + self.assertEqual((log.directory / "events.jsonl").stat().st_size, 0) + log.ping("1.1.1.1", 150, detector.observe(150, now=100)) + log.ping("1.1.1.1", None, detector.observe(None, now=101)) + self.assertEqual(display.call_count, 102) + entries = [json.loads(line) for line in (log.directory / "events.jsonl").read_text().splitlines()] + self.assertEqual([entry["latency_ms"] for entry in entries], [150, None]) + self.assertTrue(all(entry["bad"] and entry["time"] for entry in entries)) + finally: + log.file.close() + + def test_ping(self): + self.assertEqual(parse_ping("Reply from 1.1.1.1: bytes=32 time=21ms TTL=57"), 21) + self.assertEqual(parse_ping("Reply from 127.0.0.1: bytes=32 time<1ms TTL=128"), 0.5) + self.assertIsNone(parse_ping("Reply from 192.168.1.1: Destination host unreachable.")) + self.assertIsNone(parse_ping("Request timed out.")) + + def test_two_consecutive_minutes(self): + detector = Detector(start=0) + for second in range(120): + latency = 150 if second in (5, 25, 45, 65, 85, 105) else 10 + self.assertFalse(detector.observe(latency, now=second)["trigger"]) + result = detector.observe(10, now=120) + self.assertTrue(result["trigger"]) + self.assertEqual(result["consecutive_bad_periods"], 2) + self.assertFalse(detector.observe(10, now=121)["trigger"]) + self.assertFalse(detector.observe(10, now=180)["trigger"]) + + def test_single_burst_does_not_qualify_twice(self): + detector = Detector(start=0) + for second in (5, 6, 7, 8, 9, 10): + detector.observe(150, now=second) + self.assertFalse(detector.observe(10, now=60)["trigger"]) + self.assertFalse(detector.observe(10, now=120)["trigger"]) + + def test_quiet_minute_breaks_streak(self): + detector = Detector(start=0) + for second in (5, 25, 45, 125, 145, 165): + self.assertFalse(detector.observe(None, now=second)["trigger"]) + self.assertFalse(detector.observe(10, now=180)["trigger"]) + self.assertEqual(detector.streak, 1) + + def test_boundary_probe_belongs_to_new_minute(self): + detector = Detector(start=0) + for second in (1, 20, 40, 61, 80): + detector.observe(150, now=second) + result = detector.observe(150, now=120) + self.assertFalse(result["trigger"]) + self.assertEqual(result["completed_period_bad"], 2) + self.assertEqual(result["period_bad"], 1) + + def test_relative_spikes_and_loss_count(self): + detector = Detector(start=0) + for second in range(10): + detector.observe(10, now=second) + for second in (15, 35, 55, 65, 85): + detector.observe(65, now=second) + detector.observe(None, now=105) + result = detector.observe(10, now=120) + self.assertTrue(result["trigger"]) + self.assertEqual(result["baseline_ms"], 10) + + def test_sleep_breaks_streak(self): + detector = Detector(start=0) + for second in (5, 25, 45, 65, 85, 105): + detector.observe(None, now=second) + self.assertFalse(detector.observe(10, now=240)["trigger"]) + self.assertEqual(detector.streak, 0) + + def test_custom_periods(self): + detector = Detector(start=0, required=1, period=10, consecutive=3) + for second in (1, 11, 21): + self.assertFalse(detector.observe(150, now=second)["trigger"]) + self.assertTrue(detector.observe(10, now=30)["trigger"]) + + def test_trace_attribution(self): + trace = " 1 <1 ms 1 ms 1 ms 192.168.1.1\n 2 80 ms 90 ms 85 ms 10.0.0.1\n 3 90 ms 95 ms 91 ms 1.1.1.1\n" + hops = parse_trace(trace) + self.assertIn("hop 2 (10.0.0.1)", analyze_trace(hops, "1.1.1.1", 30)) + hops[-1]["median_ms"] = 10 + self.assertIn("No sustained", analyze_trace(hops, "1.1.1.1", 30)) + self.assertIn("Inconclusive", analyze_trace(hops[:-1], "1.1.1.1", 30)) + self.assertEqual(parse_trace(" 4 * * * Request timed out.")[0]["timeouts"], 3) + + def test_trace_hostnames_and_addresses(self): + hops = parse_trace(" 1 12 ms 13 ms 11 ms hop1.subsystem2.spectrum.com [192.0.2.1]\n" + " 2 * 20 ms 21 ms 198.51.100.2\n") + self.assertEqual(hops[0]["hostname"], "hop1.subsystem2.spectrum.com") + self.assertEqual(hops[0]["ip"], "192.0.2.1") + self.assertIsNone(hops[1]["hostname"]) + self.assertEqual(hops[1]["ip"], "198.51.100.2") + + def trace_args(self): + return SimpleNamespace(max_hops=30, trace_timeout=500, hop_jump=30, + threshold=100, multiplier=3, increase=40, trace_retry=0.001) + + def test_incident_retries_until_capture(self): + results = [{"spike_captured": False}, {"spike_captured": False}, + {"spike_captured": True, "raw_file": "captured.txt"}] + with patch("network_monitor.trace", side_effect=results) as tracer: + log = Mock() + trace_incident("1.1.1.1", self.trace_args(), log, threading.Event(), 10) + self.assertEqual(tracer.call_count, 3) + self.assertEqual([c.args[-1] for c in tracer.call_args_list], [1, 2, 3]) + self.assertEqual(log.write.call_args.args[0], "incident_captured") + + def test_incident_cancellation_stops_retries(self): + stop = threading.Event() + def cancel(*args): + stop.set() + return {"spike_captured": False} + with patch("network_monitor.trace", side_effect=cancel) as tracer: + log = Mock() + trace_incident("1.1.1.1", self.trace_args(), log, stop, 10) + self.assertEqual(tracer.call_count, 1) + self.assertEqual(log.write.call_args.args[0], "incident_cancelled") + + def test_only_spike_traces_are_retained(self): + cases = [ + (" 1 1 ms 1 ms 1 ms 192.168.1.1\n 2 110 ms 120 ms 115 ms 10.0.0.1\n 3 120 ms 130 ms 125 ms 1.1.1.1\n", 0, None, True), + # A normally long path must not qualify without destination spike latency. + (" 1 1 ms 1 ms 1 ms 192.168.1.1\n 2 40 ms 40 ms 40 ms 10.0.0.1\n 3 45 ms 45 ms 45 ms 1.1.1.1\n", 0, None, False), + # Relative threshold also works below the absolute threshold. + (" 1 1 ms 1 ms 1 ms 192.168.1.1\n 2 60 ms 60 ms 60 ms 10.0.0.1\n 3 65 ms 65 ms 65 ms 1.1.1.1\n", 0, 10, True), + (" 1 1 ms 1 ms 1 ms 192.168.1.1\n 2 150 ms 150 ms 150 ms 10.0.0.1\n 3 10 ms 10 ms 10 ms 1.1.1.1\n", 0, None, True), + (" 1 120 ms 120 ms 120 ms 192.168.1.1\n 2 * * * Request timed out.\n", 0, None, True), + (" 1 120 ms 120 ms 120 ms 192.168.1.1\n 2 150 ms 150 ms 150 ms 1.1.1.1\n", -1, None, True), + ] + cases.extend([ + # One brief slow response must survive median filtering and downstream recovery. + (" 1 10 ms 300 ms 11 ms hop1.subsystem2.spectrum.com [192.0.2.1]\n 2 15 ms 16 ms 17 ms 1.1.1.1\n", 0, None, True), + # Destination-only spike is still useful evidence. + (" 1 1 ms 1 ms 1 ms 192.168.1.1\n 2 12 ms 100 ms 13 ms 1.1.1.1\n", 0, None, True), + # No measured slow replies: neither timeouts nor a near-threshold response qualifies. + (" 1 * * * Request timed out.\n", 0, None, False), + (" 1 99 ms 10 ms 11 ms 192.168.1.1\n", 0, None, False), + ]) + for output, code, baseline, expected in cases: + with self.subTest(output=output, code=code), tempfile.TemporaryDirectory() as directory: + log = Log(pathlib.Path(directory)) + def command(*args, capture_path): + self.assertNotIn("-d", args[0]) + capture_path.write_bytes(output.encode()) + return code, output + try: + with patch("network_monitor.run_command", side_effect=command): + report = trace("1.1.1.1", self.trace_args(), log, threading.Event(), baseline) + self.assertEqual(report["spike_captured"], expected) + self.assertEqual(bool(report["spike_probes"]), expected) + files = list(pathlib.Path(directory).glob("trace_*")) + self.assertEqual(len(files), 2 if expected else 0) + if expected: + self.assertEqual(pathlib.Path(report["raw_file"]).read_bytes(), output.encode()) + else: + self.assertIsNone(report["raw_file"]) + self.assertNotIn('"hops"', (pathlib.Path(directory) / "events.jsonl").read_text()) + finally: + log.file.close() + + def test_entire_output_preserved(self): + with tempfile.TemporaryDirectory() as directory: + path = pathlib.Path(directory) / "trace.txt" + code, _ = run_command([sys.executable, "-c", "import os; os.write(1,b'header\\r\\n'); os.write(2,b' 1 * * *\\r\\n'); os.write(1,b'Trace complete.\\r\\n')"], 5, threading.Event(), path) + self.assertEqual(code, 0) + self.assertEqual(path.read_bytes(), b"header\r\n 1 * * *\r\nTrace complete.\r\n") + + def test_partial_output_survives_cancel(self): + with tempfile.TemporaryDirectory() as directory: + path = pathlib.Path(directory) / "trace.txt" + stop = threading.Event() + # Wait for actual output to reach disk before cancelling. + def cancel(): + import time + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if path.exists() and path.stat().st_size: + break + time.sleep(0.01) + stop.set() + watcher = threading.Thread(target=cancel) + watcher.start() + code, _ = run_command([sys.executable, "-c", "import os,time; os.write(1,b'partial hop\\r\\n'); time.sleep(30)"], 10, stop, path) + watcher.join() + self.assertEqual(code, -1) + self.assertEqual(path.read_bytes(), b"partial hop\r\n") + + +if __name__ == "__main__": + unittest.main()