#!/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" Overall Status: {data.get('overall_status')} (code: {data.get('overall_status_code')})") print(f" Plugins Executed: {data.get('plugin_count')}") # 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())