Andreas Wrede
ba96da9622
- tests/test_threshold.py: has proper pytest test functions - scripts/test_*.py: manual run scripts with no test functions Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
100 lines
2.8 KiB
Python
100 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test all plugins together.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
# Setup path
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from hbd.plugin import PluginRegistry, PluginLoader
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(name)s: %(message)s"
|
|
)
|
|
|
|
async def test_all_plugins():
|
|
"""Test loading all plugins."""
|
|
print("=" * 70)
|
|
print("Testing All Plugins")
|
|
print("=" * 70)
|
|
|
|
# Create registry and loader
|
|
registry = PluginRegistry()
|
|
loader = PluginLoader(registry)
|
|
|
|
# Configuration for plugins
|
|
config = {
|
|
"cpu_monitor": {
|
|
"interval": 30,
|
|
"per_core": False
|
|
},
|
|
"nagios_runner": {
|
|
"interval": 60,
|
|
"commands": [
|
|
{
|
|
"name": "test_ok",
|
|
"command": "echo 'OK - test passed | metric=100%;;;0;100'"
|
|
},
|
|
{
|
|
"name": "test_warning",
|
|
"command": "echo 'WARNING - test result | value=85%;80;90;0;100' && exit 1"
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
# Load plugins
|
|
plugin_dir = Path(__file__).parent / "hbd" / "plugins"
|
|
print(f"\n1. Loading plugins from: {plugin_dir}")
|
|
|
|
count = await loader.load_from_directory(plugin_dir, config)
|
|
print(f" ✓ Loaded {count} plugins")
|
|
|
|
# List loaded plugins
|
|
print(f"\n2. Loaded plugins:")
|
|
for plugin in registry.get_all():
|
|
print(f" - {plugin.name} v{plugin.version}")
|
|
print(f" Type: {plugin.__class__.__name__}")
|
|
print(f" Interval: {plugin.interval}s")
|
|
print(f" Description: {plugin.description}")
|
|
|
|
# Test collection for each plugin
|
|
print(f"\n3. Testing data collection:")
|
|
for plugin in registry.get_all():
|
|
print(f"\n {plugin.name}:")
|
|
try:
|
|
data = await plugin.collect()
|
|
print(f" ✓ Collected {len(data)} fields")
|
|
|
|
# Show sample of data
|
|
sample_count = min(5, len(data))
|
|
for key, value in list(data.items())[:sample_count]:
|
|
value_str = str(value)
|
|
if len(value_str) > 50:
|
|
value_str = value_str[:47] + "..."
|
|
print(f" {key}: {value_str}")
|
|
|
|
if len(data) > sample_count:
|
|
print(f" ... and {len(data) - sample_count} more fields")
|
|
|
|
except Exception as e:
|
|
print(f" ✗ Error: {e}")
|
|
|
|
# Cleanup
|
|
print(f"\n4. Cleanup...")
|
|
await loader.unload_all()
|
|
print(f" ✓ All plugins unloaded")
|
|
|
|
print(f"\n" + "=" * 70)
|
|
print(f"Successfully tested {count} plugins!")
|
|
print("=" * 70)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_all_plugins())
|