100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
import asyncio
|
|
import logging
|
|
import os
|
|
import stat
|
|
|
|
from hbd.client.plugins.nagios_runner import (
|
|
NagiosRunnerPlugin,
|
|
NAGIOS_OK,
|
|
NAGIOS_WARNING,
|
|
NAGIOS_CRITICAL,
|
|
NAGIOS_UNKNOWN,
|
|
)
|
|
|
|
|
|
def test_no_commands_sets_skip_reason():
|
|
plugin = NagiosRunnerPlugin(config={"commands": []})
|
|
result = asyncio.run(plugin.initialize())
|
|
assert result is False
|
|
assert plugin.skip_reason is not None
|
|
assert "nagios_runner.commands" in plugin.skip_reason
|
|
|
|
|
|
def test_stderr_used_when_stdout_empty(tmp_path):
|
|
script = tmp_path / "check_err.sh"
|
|
script.write_text("#!/bin/sh\necho 'error from stderr' >&2\nexit 2\n")
|
|
script.chmod(script.stat().st_mode | stat.S_IEXEC)
|
|
|
|
config = {"commands": [{"name": "t", "command": str(script)}], "timeout": 5}
|
|
plugin = NagiosRunnerPlugin(config=config)
|
|
asyncio.run(plugin.initialize())
|
|
data = asyncio.run(plugin._collect_metrics())
|
|
|
|
assert "error from stderr" in data["t_output"]
|
|
assert data["t_status_code"] == NAGIOS_CRITICAL
|
|
|
|
|
|
def test_stderr_appended_when_both_present(tmp_path):
|
|
script = tmp_path / "check_both.sh"
|
|
script.write_text("#!/bin/sh\necho 'OK - all good'\necho 'extra detail' >&2\nexit 0\n")
|
|
script.chmod(script.stat().st_mode | stat.S_IEXEC)
|
|
|
|
config = {"commands": [{"name": "t", "command": str(script)}], "timeout": 5}
|
|
plugin = NagiosRunnerPlugin(config=config)
|
|
asyncio.run(plugin.initialize())
|
|
data = asyncio.run(plugin._collect_metrics())
|
|
|
|
assert "OK - all good" in data["t_output"]
|
|
assert "extra detail" in data["t_output"]
|
|
assert data["t_status_code"] == NAGIOS_OK
|
|
|
|
|
|
def test_negative_returncode_maps_to_unknown():
|
|
# kill -9 $$ kills the shell itself; asyncio sees returncode -9
|
|
config = {"commands": [{"name": "t", "command": "kill -9 $$"}], "timeout": 5}
|
|
plugin = NagiosRunnerPlugin(config=config)
|
|
asyncio.run(plugin.initialize())
|
|
data = asyncio.run(plugin._collect_metrics())
|
|
|
|
assert data["t_status_code"] == NAGIOS_UNKNOWN
|
|
assert "signal" in data["t_output"].lower()
|
|
|
|
|
|
def test_absolute_path_not_found_warns(caplog):
|
|
fake_cmd = "/nonexistent_hbc_test_path/check_something"
|
|
config = {"commands": [{"name": "t", "command": fake_cmd}]}
|
|
plugin = NagiosRunnerPlugin(config=config)
|
|
|
|
with caplog.at_level(logging.WARNING, logger="plugin.nagios_runner"):
|
|
asyncio.run(plugin.initialize())
|
|
|
|
assert any("not found" in r.message for r in caplog.records)
|
|
|
|
|
|
def test_absolute_path_not_executable_warns(caplog, tmp_path):
|
|
non_exec = tmp_path / "check_test"
|
|
non_exec.write_text("#!/bin/sh\necho OK\n")
|
|
non_exec.chmod(0o644) # readable but not executable
|
|
|
|
config = {"commands": [{"name": "t", "command": str(non_exec)}]}
|
|
plugin = NagiosRunnerPlugin(config=config)
|
|
|
|
with caplog.at_level(logging.WARNING, logger="plugin.nagios_runner"):
|
|
asyncio.run(plugin.initialize())
|
|
|
|
assert any("not executable" in r.message for r in caplog.records)
|
|
|
|
|
|
def test_relative_path_not_checked(caplog):
|
|
# Relative paths (resolved via PATH) must not generate warnings
|
|
config = {"commands": [{"name": "t", "command": "echo OK"}]}
|
|
plugin = NagiosRunnerPlugin(config=config)
|
|
|
|
with caplog.at_level(logging.WARNING, logger="plugin.nagios_runner"):
|
|
asyncio.run(plugin.initialize())
|
|
|
|
assert not any(
|
|
"not found" in r.message or "not executable" in r.message
|
|
for r in caplog.records
|
|
)
|