Nagios check implementation.

import logging

import nagios

import requests

def notify(message):

    """Send a notification."""

    requests.post(

        "https://notify.shore.co.il/send", params={"message": message}

    )

def main():

    if args.verbose:

        logging.basicConfig(level=logging.INFO)

    check = nagios.Check(args.command, args.arguments)

    try:

        check.run(args.timeout, args.dry_run)

    except Exception as ex:

        parser.error(str(ex))

if __name__ == "__main__":

    main()

"""Nagios check implementation."""

import enum

import subprocess

DEFAULT_TIMEOUT = 10  # In seconds.

class NagiosCode(enum.IntEnum):

    """Enum for agreed upon exit codes for a Nagios plugin."""

    OK = 0

    WARNING = 1

    CRITICAL = 2

    UNKNOWN = 3

class Check:

    """Nagios check using a plugin."""

    Command = None

    Arguments = []

    ExitCode = None

    Result = None

    Output = None

    PerfData = None

    AdditionalOutput = None

    StandardError = None

    def __init__(self, command, args):

        if not command:

            raise ValueError("Command is empty.")

        self.Command = command

        if args:

            self.Arguments = args

    def run(self, timeout=DEFAULT_TIMEOUT, dryrun=False):

        """Run the check (if it's hasn't run yet)."""

        if self.Result is not None:

            raise Exception("Check has already run.")

        try:

            proc = subprocess.run(

                [

                    self.Command,

                ]

                + self.Arguments,

                capture_output=True,

                check=False,

                text=True,

                timeout=timeout,

            )

        except FileNotFoundError:

            raise RuntimeError(f"Command {self.Command} not found.")

        except subprocess.TimeoutExpired:

            raise RuntimeError("Timeout exceeded.")