Fix #1: [$50 BOUNTY] [Python] Add cross-platform health check fallbacks

tests/test_health_check.py

@@ -0,0 +1,72 @@
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+ROOT = Path(__file__).resolve().parents[1]
+HEALTH_CHECK_PATH = ROOT / "tools" / "health_check.py"
+
+spec = importlib.util.spec_from_file_location("health_check", HEALTH_CHECK_PATH)
+health_check = importlib.util.module_from_spec(spec)
+assert spec.loader is not None
+spec.loader.exec_module(health_check)
+
+
+class HealthCheckSystemFallbackTests(unittest.TestCase):
+    def test_proc_memory_provider_parses_meminfo(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            meminfo_path = Path(tmpdir) / "meminfo"
+            meminfo_path.write_text(
+                "MemTotal:       2048000 kB\n"
+                "MemAvailable:   512000 kB\n",
+                encoding="utf-8",
+            )
+
+            total, available = health_check._memory_from_proc(str(meminfo_path))
+
+        self.assertEqual(total, 2048000 * 1024)
+        self.assertEqual(available, 512000 * 1024)
+
+    def test_memory_check_uses_non_proc_fallback(self):
+        with mock.patch.object(health_check, "_memory_from_proc", side_effect=FileNotFoundError), \
+             mock.patch.object(health_check, "_memory_from_sysconf", return_value=(100, 25)), \
+             mock.patch.object(health_check, "_memory_from_macos_vm_stat", return_value=None), \
+             mock.patch.object(health_check, "_memory_from_windows", return_value=None):
+            status, detail, pct = health_check.check_memory_usage()
+
+        self.assertEqual(status, "OK")
+        self.assertEqual(pct, 75)
+        self.assertIn("75.0% used", detail)
+
+    def test_load_check_uses_non_proc_fallback(self):
+        with mock.patch.object(health_check, "_load_from_proc", side_effect=FileNotFoundError), \
+             mock.patch.object(health_check, "_load_from_getloadavg", return_value=1.0), \
+             mock.patch.object(health_check.os, "cpu_count", return_value=2):
+            status, detail, load = health_check.check_load_average()
+
+        self.assertEqual(status, "OK")
+        self.assertEqual(load, 1.0)
+        self.assertEqual(detail, "Load: 1.0 (50% of 2 cores)")
+
+    def test_load_check_reports_platform_unavailable_without_proc_path(self):
+        with mock.patch.object(health_check, "_load_from_proc", return_value=None), \
+             mock.patch.object(health_check, "_load_from_getloadavg", return_value=None):
+            status, detail, load = health_check.check_load_average()
+
+        self.assertEqual(status, "WARNING")
+        self.assertEqual(load, 0)
+        self.assertNotIn("/proc/loadavg", detail)
+
+    def test_disk_check_uses_cross_platform_disk_usage(self):
+        with mock.patch.object(health_check.shutil, "disk_usage", return_value=(100, 85, 15)):
+            status, detail, pct = health_check.check_disk_usage("/tmp")
+
+        self.assertEqual(status, "WARNING")
+        self.assertEqual(pct, 85)
+        self.assertIn("85.0% used", detail)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/health_check.py

@@ -31,8 +31,12 @@
 """
 
 import argparse
+import ctypes
 import json
 import os
+import platform
+import re
+import shutil
 import socket
 import ssl
 import subprocess
@@ -133,9 +137,7 @@ def check_certificate_expiry(host: str, port: int = 443) -> Tuple[str, str, int]
 
 def check_disk_usage(path: str = "/") -> Tuple[str, str, float]:
     try:
-        stat = os.statvfs(path)
-        total = stat.f_frsize * stat.f_blocks
-        free = stat.f_frsize * stat.f_bavail
+        total, used, free = shutil.disk_usage(path)
         used = total - free
         pct = (used / total) * 100
 
@@ -149,22 +151,150 @@ def check_disk_usage(path: str = "/") -> Tuple[str, str, float]:
         return "WARNING", f"Cannot check: {e}", 0
 
 
+def _memory_from_proc(meminfo_path: str = "/proc/meminfo") -> Optional[Tuple[int, int]]:
+    meminfo = {}
+    with open(meminfo_path) as f:
+        for line in f:
+            parts = line.split(":")
+            if len(parts) == 2:
+                key = parts[0].strip()
+                value = parts[1].strip().replace(" kB", "")
+                try:
+                    meminfo[key] = int(value) * 1024
+                except ValueError:
+                    pass
+
+    total = meminfo.get("MemTotal", 0)
+    available = meminfo.get("MemAvailable", 0)
+    if total <= 0 or available < 0 or available > total:
+        return None
+    return total, available
+
+
+def _memory_from_sysconf() -> Optional[Tuple[int, int]]:
+    if not hasattr(os, "sysconf"):
+        return None
+
+    try:
+        page_size = os.sysconf("SC_PAGE_SIZE")
+        total_pages = os.sysconf("SC_PHYS_PAGES")
+        available_pages = os.sysconf("SC_AVPHYS_PAGES")
+    except (OSError, ValueError, AttributeError):
+        return None
+
+    total = int(page_size) * int(total_pages)
+    available = int(page_size) * int(available_pages)
+    if total <= 0 or available < 0 or available > total:
+        return None
+    return total, available
+
+
+def _memory_from_macos_vm_stat() -> Optional[Tuple[int, int]]:
+    if platform.system() != "Darwin":
+        return None
+
+    try:
+        vm_result = subprocess.run(
+            ["vm_stat"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+            check=True,
+        )
+    except (OSError, subprocess.SubprocessError, ValueError):
+        return None
+
+    page_size_match = re.search(r"page size of (\d+) bytes", vm_result.stdout)
+    if not page_size_match:
+        return None
+    page_size = int(page_size_match.group(1))
+
+    total = 0
+    try:
+        total = int(os.sysconf("SC_PHYS_PAGES")) * page_size
+    except (OSError, ValueError, AttributeError):
+        try:
+            total_result = subprocess.run(
+                ["sysctl", "-n", "hw.memsize"],
+                capture_output=True,
+                text=True,
+                timeout=5,
+                check=True,
+            )
+            total = int(total_result.stdout.strip())
+        except (OSError, subprocess.SubprocessError, ValueError):
+            return None
+
+    pages = {}
+    for line in vm_result.stdout.splitlines():
+        match = re.match(r"Pages ([^:]+):\s+([0-9.]+)", line.strip())
+        if match:
+            pages[match.group(1).lower()] = int(match.group(2).replace(".", ""))
+
+    available_pages = (
+        pages.get("free", 0)
+        + pages.get("inactive", 0)
+        + pages.get("speculative", 0)
+    )
+    available = available_pages * page_size
+    if total <= 0 or available < 0 or available > total:
+        return None
+    return total, available
+
+
+def _memory_from_windows() -> Optional[Tuple[int, int]]:
+    if platform.system() != "Windows":
+        return None
+
+    class MEMORYSTATUSEX(ctypes.Structure):
+        _fields_ = [
+            ("dwLength", ctypes.c_ulong),
+            ("dwMemoryLoad", ctypes.c_ulong),
+            ("ullTotalPhys", ctypes.c_ulonglong),
+            ("ullAvailPhys", ctypes.c_ulonglong),
+            ("ullTotalPageFile", ctypes.c_ulonglong),
+            ("ullAvailPageFile", ctypes.c_ulonglong),
+            ("ullTotalVirtual", ctypes.c_ulonglong),
+            ("ullAvailVirtual", ctypes.c_ulonglong),
+            ("sullAvailExtendedVirtual", ctypes.c_ulonglong),
+        ]
+
+    status = MEMORYSTATUSEX()
+    status.dwLength = ctypes.sizeof(MEMORYSTATUSEX)
+    try:
+        if not ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(status)):
+            return None
+    except AttributeError:
+        return None
+
+    if status.ullTotalPhys <= 0 or status.ullAvailPhys > status.ullTotalPhys:
+        return None
+    return int(status.ullTotalPhys), int(status.ullAvailPhys)
+
+
+def _get_memory_totals() -> Optional[Tuple[int, int]]:
+    for provider in (
+        _memory_from_proc,
+        _memory_from_sysconf,
+        _memory_from_macos_vm_stat,
+        _memory_from_windows,
+    ):
+        try:
+            totals = provider()
+        except Exception:
+            totals = None
+        if totals is not None:
+            return totals
+    return None
+
+
 def check_memory_usage() -> Tuple[str, str, float]:
     try:
-        with open("/proc/meminfo") as f:
-            meminfo = {}
-            for line in f:
-                parts = line.split(":")
-                if len(parts) == 2:
-                    key = parts[0].strip()
-                    value = parts[1].strip().replace(" kB", "")
-                    try:
-                        meminfo[key] = int(value) * 1024
-                    except ValueError:
-                        pass
-
-        total = meminfo.get("MemTotal", 0)
-        available = meminfo.get("MemAvailable", 0)
+        totals = _get_memory_totals()
+        if totals is None:
+            return "WARNING", "Cannot check: no supported memory provider", 0
+
+        total, available = totals
         used = total - available
         pct = (used / total) * 100 if total > 0 else 0
 
@@ -178,20 +308,46 @@ def check_memory_usage() -> Tuple[str, str, float]:
         return "WARNING", f"Cannot check: {e}", 0
 
 
+def _load_from_proc(loadavg_path: str = "/proc/loadavg") -> Optional[float]:
+    with open(loadavg_path) as f:
+        parts = f.read().strip().split()
+        if not parts:
+            return None
+        return float(parts[0])
+
+
+def _load_from_getloadavg() -> Optional[float]:
+    if not hasattr(os, "getloadavg"):
+        return None
+    return float(os.getloadavg()[0])
+
+
+def _get_load_average() -> Optional[float]:
+    for provider in (_load_from_proc, _load_from_getloadavg):
+        try:
+            load = provider()
+        except (OSError, ValueError):
+            load = None
+        if load is not None:
+            return load
+    return None
+
+
 def check_load_average() -> Tuple[str, str, float]:
     try:
-        with open("/proc/loadavg") as f:
-            parts = f.read().strip().split()
-            load = float(parts[0])
-            cpu_count = os.cpu_count() or 1
-            load_pct = (load / cpu_count) * 100
-
-            if load_pct < 70:
-                return "OK", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
-            elif load_pct < 90:
-                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
+        load = _get_load_average()
+        if load is None:
+            return "WARNING", "Cannot check: load average is not available on this platform", 0
+
+        cpu_count = os.cpu_count() or 1
+        load_pct = (load / cpu_count) * 100
+
+        if load_pct < 70:
+            return "OK", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
+        elif load_pct < 90:
+            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