from __future__ import annotations import json import tempfile import unittest from pathlib import Path from servermon.demo import DemoSource from servermon.metrics import ( SystemCollector, build_alerts, parse_cpu_stat, parse_diskstats, parse_loadavg, parse_meminfo, parse_mounts, parse_net_dev, ) from servermon.service import MonitorService from servermon.web import AuthConfig CPU_STAT_1 = """cpu 100 0 100 800 0 0 0 0 0 0 cpu0 50 0 50 400 0 0 0 0 0 0 cpu1 50 0 50 400 0 0 0 0 0 0 intr 0 """ CPU_STAT_2 = """cpu 150 0 150 900 0 0 0 0 0 0 cpu0 75 0 75 450 0 0 0 0 0 0 cpu1 75 0 75 450 0 0 0 0 0 0 intr 0 """ MEMINFO = """MemTotal: 1048576 kB MemFree: 262144 kB MemAvailable: 524288 kB Buffers: 65536 kB Cached: 131072 kB SReclaimable: 16384 kB Shmem: 2048 kB SwapTotal: 262144 kB SwapFree: 196608 kB Dirty: 128 kB """ NET_DEV_1 = """Inter-| Receive | Transmit face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed lo: 100000 100 0 0 0 0 0 0 100000 100 0 0 0 0 0 0 eth0: 1000 10 1 2 0 0 0 0 500 5 0 0 0 0 0 0 """ NET_DEV_2 = """Inter-| Receive | Transmit face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed lo: 100000 100 0 0 0 0 0 0 100000 100 0 0 0 0 0 0 eth0: 2000 20 1 2 0 0 0 0 1000 10 0 0 0 0 0 0 """ DISKSTATS_1 = """ 8 0 sda 10 0 1000 5 10 0 500 3 0 8 0 0 0 0 0 8 1 sda1 10 0 1000 5 10 0 500 3 0 8 0 0 0 0 0 """ DISKSTATS_2 = """ 8 0 sda 20 0 3000 10 20 0 1500 6 0 16 2 0 0 0 0 8 1 sda1 20 0 3000 10 20 0 1500 6 0 16 2 0 0 0 0 """ class ParserTests(unittest.TestCase): def test_cpu_stat(self) -> None: aggregate, cores = parse_cpu_stat(CPU_STAT_1) self.assertIsNotNone(aggregate) self.assertEqual(aggregate["total"], 1000) self.assertEqual(aggregate["idle_all"], 800) self.assertEqual(len(cores), 2) self.assertEqual(cores[0]["index"], 0) def test_meminfo(self) -> None: info = parse_meminfo(MEMINFO) self.assertEqual(info["total"], 1048576 * 1024) self.assertEqual(info["available"], 524288 * 1024) self.assertEqual(info["swap_total"], 262144 * 1024) def test_loadavg(self) -> None: load = parse_loadavg("1.50 1.00 0.50 2/100 1234\n") self.assertEqual(load["load1"], 1.5) self.assertEqual(load["running"], 2) self.assertEqual(load["threads"], 100) def test_net_dev(self) -> None: net = parse_net_dev(NET_DEV_1) self.assertEqual(net["eth0"]["rx_bytes"], 1000) self.assertEqual(net["eth0"]["tx_bytes"], 500) self.assertEqual(net["eth0"]["rx_errors"], 1) self.assertEqual(net["eth0"]["rx_drops"], 2) self.assertIn("lo", net) def test_diskstats(self) -> None: stats = parse_diskstats(DISKSTATS_1) self.assertEqual(stats["sda"]["sectors_read"], 1000) self.assertEqual(stats["sda"]["sectors_written"], 500) self.assertEqual(stats["sda1"]["reads"], 10) def test_mounts_unescape(self) -> None: mounts = parse_mounts("/dev/sda1 /mnt/my\\040disk ext4 rw,relatime 0 0\n") self.assertEqual(mounts[0]["mount"], "/mnt/my disk") self.assertEqual(mounts[0]["fstype"], "ext4") def test_alerts(self) -> None: alerts = build_alerts( cpu_percent=96, memory_percent=62, swap_percent=10, load1=2, cores=8, disks=[{"mount": "/data", "percent": 91}], temp_c=55, ) self.assertTrue(any(a["severity"] == "crit" for a in alerts)) self.assertTrue(any("/data" in a["message"] for a in alerts)) class AuthTests(unittest.TestCase): def test_basic_auth(self) -> None: import base64 auth = AuthConfig.parse("admin:s3cret") assert auth is not None token = base64.b64encode(b"admin:s3cret").decode("ascii") self.assertTrue(auth.check_header("Basic " + token)) self.assertFalse(auth.check_header("Basic " + base64.b64encode(b"admin:wrong").decode("ascii"))) self.assertFalse(auth.check_header("Bearer " + token)) self.assertFalse(auth.check_header("")) def test_invalid_auth_config(self) -> None: with self.assertRaises(ValueError): AuthConfig.parse("missing-colon") with self.assertRaises(ValueError): AuthConfig.parse(":password") self.assertIsNone(AuthConfig.parse("")) class CollectorTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() root = Path(self.tmp.name) self.proc = root / "proc" self.sys = root / "sys" self.proc.mkdir() self.sys.mkdir() (root / "etc").mkdir() (root / "etc" / "os-release").write_text( 'PRETTY_NAME="Test Linux 1.0"\nNAME="Test"\nVERSION="1.0"\n', encoding="utf-8" ) self._write_fixture(CPU_STAT_1, NET_DEV_1, DISKSTATS_1) (self.proc / "meminfo").write_text(MEMINFO, encoding="utf-8") (self.proc / "loadavg").write_text("1.50 1.00 0.50 2/100 1234\n", encoding="utf-8") (self.proc / "uptime").write_text("3600.00 1200.00\n", encoding="utf-8") (self.proc / "cpuinfo").write_text( "processor\t: 0\nmodel name\t: Test CPU\nphysical id\t: 0\ncore id\t: 0\n\n" "processor\t: 1\nmodel name\t: Test CPU\nphysical id\t: 0\ncore id\t: 1\n\n", encoding="utf-8", ) (self.proc / "mounts").write_text("", encoding="utf-8") (self.proc / "route").write_text( "Iface\tDestination\tGateway \tFlags\tRefCnt\tUse\tMetric\tMask\t\tMTU\tWindow\tIRTT\n" "eth0\t00000000\t0100000A\t0003\t0\t0\t100\t00000000\t0\t0\t0\n", encoding="utf-8", ) self.clock_values = iter([100.0, 101.0]) self.collector = SystemCollector( proc_root=str(self.proc), sys_root=str(self.sys), clock=lambda: next(self.clock_values), wall_clock=lambda: 1000.0, ) def _write_fixture(self, cpu: str, net: str, disk: str) -> None: (self.proc / "stat").write_text(cpu, encoding="utf-8") (self.proc / "net").mkdir(exist_ok=True) (self.proc / "net" / "dev").write_text(net, encoding="utf-8") (self.proc / "diskstats").write_text(disk, encoding="utf-8") def tearDown(self) -> None: self.tmp.cleanup() def test_deltas(self) -> None: first = self.collector.snapshot() self.assertEqual(first["cpu"]["percent"], 20.0) self.assertEqual(first["memory"]["percent"], 50.0) self.assertEqual(first["load"]["load1"], 1.5) self.assertEqual(first["host"]["uptime_seconds"], 3600.0) self._write_fixture(CPU_STAT_2, NET_DEV_2, DISKSTATS_2) second = self.collector.snapshot() self.assertEqual(second["cpu"]["percent"], 50.0) self.assertEqual(second["cpu"]["per_core"], [50.0, 50.0]) self.assertEqual(second["network"]["total_rx_bps"], 1000) self.assertEqual(second["network"]["total_tx_bps"], 500) self.assertEqual(second["disk_io"]["read_bps"], 1024000) self.assertEqual(second["disk_io"]["write_bps"], 512000) self.assertEqual(len(second["disk_io"]["devices"]), 1) self.assertEqual(second["disk_io"]["devices"][0]["device"], "sda") self.assertNotIn("sda1", [d["device"] for d in second["disk_io"]["devices"]]) def test_demo_service_history(self) -> None: service = MonitorService(DemoSource(), interval=1.0) service.bootstrap() payload = service.snapshot() self.assertIsNotNone(payload) assert payload is not None self.assertEqual(len(payload["history"]["cpu"]), 60) self.assertEqual(len(payload["history"]["disk_r"]), 60) self.assertEqual(payload["meta"]["demo"], True) json.dumps(payload, ensure_ascii=False) if __name__ == "__main__": unittest.main()