84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
import asyncio
|
|
import logging
|
|
import textwrap
|
|
|
|
from hbd.client.plugin import PluginLoader, PluginRegistry
|
|
|
|
|
|
def test_plugin_skip_reason_defaults_none(tmp_path):
|
|
plugin_code = textwrap.dedent("""
|
|
from hbd.client.plugin import MonitorPlugin
|
|
|
|
class MinimalPlugin(MonitorPlugin):
|
|
name = "minimal"
|
|
version = "1.0.0"
|
|
interval = 60
|
|
|
|
async def initialize(self):
|
|
return True
|
|
|
|
async def _collect_metrics(self):
|
|
return {}
|
|
""")
|
|
(tmp_path / "minimal.py").write_text(plugin_code)
|
|
registry = PluginRegistry()
|
|
loader = PluginLoader(registry)
|
|
asyncio.run(loader.load_from_directory(tmp_path))
|
|
plugin = registry.get("minimal")
|
|
assert plugin is not None
|
|
assert plugin.skip_reason is None
|
|
|
|
|
|
def test_loader_logs_info_when_skip_reason_set(tmp_path, caplog):
|
|
plugin_code = textwrap.dedent("""
|
|
from hbd.client.plugin import MonitorPlugin
|
|
|
|
class SkippablePlugin(MonitorPlugin):
|
|
name = "skippable"
|
|
version = "1.0.0"
|
|
interval = 60
|
|
|
|
async def initialize(self):
|
|
self.skip_reason = "not configured in yaml"
|
|
return False
|
|
|
|
async def _collect_metrics(self):
|
|
return {}
|
|
""")
|
|
(tmp_path / "skippable.py").write_text(plugin_code)
|
|
registry = PluginRegistry()
|
|
loader = PluginLoader(registry)
|
|
|
|
with caplog.at_level(logging.INFO, logger="plugin.loader"):
|
|
count = asyncio.run(loader.load_from_directory(tmp_path))
|
|
|
|
assert count == 0
|
|
assert any("skipped: not configured in yaml" in r.message for r in caplog.records)
|
|
assert not any("failed initialization" in r.message for r in caplog.records)
|
|
|
|
|
|
def test_loader_logs_warning_when_no_skip_reason(tmp_path, caplog):
|
|
plugin_code = textwrap.dedent("""
|
|
from hbd.client.plugin import MonitorPlugin
|
|
|
|
class FailPlugin(MonitorPlugin):
|
|
name = "fail"
|
|
version = "1.0.0"
|
|
interval = 60
|
|
|
|
async def initialize(self):
|
|
return False
|
|
|
|
async def _collect_metrics(self):
|
|
return {}
|
|
""")
|
|
(tmp_path / "fail_plugin.py").write_text(plugin_code)
|
|
registry = PluginRegistry()
|
|
loader = PluginLoader(registry)
|
|
|
|
with caplog.at_level(logging.WARNING, logger="plugin.loader"):
|
|
count = asyncio.run(loader.load_from_directory(tmp_path))
|
|
|
|
assert count == 0
|
|
assert any("failed initialization" in r.message for r in caplog.records)
|