Skip to content
Snippets Groups Projects
Commit 4f1a6a6c authored by nimrod's avatar nimrod
Browse files

Progress.

- More tests, now using mocks.
- Silence some linter warnings.
- Start thinking about parsing perfdata.
parent 343ea3da
Branches
Tags
No related merge requests found
Pipeline #1923 passed
...@@ -62,6 +62,7 @@ repos: ...@@ -62,6 +62,7 @@ repos:
additional_dependencies: additional_dependencies:
- bandit - bandit
- pyroma - pyroma
exclude: ^tests/.*
- repo: https://gitlab.com/pycqa/flake8.git - repo: https://gitlab.com/pycqa/flake8.git
rev: 3.9.2 rev: 3.9.2
......
"""My Nagios plugin wrapper.""" """My Nagios plugin wrapper."""
# pylint: disable=logging-fstring-interpolation
__version__ = "0.1.0" __version__ = "0.1.0"
import argparse import argparse
......
"""Nagios check implementation. """Nagios check implementation.
Based on the specification from Based on the specification from
https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/pluginapi.html""" https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/pluginapi.html
"""
# pylint: disable=logging-fstring-interpolation
import collections
import enum import enum
import logging import logging
import subprocess import subprocess # nosec
DEFAULT_TIMEOUT = 10 # In seconds. DEFAULT_TIMEOUT = 10 # In seconds.
...@@ -19,7 +22,13 @@ class NagiosCode(enum.IntEnum): ...@@ -19,7 +22,13 @@ class NagiosCode(enum.IntEnum):
UNKNOWN = 3 UNKNOWN = 3
PerfData = collections.namedtuple(
"PerfData", ["name", "value", "unit", "warning", "critical", "min", "max"]
)
class Check: class Check:
# pylint: disable=invalid-name
"""Nagios check using a plugin.""" """Nagios check using a plugin."""
Command = None Command = None
...@@ -32,7 +41,7 @@ class Check: ...@@ -32,7 +41,7 @@ class Check:
stderr = None stderr = None
_stdout = None _stdout = None
def __init__(self, command, args): def __init__(self, command, args=None):
if not command: if not command:
raise ValueError("Command is empty.") raise ValueError("Command is empty.")
self.Command = command self.Command = command
...@@ -55,7 +64,7 @@ class Check: ...@@ -55,7 +64,7 @@ class Check:
] + self.Arguments ] + self.Arguments
logging.info(f"Running command {_cmd}.") logging.info(f"Running command {_cmd}.")
try: try:
proc = subprocess.run( proc = subprocess.run( # nosec
[ [
self.Command, self.Command,
] ]
......
import subprocess
import unittest.mock
import pytest import pytest
from mnpw import nagios from mnpw import nagios
OUTPUT_1 = """DISK OK - free space: / 3326 MB (56%);"""
OUTPUT_2 = (
""" DISK OK - free space: / 3326 MB (56%); | /=2643MB;5948;5958;0;5968"""
)
OUTPUT_3 = """\
DISK OK - free space: / 3326 MB (56%); | /=2643MB;5948;5958;0;5968
/ 15272 MB (77%);
/boot 68 MB (69%);
/home 69357 MB (27%);
/var/log 819 MB (84%); | /boot=68MB;88;93;0;98
/home=69357MB;253404;253409;0;253414
/var/log=818MB;970;975;0;980
"""
@pytest.mark.parametrize( @pytest.mark.parametrize(
"command,args,exit_code", "command,args,exit_code",
...@@ -30,4 +47,48 @@ def test_check_timeout(): ...@@ -30,4 +47,48 @@ def test_check_timeout():
"""Test check with a timed out command.""" """Test check with a timed out command."""
check = nagios.Check("sleep", ["20"]) check = nagios.Check("sleep", ["20"])
with pytest.raises(RuntimeError): with pytest.raises(RuntimeError):
check.run(3) check.run(timeout=3)
def _mock_run(*args, **kwargs):
"""Mock subprocess' run."""
return unittest.mock.patch.object(
nagios.subprocess,
"run",
*args,
**kwargs,
)
@pytest.mark.parametrize(
"exception",
[
FileNotFoundError,
subprocess.TimeoutExpired(cmd="true", timeout=nagios.DEFAULT_TIMEOUT),
],
)
def test_check_exceptions(exception):
"""Test handling of subprocess' run exceptions.
See
https://docs.python.org/3/library/subprocess.html?highlight=popen#subprocess.SubprocessError
"""
check = nagios.Check("true")
with _mock_run(side_effect=exception), pytest.raises(RuntimeError):
check.run()
@pytest.mark.parametrize("output", [OUTPUT_1, OUTPUT_2, OUTPUT_3])
def test_output_parsing(output):
"""Test parsing of plugin output."""
check = nagios.Check("true")
proc = subprocess.CompletedProcess(
[
"true",
],
0,
output,
"",
)
with _mock_run(return_value=proc):
check.run()
# TODO: validate something.
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment