diff --git a/tools/health_check.py b/tools/health_check.py index 5cd0a613..ba6d9a95 100644 --- a/tools/health_check.py +++ b/tools/health_check.py @@ -149,7 +149,7 @@ def check_disk_usage(path: str = "/") -> Tuple[str, str, float]: return "WARNING", f"Cannot check: {e}", 0 -def check_memory_usage() -> Tuple[str, str, float]: +def _check_memory_linux() -> Optional[Tuple[str, str, float]]: try: with open("/proc/meminfo") as f: meminfo = {} @@ -174,11 +174,86 @@ def check_memory_usage() -> Tuple[str, str, float]: return "WARNING", f"{pct:.1f}% used", pct else: return "CRITICAL", f"{pct:.1f}% used", pct - except Exception as e: - return "WARNING", f"Cannot check: {e}", 0 + except Exception: + return None -def check_load_average() -> Tuple[str, str, float]: +def _check_memory_psutil() -> Optional[Tuple[str, str, float]]: + try: + import psutil + mem = psutil.virtual_memory() + pct = mem.percent + used_gb = mem.used / (1024 ** 3) + total_gb = mem.total / (1024 ** 3) + + if pct < MEMORY_THRESHOLD_WARNING: + return "OK", f"{pct:.1f}% used ({used_gb:.1f}GB/{total_gb:.1f}GB)", pct + elif pct < MEMORY_THRESHOLD_CRITICAL: + return "WARNING", f"{pct:.1f}% used", pct + else: + return "CRITICAL", f"{pct:.1f}% used", pct + except Exception: + return None + + +def _check_memory_platform() -> Tuple[str, str, float]: + try: + import subprocess + result = subprocess.run( + ["sysctl", "-n", "hw.memsize"], + capture_output=True, text=True, timeout=5 + ) + total = int(result.stdout.strip()) + + result = subprocess.run( + ["vm_stat"], + capture_output=True, text=True, timeout=5 + ) + lines = result.stdout.strip().split("\n") + page_size = 4096 + for line in lines: + if "page size of" in line: + page_size = int(line.split()[-2]) + break + + free_pages = 0 + inactive_pages = 0 + speculative_pages = 0 + for line in lines[1:]: + if line.startswith("Pages free"): + free_pages = int(line.split()[-1].rstrip(".")) + elif line.startswith("Pages inactive"): + inactive_pages = int(line.split()[-1].rstrip(".")) + elif line.startswith("Pages speculative"): + speculative_pages = int(line.split()[-1].rstrip(".")) + + available = (free_pages + inactive_pages + speculative_pages) * page_size + used = total - available + pct = (used / total) * 100 if total > 0 else 0 + + if pct < MEMORY_THRESHOLD_WARNING: + return "OK", f"{pct:.1f}% used ({used // (1024**3)}GB/{total // (1024**3)}GB)", pct + elif pct < MEMORY_THRESHOLD_CRITICAL: + return "WARNING", f"{pct:.1f}% used", pct + else: + return "CRITICAL", f"{pct:.1f}% used", pct + except Exception: + return "WARNING", "Cannot check memory: no fallback available", 0 + + +def check_memory_usage() -> Tuple[str, str, float]: + result = _check_memory_linux() + if result is not None: + return result + + result = _check_memory_psutil() + if result is not None: + return result + + return _check_memory_platform() + + +def _check_load_linux() -> Optional[Tuple[str, str, float]]: try: with open("/proc/loadavg") as f: parts = f.read().strip().split() @@ -192,8 +267,36 @@ def check_load_average() -> Tuple[str, str, float]: return "WARNING", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load else: return "CRITICAL", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load - except Exception as e: - return "WARNING", f"Cannot check: {e}", 0 + except Exception: + return None + + +def _check_load_getloadavg() -> Optional[Tuple[str, str, float]]: + try: + load1, load5, load15 = os.getloadavg() + cpu_count = os.cpu_count() or 1 + load_pct = (load1 / cpu_count) * 100 + + if load_pct < 70: + return "OK", f"Load: {load1} ({load_pct:.0f}% of {cpu_count} cores)", load1 + elif load_pct < 90: + return "WARNING", f"Load: {load1} ({load_pct:.0f}% of {cpu_count} cores)", load1 + else: + return "CRITICAL", f"Load: {load1} ({load_pct:.0f}% of {cpu_count} cores)", load1 + except Exception: + return None + + +def check_load_average() -> Tuple[str, str, float]: + result = _check_load_linux() + if result is not None: + return result + + result = _check_load_getloadavg() + if result is not None: + return result + + return "WARNING", "Cannot check load average: no fallback available", 0 # --------------------------------------------------------------------------- diff --git a/tools/test_health_check.py b/tools/test_health_check.py new file mode 100644 index 00000000..1ffeb89a --- /dev/null +++ b/tools/test_health_check.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +"""Tests for cross-platform health check fallbacks.""" + +import os +import sys +import unittest +from unittest.mock import patch, mock_open + +sys.path.insert(0, os.path.dirname(__file__)) + +from health_check import ( + check_memory_usage, + check_load_average, + _check_memory_linux, + _check_load_linux, + MEMORY_THRESHOLD_WARNING, + MEMORY_THRESHOLD_CRITICAL, +) + + +class TestCheckMemoryLinux(unittest.TestCase): + def test_linux_meminfo_parses_correctly(self): + meminfo_content = ( + "MemTotal: 16384000 kB\n" + "MemAvailable: 8192000 kB\n" + "MemFree: 4096000 kB\n" + ) + with patch("builtins.open", mock_open(read_data=meminfo_content)): + result = _check_memory_linux() + self.assertIsNotNone(result) + status, detail, pct = result + self.assertEqual(status, "OK") + self.assertIn("50.0% used", detail) + + def test_linux_meminfo_critical(self): + meminfo_content = ( + "MemTotal: 16384000 kB\n" + "MemAvailable: 819200 kB\n" + ) + with patch("builtins.open", mock_open(read_data=meminfo_content)): + result = _check_memory_linux() + self.assertIsNotNone(result) + status, detail, pct = result + self.assertEqual(status, "CRITICAL") + + def test_linux_meminfo_file_missing(self): + with patch("builtins.open", side_effect=FileNotFoundError): + result = _check_memory_linux() + self.assertIsNone(result) + + def test_fallback_when_proc_missing(self): + with patch("builtins.open", side_effect=FileNotFoundError): + result = check_memory_usage() + self.assertIsNotNone(result) + status, detail, pct = result + self.assertIn(status, ("OK", "WARNING", "CRITICAL")) + + +class TestCheckLoadLinux(unittest.TestCase): + def test_linux_loadavg_parses_correctly(self): + loadavg_content = "1.50 2.00 3.00 1/400 12345\n" + with patch("builtins.open", mock_open(read_data=loadavg_content)): + result = _check_load_linux() + self.assertIsNotNone(result) + status, detail, load = result + self.assertEqual(status, "OK") + self.assertIn("Load: 1.5", detail) + + def test_linux_loadavg_critical(self): + loadavg_content = "16.00 15.00 14.00 1/400 12345\n" + with patch("builtins.open", mock_open(read_data=loadavg_content)): + result = _check_load_linux() + self.assertIsNotNone(result) + status, detail, load = result + self.assertEqual(status, "CRITICAL") + + def test_linux_loadavg_file_missing(self): + with patch("builtins.open", side_effect=FileNotFoundError): + result = _check_load_linux() + self.assertIsNone(result) + + def test_fallback_when_proc_missing(self): + with patch("builtins.open", side_effect=FileNotFoundError): + with patch("os.getloadavg", return_value=(2.0, 3.0, 4.0)): + result = check_load_average() + self.assertIsNotNone(result) + status, detail, load = result + self.assertIn(status, ("OK", "WARNING", "CRITICAL")) + self.assertIn("Load: 2.0", detail) + + +class TestCheckMemoryPsutil(unittest.TestCase): + def test_psutil_fallback(self): + mock_mem = unittest.mock.MagicMock() + mock_mem.percent = 45.0 + mock_mem.used = 7 * 1024 ** 3 + mock_mem.total = 16 * 1024 ** 3 + + with patch("builtins.open", side_effect=FileNotFoundError): + with patch.dict("sys.modules", {"psutil": unittest.mock.MagicMock(psutil=unittest.mock.MagicMock(virtual_memory=mock_mem))}): + import importlib + import health_check + original = health_check._check_memory_psutil + health_check._check_memory_psutil = lambda: ("OK", "45.0% used (7.0GB/16.0GB)", 45.0) + try: + result = check_memory_usage() + self.assertEqual(result[0], "OK") + finally: + health_check._check_memory_psutil = original + + +class TestCheckLoadGetloadavg(unittest.TestCase): + def test_getloadavg_fallback(self): + with patch("builtins.open", side_effect=FileNotFoundError): + with patch("os.getloadavg", return_value=(1.0, 2.0, 3.0)): + result = check_load_average() + self.assertIsNotNone(result) + status, detail, load = result + self.assertIn(status, ("OK", "WARNING", "CRITICAL")) + self.assertIn("Load: 1.0", detail) + + def test_getloadavg_critical(self): + with patch("builtins.open", side_effect=FileNotFoundError): + with patch("os.getloadavg", return_value=(20.0, 18.0, 16.0)): + result = check_load_average() + self.assertEqual(result[0], "CRITICAL") + + +class TestBackwardCompatibility(unittest.TestCase): + def test_json_output_format(self): + meminfo_content = ( + "MemTotal: 16384000 kB\n" + "MemAvailable: 8192000 kB\n" + ) + with patch("builtins.open", mock_open(read_data=meminfo_content)): + status, detail, pct = check_memory_usage() + self.assertIsInstance(pct, float) + self.assertGreaterEqual(pct, 0) + self.assertLessEqual(pct, 100) + + def test_load_returns_float(self): + loadavg_content = "1.50 2.00 3.00 1/400 12345\n" + with patch("builtins.open", mock_open(read_data=loadavg_content)): + status, detail, load = check_load_average() + self.assertIsInstance(load, float) + self.assertGreaterEqual(load, 0) + + +if __name__ == "__main__": + unittest.main()