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()