heartbeat/test_nagios.py

#!/usr/bin/env python3
"""
Test the Nagios Runner Plugin.
"""

import asyncio
import logging
from pathlib import Path

# Setup path
import sys
sys.path.insert(0, str(Path(__file__).parent))

from hbd.plugins.nagios_runner import NagiosRunnerPlugin, NAGIOS_OK, NAGIOS_WARNING

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(name)s %(levelname)s: %(message)s"
)

async def test_nagios_runner():
    """Test Nagios runner plugin."""
    print("=" * 70)
    print("Testing Nagios Runner Plugin")
    print("=" * 70)
    
    # Create test configuration with simple shell commands
    # These mimic Nagios plugin output format
    config = {
        "interval": 60,
        "timeout": 10,
        "commands": [
            {
                "name": "check_uptime",
                "command": "echo 'OK - uptime is 5 days | uptime=432000s;;;0'"
            },
            {
                "name": "check_memory",
                "command": "echo 'OK - Memory usage 45% | memory=45%;80;90;0;100'"
            },
            {
                "name": "check_cpu",
                "command": "echo 'WARNING - CPU load high | load1=5.2;5.0;10.0;0 load5=4.8;4.0;8.0;0 load15=3.2;3.0;6.0;0' && exit 1"
            },
            {
                "name": "check_disk",
                "command": "echo 'OK - Disk usage 62% | /=62%;80;90;0;100 /home=45%;80;90;0;100'"
            }
        ]
    }
    
    print("\n1. Creating Nagios Runner plugin with test configuration")
    print(f"   Configured {len(config['commands'])} test commands")
    
    plugin = NagiosRunnerPlugin(config)
    
    print(f"\n2. Initializing plugin...")
    initialized = await plugin.initialize()
    print(f"   Initialized: {initialized}")
    
    if not initialized:
        print("   ERROR: Plugin failed to initialize!")
        return
    
    print(f"\n3. Collecting metrics from Nagios plugins...")
    data = await plugin.collect()
    
    print(f"   ✓ Collected {len(data)} data points")
    
    print(f"\n4. Results:")
    print(f"   Data points collected: {len(data)}")
    
    # Show individual plugin results
    print(f"\n5. Individual Plugin Results:")
    for cmd_config in config["commands"]:
        name = cmd_config["name"]
        status = data.get(f"{name}_status", "N/A")
        status_code = data.get(f"{name}_status_code", "N/A")
        output = data.get(f"{name}_output", "N/A")
        
        print(f"\n   {name}:")
        print(f"     Status: {status} (code: {status_code})")
        print(f"     Output: {output}")
        
        # Show performance data if present
        perf_keys = [k for k in data.keys() if k.startswith(f"{name}_") and 
                     k not in [f"{name}_status", f"{name}_status_code", f"{name}_output"]]
        if perf_keys:
            print(f"     Performance Data:")
            for key in perf_keys:
                metric_name = key.replace(f"{name}_", "")
                print(f"       {metric_name}: {data[key]}")
    
    print(f"\n6. Testing Nagios plugin detection (if available)...")
    
    # Try to find actual Nagios plugins on the system
    common_nagios_paths = [
        "/usr/lib/nagios/plugins",
        "/usr/local/nagios/libexec",
        "/usr/lib64/nagios/plugins"
    ]
    
    nagios_plugin_dir = None
    for path in common_nagios_paths:
        if Path(path).exists():
            nagios_plugin_dir = Path(path)
            print(f"   ✓ Found Nagios plugins at: {nagios_plugin_dir}")
            break
    
    if nagios_plugin_dir:
        # Try check_users if it exists
        check_users = nagios_plugin_dir / "check_users"
        if check_users.exists():
            print(f"\n   Testing real Nagios plugin: check_users")
            real_config = {
                "commands": [{
                    "name": "users",
                    "command": f"{check_users} -w 10 -c 20"
                }]
            }
            real_plugin = NagiosRunnerPlugin(real_config)
            await real_plugin.initialize()
            real_data = await real_plugin.collect()
            
            print(f"     Status: {real_data.get('users_status')}")
            print(f"     Output: {real_data.get('users_output')}")
            
            # Show any performance data
            for key in real_data:
                if key.startswith("users_") and "status" not in key and "output" not in key:
                    print(f"     {key}: {real_data[key]}")
        else:
            print(f"   check_users not found at {check_users}")
    else:
        print(f"   No Nagios plugins directory found")
        print(f"   Install nagios-plugins to test with real plugins:")
        print(f"     sudo apt-get install nagios-plugins  # Debian/Ubuntu")
        print(f"     sudo yum install nagios-plugins-all   # RHEL/CentOS")
    
    print(f"\n7. Cleanup...")
    await plugin.cleanup()
    print(f"   ✓ Cleanup complete")
    
    print(f"\n" + "=" * 70)
    print("Test complete!")
    print("=" * 70)

if __name__ == "__main__":
    asyncio.run(test_nagios_runner())