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>
120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for plugin system.
|
|
"""
|
|
|
|
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.DEBUG,
|
|
format="%(asctime)s %(name)s %(levelname)s: %(message)s"
|
|
)
|
|
|
|
async def test_plugins():
|
|
"""Test plugin loading and collection."""
|
|
print("=" * 60)
|
|
print("Testing Plugin System")
|
|
print("=" * 60)
|
|
|
|
# Create registry and loader
|
|
registry = PluginRegistry()
|
|
loader = PluginLoader(registry)
|
|
|
|
# Load plugins
|
|
plugin_dir = Path(__file__).parent / "hbd" / "plugins"
|
|
print(f"\n1. Loading plugins from: {plugin_dir}")
|
|
|
|
if not plugin_dir.exists():
|
|
print(f" ERROR: Plugin directory does not exist!")
|
|
return
|
|
|
|
count = await loader.load_from_directory(plugin_dir)
|
|
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} ({plugin.__class__.__name__})")
|
|
print(f" Description: {plugin.description}")
|
|
print(f" Interval: {plugin.interval}s")
|
|
|
|
# Test InfoPlugins
|
|
print(f"\n3. Testing InfoPlugins (collect once):")
|
|
from hbd.plugin import InfoPlugin
|
|
for plugin in registry.get_by_type(InfoPlugin):
|
|
print(f"\n Collecting {plugin.name}...")
|
|
try:
|
|
data = await plugin.collect()
|
|
print(f" ✓ Success! Got {len(data)} fields")
|
|
for key, value in list(data.items())[:5]: # Show first 5 fields
|
|
print(f" {key}: {value}")
|
|
if len(data) > 5:
|
|
print(f" ... and {len(data) - 5} more fields")
|
|
except Exception as e:
|
|
print(f" ✗ Error: {e}")
|
|
|
|
# Test MonitorPlugins
|
|
print(f"\n4. Testing MonitorPlugins (periodic collection):")
|
|
from hbd.plugin import MonitorPlugin
|
|
for plugin in registry.get_by_type(MonitorPlugin):
|
|
print(f"\n Collecting {plugin.name}...")
|
|
try:
|
|
data = await plugin.collect()
|
|
print(f" ✓ Success! Got {len(data)} fields")
|
|
for key, value in list(data.items())[:8]: # Show first 8 fields
|
|
print(f" {key}: {value}")
|
|
if len(data) > 8:
|
|
print(f" ... and {len(data) - 8} more fields")
|
|
except Exception as e:
|
|
print(f" ✗ Error: {e}")
|
|
|
|
# Test protocol encoding
|
|
print(f"\n5. Testing protocol encoding:")
|
|
from hbd.proto import dicttos, stodict
|
|
|
|
# Create sample plugin data
|
|
test_data = {
|
|
"plugin": "test_plugin",
|
|
"cpu_percent": 42.5,
|
|
"memory_mb": 1024,
|
|
"processes": 156,
|
|
"load_avg": [1.2, 0.8, 0.5],
|
|
"disk_info": {"sda": {"used": 50, "total": 100}}
|
|
}
|
|
|
|
print(f" Original data: {test_data}")
|
|
|
|
# Encode
|
|
encoded = dicttos("PLG", test_data)
|
|
print(f" Encoded ({len(encoded)} bytes): {encoded[:50]}...")
|
|
|
|
# Decode
|
|
decoded = stodict(encoded)
|
|
print(f" Decoded: {decoded}")
|
|
|
|
# Verify
|
|
if decoded.get("ID") == "PLG" and decoded.get("plugin") == "test_plugin":
|
|
print(f" ✓ Protocol encoding/decoding works!")
|
|
else:
|
|
print(f" ✗ Protocol encoding/decoding failed!")
|
|
|
|
# Cleanup
|
|
print(f"\n6. Cleaning up...")
|
|
await loader.unload_all()
|
|
print(f" ✓ Cleanup complete")
|
|
|
|
print(f"\n" + "=" * 60)
|
|
print("Test complete!")
|
|
print("=" * 60)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_plugins())
|