284 lines
16 KiB
Python
284 lines
16 KiB
Python
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()
|