import pathlib import json import sys import subprocess import shutil 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, Log, ping_command, trace_command, main class MonitorTests(unittest.TestCase): @unittest.skipUnless(shutil.which("ping") and shutil.which("tracert" if sys.platform == "win32" else "traceroute"), "requires ping and the platform traceroute command") def test_main_starts_trace_without_trigger_and_closes_log(self): # Exercise real executor shutdown on the interpreter running the suite, # including Python 3.8, where cancel_futures is unavailable. with tempfile.TemporaryDirectory() as directory: result = subprocess.run( [sys.executable, str(pathlib.Path(__file__).with_name("network_monitor.py")), "--target", "127.0.0.1", "--count", "1", "--log-dir", directory], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=20) self.assertEqual(result.returncode, 0, result.stdout.decode(errors="replace")) entries = [json.loads(line) for line in (pathlib.Path(directory) / "events.jsonl").read_text().splitlines()] self.assertEqual(entries[-1]["event"], "stop") self.assertEqual(entries[-1]["samples"], 1) self.assertTrue(any(entry["event"] == "trace_complete" for entry in entries)) self.assertFalse(any(entry["event"] == "error" for entry in entries)) 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_linux_ping(self): self.assertEqual(parse_ping("64 bytes from 127.0.0.1: icmp_seq=1 ttl=64 time=0.026 ms"), 0.026) self.assertIsNone(parse_ping("From 192.0.2.1 icmp_seq=1 Destination Host Unreachable")) self.assertIsNone(parse_ping("1 packets transmitted, 0 received, 100% packet loss")) def test_platform_commands(self): with patch("network_monitor.sys.platform", "linux"): self.assertEqual(ping_command("127.0.0.1", 500), ["ping", "-4", "-n", "-c", "1", "-W", "0.5", "127.0.0.1"]) self.assertEqual(trace_command("127.0.0.1", self.trace_args()), ["traceroute", "-4", "-N", "1", "-q", "3", "-m", "30", "-w", "0.5", "127.0.0.1"]) with patch("network_monitor.sys.platform", "win32"): self.assertEqual(ping_command("127.0.0.1", 500), ["ping", "-4", "-n", "1", "-w", "500", "127.0.0.1"]) self.assertEqual(trace_command("127.0.0.1", self.trace_args()), ["tracert", "-4", "-h", "30", "-w", "500", "127.0.0.1"]) def test_missing_linux_dependency(self): with patch("network_monitor.sys.platform", "linux"), \ patch("network_monitor.sys.argv", ["network_monitor.py"]), \ patch("network_monitor.shutil.which", side_effect=lambda cmd: "/usr/bin/ping" if cmd == "ping" else None), \ patch("sys.stderr") as stderr, patch("network_monitor.Log") as log: self.assertEqual(main(), 1) self.assertIn("sudo apt install", "".join(c.args[0] for c in stderr.write.call_args_list)) log.assert_not_called() def test_linux_trace(self): hops = parse_trace("traceroute to 1.1.1.1 (1.1.1.1), 30 hops max, 60 byte packets\n" " 1 router.example (192.0.2.1) 0.123 ms 150.456 ms 0.321 ms\n" " 2 * * *\n" " 3 1.1.1.1 (1.1.1.1) 12.1 ms * 13.2 ms\n") self.assertEqual(len(hops), 3) self.assertEqual(hops[0]["hostname"], "router.example") self.assertEqual(hops[0]["ip"], "192.0.2.1") self.assertEqual(hops[0]["rtt_ms"], [0.123, 150.456, 0.321]) self.assertEqual(hops[1]["timeouts"], 3) self.assertEqual(hops[2]["timeouts"], 1) def test_linux_trace_multiple_routers(self): hops = parse_trace(" 1 a.example (192.0.2.1) 150 ms b.example (192.0.2.2) 2 ms 3 ms\n" " 2 1.1.1.1 (1.1.1.1) 4 ms 5 ms 6 ms\n") self.assertIsNone(hops[0]["ip"]) self.assertIsNone(hops[0]["hostname"]) self.assertEqual(hops[0]["probes"][0], {"ip": "192.0.2.1", "hostname": "a.example", "rtt_ms": 150}) self.assertEqual(hops[0]["probes"][2]["ip"], "192.0.2.2") self.assertIn("Inconclusive", analyze_trace(hops, "1.1.1.1", 30)) 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_continuous_tracing_repeats_after_capture_and_healthy_trace(self): from concurrent.futures import Future with tempfile.TemporaryDirectory() as directory: executor = Mock() reports = iter([False, True, False]) def submit(*args): future = Future() future.set_result({"spike_captured": next(reports)}) return future executor.submit.side_effect = submit with patch("network_monitor.sys.argv", ["network_monitor.py", "--count", "2", "--trace-retry", "0.001", "--log-dir", directory]), \ patch("network_monitor.shutil.which", return_value="command"), \ patch("network_monitor.socket.gethostbyname", return_value="1.1.1.1"), \ patch("network_monitor.ThreadPoolExecutor", return_value=executor), \ patch("network_monitor.time.monotonic", side_effect=range(100)), \ patch("network_monitor.run_command", return_value=(0, "reply time=10ms TTL=64")), \ patch("builtins.print"): self.assertEqual(main(), 0) self.assertEqual(executor.submit.call_count, 3) self.assertEqual([call.args[-1] for call in executor.submit.call_args_list], [1, 2, 3]) self.assertTrue(executor.submit.call_args.args[4].is_set()) executor.shutdown.assert_called_once_with(wait=True) 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([ (" 1 router.example (192.0.2.1) 0.123 ms 150.456 ms 0.321 ms\n 2 * * *\n", 0, None, True), # 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()