diff --git a/check_hwgroup.py b/check_hwgroup.py index 1c669a8..6b11136 100755 --- a/check_hwgroup.py +++ b/check_hwgroup.py @@ -1,11 +1,8 @@ -#!/usr/bin/env python +#!/usr/bin/python3 # check_hwgroup.py - checks the hwgroup environmental devices # Copyright (C) 2014 NETWAYS GmbH # -# Version: 1.0 -# Author: Alexander Klimov -# # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 2 of the License, or @@ -20,55 +17,66 @@ # along with this program. If not, see . +import asyncio import sys from argparse import ArgumentParser -from pysnmp.entity.rfc3413.oneliner import cmdgen + +# pylint import-untyped import nagiosplugin +# pylint: disable=import-error,no-name-in-module,import-untyped +import pysnmp +# pylint: disable=import-error,no-name-in-module +from pysnmp.hlapi.v3arch.asyncio import SnmpEngine as pySnmp_engine +from pysnmp.hlapi.v3arch.asyncio import next_cmd as pySnmp_next_cmd -__version__ = '1.2.0' +__version__ = "1.2.0" -enterprise = '.1.3.6.1.4.1.21796' +enterprise = ".1.3.6.1.4.1.21796" sensor_ids = { - 'Poseidon': '3.3.3.1.8', - 'Damocles': '3.4.3.1.8', - 'STE2': '4.9.3.1.8', - 'STE': '4.1.3.1.8', - 'WLD': '4.5.4.1.5' + "Poseidon": "3.3.3.1.8", + "Damocles": "3.4.3.1.8", + "STE2": "4.9.3.1.8", + "STE": "4.1.3.1.8", + "WLD": "4.5.4.1.5", } sensor_paths = { - 'Poseidon': ( - '3.3.3.1.2', - '3.3.3.1.4', - '3.3.3.1.6' - ), - 'Damocles': ( - '3.4.3.1.2', - '3.4.3.1.4', - '3.4.3.1.6' - ), - 'STE2': ( - '4.9.3.1.2', - '4.9.3.1.3', - '4.9.3.1.5' - ), - 'STE': ( - '4.1.3.1.2', - '4.1.3.1.3', - '4.1.3.1.5' - ), - 'WLD': ( - '4.5.4.1.2', - '4.5.4.1.3', - '4.5.4.1.6' - ) + "Poseidon": ("3.3.3.1.2", "3.3.3.1.4", "3.3.3.1.6"), + "Damocles": ("3.4.3.1.2", "3.4.3.1.4", "3.4.3.1.6"), + "STE2": ("4.9.3.1.2", "4.9.3.1.3", "4.9.3.1.5"), + "STE": ("4.1.3.1.2", "4.1.3.1.3", "4.1.3.1.5"), + "WLD": ("4.5.4.1.2", "4.5.4.1.3", "4.5.4.1.6"), } -device_types = { - 'Damocles': 4, - 'Poseidon': 3 -} +device_types = {"Damocles": 4, "Poseidon": 3} + + +def SNMPRequest_sync(host: str, port: int, community: str, oid: str): + """ + snmp_query executes the actual query + """ + + snmp_engine = pySnmp_engine() + + snmp_object = pysnmp.smi.rfc1902.ObjectType( + pysnmp.smi.rfc1902.ObjectIdentity(oid) + ) + # pylint: disable=c-extension-no-member + error_indication, error_status, _, result = asyncio.run( + pySnmp_next_cmd( + snmp_engine, + pysnmp.hlapi.v3arch.asyncio.auth.CommunityData(community), + asyncio.run( + pysnmp.hlapi.v3arch.asyncio.UdpTransportTarget.create((host, port)) + ), + pysnmp.hlapi.v3arch.asyncio.ContextData(), + snmp_object, + ) + ) + + return error_indication, error_status, result + class CheckHWGroupError(Exception): """ @@ -82,6 +90,7 @@ class CheckHWGroupResource(nagiosplugin.Resource): """ The resource for fetching sensor/contact/output values """ + def __init__(self, host, community, port, sensor, contact, output): """ Prepare the resource @@ -103,10 +112,12 @@ def __init__(self, host, community, port, sensor, contact, output): self.contact = contact self.output = output - self.deviceName = str(self.SNMPReq('.1.3.6.1.2.1.1.1.0')) - self.notSupported = CheckHWGroupError("Device '{}' not supported".format(self.deviceName)) + self.deviceName = str(self.SNMPReq(".1.3.6.1.2.1.1.1.0")) + self.notSupported = CheckHWGroupError( + "Device '{}' not supported".format(self.deviceName) + ) - supportedDevices = ('Poseidon', 'Damocles', 'STE2', 'STE', 'WLD') + supportedDevices = ("Poseidon", "Damocles", "STE2", "STE", "WLD") # Check if supported Device string is in deviceName, if empty list throw error for device in supportedDevices: if device in self.deviceName: @@ -115,26 +126,24 @@ def __init__(self, host, community, port, sensor, contact, output): else: raise self.notSupported - def SNMPReq(self, MIBInit): + def SNMPReq(self, oid): """Send a SNMP request and return the response - :param MIBInit: MIB (initializer) - :type MIBInit: str + :param oid: MIB (initializer) + :type oid: str :rtype: str :raise CheckHWGroupError: if a SNMP error occurs """ - errorIndication, errorStatus, _, varBinds = cmdgen.CommandGenerator().getCmd( - cmdgen.CommunityData(self.community, mpModel=0), - cmdgen.UdpTransportTarget((self.host, self.port)), - cmdgen.MibVariable(MIBInit), - lookupNames=True, - lookupValues=True) + + errorIndication, errorStatus, varBinds = SNMPRequest_sync( + self.host, self.port, self.community, oid + ) if errorIndication: - raise CheckHWGroupError('SNMP error: {}'.format(errorIndication)) + raise CheckHWGroupError("SNMP error: {}".format(errorIndication)) if errorStatus: - raise CheckHWGroupError('SNMP error: {}'.format(errorStatus)) + raise CheckHWGroupError("SNMP error: {}".format(errorStatus)) return str(varBinds[0][1]) @@ -144,47 +153,71 @@ def probe(self): :rtype: nagiosplugin.Metric :raise CheckHWGroupError: various reasons, see error messages below """ - yield nagiosplugin.Metric( - *self._probe(), - context='check_hwgroup' - ) + yield nagiosplugin.Metric(*self._probe(), context="check_hwgroup") def _probe_sensor(self): for ident in (1, 2): try: - sensorID = int(self.SNMPReq('{}.{}.{}'.format(enterprise, sensor_ids[self.deviceType], ident))) + sensorID = int( + self.SNMPReq( + "{}.{}.{}".format( + enterprise, sensor_ids[self.deviceType], ident + ) + ) + ) except (ValueError, CheckHWGroupError): continue if sensorID == self.sensor: sensorID = ident break else: - raise CheckHWGroupError('Sensor ID ({}) not found'.format(self.sensor)) + raise CheckHWGroupError("Sensor ID ({}) not found".format(self.sensor)) - (sensName, sensState, sensValue) = [self.SNMPReq('{}.{}.{}'.format(enterprise, OID, sensorID)) for OID in sensor_paths[self.deviceType]] + sensName, sensState, sensValue = [ + self.SNMPReq("{}.{}.{}".format(enterprise, OID, sensorID)) + for OID in sensor_paths[self.deviceType] + ] if not int(sensState): - raise CheckHWGroupError('Error. No sensor attached') + raise CheckHWGroupError("Error. No sensor attached") return (sensName, float(sensValue) / 10) def _probe_contact(self): - (inpValue, inpName, inpAlarmSetup, inpAlarmState) = [self.SNMPReq('{}.3.{}.1.1.{}.{}'.format(enterprise, device_types[self.deviceType], OID, self.contact)) for OID in range(2, 6)] - - contactName = '{} [AlarmState: {}, AlarmSetup: {}]'.format( + inpValue, inpName, inpAlarmSetup, inpAlarmState = [ + self.SNMPReq( + "{}.3.{}.1.1.{}.{}".format( + enterprise, device_types[self.deviceType], OID, self.contact + ) + ) + for OID in range(2, 6) + ] + + contactName = "{} [AlarmState: {}, AlarmSetup: {}]".format( inpName, - ('normal', 'alarm')[int(inpAlarmState)], - ('inactive', 'activeOff', 'activeOn')[int(inpAlarmSetup)]) + ("normal", "alarm")[int(inpAlarmState)], + ("inactive", "activeOff", "activeOn")[int(inpAlarmSetup)], + ) return (contactName, float(inpValue)) def _probe_output(self): - (outValue, outName, outType, outMode) = [self.SNMPReq('{}.3.{}.2.1.{}.{}'.format(enterprise, device_types[self.deviceType], OID, self.output)) for OID in range(2, 6)] - - outputName = '{} [Type: {}, Mode: {}]'.format( + outValue, outName, outType, outMode = [ + self.SNMPReq( + "{}.3.{}.2.1.{}.{}".format( + enterprise, device_types[self.deviceType], OID, self.output + ) + ) + for OID in range(2, 6) + ] + + outputName = "{} [Type: {}, Mode: {}]".format( outName, - ('relay (off, on)', 'rts (-10V,+10V)', 'dtr (0V,+10V)')[int(outType)], - ('manual', 'autoAlarm','autoTriggerEq', 'autoTriggerHi', 'autoTriggerLo')[int(outMode)]) + ("relay (off, on)", "rts (-10V,+10V)", "dtr (0V,+10V)")[int(outType)], + ("manual", "autoAlarm", "autoTriggerEq", "autoTriggerHi", "autoTriggerLo")[ + int(outMode) + ], + ) return (outputName, float(outValue)) @@ -194,48 +227,80 @@ def _probe(self): if not self.sensor and self.contact: - if self.deviceType not in ['Damocles', 'Poseidon']: - raise CheckHWGroupError("Checking contact is not supported with this device") + if self.deviceType not in ["Damocles", "Poseidon"]: + raise CheckHWGroupError( + "Checking contact is not supported with this device" + ) return self._probe_contact() if not self.sensor and not self.contact: - if self.deviceType not in ['Damocles', 'Poseidon']: - raise CheckHWGroupError("Checking output is not supported with this device") + if self.deviceType not in ["Damocles", "Poseidon"]: + raise CheckHWGroupError( + "Checking output is not supported with this device" + ) return self._probe_output() + def commandline(args): - argp = ArgumentParser(description='Check Plugin for the hwgroup environmental devices.') - - argp.add_argument('-V', '--version', action='version', version=__version__) - argp.add_argument('-v', '--verbose', action='count', default=0) - argp.add_argument('-H', '--host', type=str, required=True, - help='The hostname or ipaddress of the hwgroup device') - argp.add_argument('-C', '--community', type=str, default='public', - help='The SNMP community of the hwgroup device') - argp.add_argument('-P', '--port', default=161, - help='The port of the hwgroup device') - argp.add_argument('-w', '--warning', type=float, required=True, - help='Warning threshold') - argp.add_argument('-c', '--critical', type=float, required=True, - help='Critical threshold') + argp = ArgumentParser( + description="Check Plugin for the hwgroup environmental devices." + ) + + argp.add_argument("-V", "--version", action="version", version=__version__) + argp.add_argument("-v", "--verbose", action="count", default=0) + argp.add_argument( + "-H", + "--host", + type=str, + required=True, + help="The hostname or ipaddress of the hwgroup device", + ) + argp.add_argument( + "-C", + "--community", + type=str, + default="public", + help="The SNMP community of the hwgroup device", + ) + argp.add_argument( + "-P", "--port", default=161, help="The port of the hwgroup device" + ) + argp.add_argument( + "-w", "--warning", type=float, required=True, help="Warning threshold" + ) + argp.add_argument( + "-c", "--critical", type=float, required=True, help="Critical threshold" + ) SIO = argp.add_mutually_exclusive_group(required=True) - SIO.add_argument('-S', '--sensor', type=int, help='The sensor to check') - SIO.add_argument('-I', '--contact', type=int, help='The dry contact to check the sensor to. Only for Damocles, Poseidon') - SIO.add_argument('-O', '--output', type=int, help='The relay output to check. Only for Damocles, Poseidon') + SIO.add_argument("-S", "--sensor", type=int, help="The sensor to check") + SIO.add_argument( + "-I", + "--contact", + type=int, + help="The dry contact to check the sensor to. Only for Damocles, Poseidon", + ) + SIO.add_argument( + "-O", + "--output", + type=int, + help="The relay output to check. Only for Damocles, Poseidon", + ) return argp.parse_args(args) def main(args): try: - resource = CheckHWGroupResource(args.host, args.community, args.port, args.sensor, args.contact, args.output) + resource = CheckHWGroupResource( + args.host, args.community, args.port, args.sensor, args.contact, args.output + ) check = nagiosplugin.Check( resource, - nagiosplugin.ScalarContext('check_hwgroup', args.warning, args.critical), + nagiosplugin.ScalarContext("check_hwgroup", args.warning, args.critical), ) check.name = resource.deviceName check.main(min(args.verbose, 3)) @@ -244,13 +309,15 @@ def main(args): return 3 -if __name__ == '__main__': # pragma: no cover +if __name__ == "__main__": # pragma: no cover try: ARGS = commandline(sys.argv[1:]) sys.exit(main(ARGS)) except SystemExit: # Re-throw the exception - raise sys.exc_info()[1].with_traceback(sys.exc_info()[2]) # pylint: disable=raise-missing-from - except: # pylint: disable=bare-except + raise sys.exc_info()[1].with_traceback( + sys.exc_info()[2] + ) # pylint: disable=raise-missing-from + except: # pylint: disable=bare-except # noqa: E722 print("[UNKNOWN] - Error: %s" % (str(sys.exc_info()[1]))) sys.exit(3) diff --git a/requirements.txt b/requirements.txt index 3a21057..3751026 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -pysnmplib==5.0.24 +pysnmplib==7.1.22 nagiosplugin==1.3.3 pyasyncore==1.0.4 diff --git a/test_check_hwgroup.py b/test_check_hwgroup.py index c1fbec7..189a36f 100644 --- a/test_check_hwgroup.py +++ b/test_check_hwgroup.py @@ -1,62 +1,82 @@ #!/usr/bin/env python3 +import sys import unittest import unittest.mock as mock -import sys -sys.path.append('..') +sys.path.append("..") -from check_hwgroup import commandline -from check_hwgroup import CheckHWGroupResource -from check_hwgroup import CheckHWGroupError -from check_hwgroup import main +from check_hwgroup import (CheckHWGroupError, CheckHWGroupResource, + commandline, main) class CLITesting(unittest.TestCase): def test_commandline(self): - actual = commandline(['-H', 'localhost', '-C', 'foobar', '-w', '5', '-c', '10', '-S', '216']) - self.assertEqual(actual.host, 'localhost') - self.assertEqual(actual.community, 'foobar') + actual = commandline( + ["-H", "localhost", "-C", "foobar", "-w", "5", "-c", "10", "-S", "216"] + ) + self.assertEqual(actual.host, "localhost") + self.assertEqual(actual.community, "foobar") self.assertEqual(actual.critical, 10) def test_commandline_error(self): with self.assertRaises(SystemExit): # Mutual Exclusive - commandline(['-H', 'localhost', '-C', 'foobar', '-w', '5', '-c', '10', '-S', '216', '-I', 0]) + commandline( + [ + "-H", + "localhost", + "-C", + "foobar", + "-w", + "5", + "-c", + "10", + "-S", + "216", + "-I", + 0, + ] + ) with self.assertRaises(SystemExit): # Not an int - commandline(['-H', 'localhost', '-C', 'foobar', '-w', '5', '-c', '10', '-S', 'foo']) + commandline( + ["-H", "localhost", "-C", "foobar", "-w", "5", "-c", "10", "-S", "foo"] + ) + class MainTesting(unittest.TestCase): - @mock.patch('builtins.print') + @mock.patch("builtins.print") def test_main_error(self, mock_print): - def SNMPReq(x,y): - return 'Poseidon 1337' + def SNMPReq(x, y): + return "Poseidon 1337" CheckHWGroupResource.SNMPReq = SNMPReq - args = commandline(['-H', 'localhost', '-C', 'foobar', '-w', '5', '-c', '10', '-S', '216']) + args = commandline( + ["-H", "localhost", "-C", "foobar", "-w", "5", "-c", "10", "-S", "216"] + ) actual = main(args) self.assertEqual(actual, 3) - mock_print.assert_called_with('[UNKNOWN] - Error: Sensor ID (216) not found') + mock_print.assert_called_with("[UNKNOWN] - Error: Sensor ID (216) not found") - @mock.patch('builtins.print') + @mock.patch("builtins.print") def test_main(self, mock_print): m = mock.MagicMock() def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'Poseidon 1337', - '.1.3.6.1.4.1.21796.3.3.3.1.8.1': '666', # SensorID - '.1.3.6.1.4.1.21796.3.3.3.1.8.2': '111', # SensState Poseidon - '.1.3.6.1.4.1.21796.3.3.3.1.2.1': 'poseidon', # SensName - '.1.3.6.1.4.1.21796.3.3.3.1.4.1': '222', # SensState - '.1.3.6.1.4.1.21796.3.3.3.1.6.1': '333' # SensValue + ".1.3.6.1.2.1.1.1.0": "Poseidon 1337", + ".1.3.6.1.4.1.21796.3.3.3.1.8.1": "666", # SensorID + ".1.3.6.1.4.1.21796.3.3.3.1.8.2": "111", # SensState Poseidon + ".1.3.6.1.4.1.21796.3.3.3.1.2.1": "poseidon", # SensName + ".1.3.6.1.4.1.21796.3.3.3.1.4.1": "222", # SensState + ".1.3.6.1.4.1.21796.3.3.3.1.6.1": "333", # SensValue } return values[arg] @@ -65,57 +85,71 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - args = commandline(['-H', 'localhost', '-C', 'foobar', '-w', '5', '-c', '10', '-S', '666']) + args = commandline( + ["-H", "localhost", "-C", "foobar", "-w", "5", "-c", "10", "-S", "666"] + ) with self.assertRaises(SystemExit): main(args) - mock_print.assert_called_with('POSEIDON 1337 CRITICAL - poseidon is 33.3 (outside range 0:10.0) | poseidon=33.3;5.0;10.0\n', end='', file=None) + mock_print.assert_called_with( + "POSEIDON 1337 CRITICAL - poseidon is 33.3 (outside range 0:10.0) | poseidon=33.3;5.0;10.0\n", + end="", + file=None, + ) + class CheckTesting(unittest.TestCase): def test_checkhw(self): - def SNMPReq(x,y): - return 'Poseidon 1337' + def SNMPReq(x, y): + return "Poseidon 1337" CheckHWGroupResource.SNMPReq = SNMPReq - check = CheckHWGroupResource('host', 'community', 1234, 'sensor', 'contact', 'output') + check = CheckHWGroupResource( + "host", "community", 1234, "sensor", "contact", "output" + ) - self.assertEqual(check.deviceName, 'Poseidon 1337') - self.assertEqual(check.deviceType, 'Poseidon') + self.assertEqual(check.deviceName, "Poseidon 1337") + self.assertEqual(check.deviceType, "Poseidon") def test_checkhw_ste2(self): - def SNMPReq(x,y): - return 'STE2 r2, fw:1.5.4_2373' + def SNMPReq(x, y): + return "STE2 r2, fw:1.5.4_2373" CheckHWGroupResource.SNMPReq = SNMPReq - check = CheckHWGroupResource('host', 'community', 1234, 'sensor', 'contact', 'output') + check = CheckHWGroupResource( + "host", "community", 1234, "sensor", "contact", "output" + ) - self.assertEqual(check.deviceName, 'STE2 r2, fw:1.5.4_2373') - self.assertEqual(check.deviceType, 'STE2') + self.assertEqual(check.deviceName, "STE2 r2, fw:1.5.4_2373") + self.assertEqual(check.deviceType, "STE2") def test_checkhw_unsupported(self): - def SNMPReq(x,y): - return 'Foobar Bar 1337' + def SNMPReq(x, y): + return "Foobar Bar 1337" + CheckHWGroupResource.SNMPReq = SNMPReq with self.assertRaises(CheckHWGroupError) as hwerror: - CheckHWGroupResource('host', 'community', 1234, 'sensor', 'contact', 'output') + CheckHWGroupResource( + "host", "community", 1234, "sensor", "contact", "output" + ) def test_probe_poseidon_sensor(self): m = mock.MagicMock() def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'Poseidon 1337', - '.1.3.6.1.4.1.21796.3.3.3.1.8.1': '666', # SensorID - '.1.3.6.1.4.1.21796.3.3.3.1.8.2': '111', # SensState Poseidon - '.1.3.6.1.4.1.21796.3.3.3.1.2.1': 'poseidon', # SensName - '.1.3.6.1.4.1.21796.3.3.3.1.4.1': '222', # SensState - '.1.3.6.1.4.1.21796.3.3.3.1.6.1': '333' # SensValue + ".1.3.6.1.2.1.1.1.0": "Poseidon 1337", + ".1.3.6.1.4.1.21796.3.3.3.1.8.1": "666", # SensorID + ".1.3.6.1.4.1.21796.3.3.3.1.8.2": "111", # SensState Poseidon + ".1.3.6.1.4.1.21796.3.3.3.1.2.1": "poseidon", # SensName + ".1.3.6.1.4.1.21796.3.3.3.1.4.1": "222", # SensState + ".1.3.6.1.4.1.21796.3.3.3.1.6.1": "333", # SensValue } return values[arg] @@ -124,10 +158,12 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - check = CheckHWGroupResource('host', 'community', 1234, sensor=666, contact=None, output=None) + check = CheckHWGroupResource( + "host", "community", 1234, sensor=666, contact=None, output=None + ) actual = check._probe() - expected = ('poseidon', 33.3) + expected = ("poseidon", 33.3) self.assertEqual(actual, expected) def test_probe_damocles_error(self): @@ -135,9 +171,9 @@ def test_probe_damocles_error(self): def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'Damocles', - '.1.3.6.1.4.1.21796.3.4.3.1.8.1': 'nosuchsensor', # SensorID - '.1.3.6.1.4.1.21796.3.4.3.1.8.2': '' # SensState + ".1.3.6.1.2.1.1.1.0": "Damocles", + ".1.3.6.1.4.1.21796.3.4.3.1.8.1": "nosuchsensor", # SensorID + ".1.3.6.1.4.1.21796.3.4.3.1.8.2": "", # SensState } return values[arg] @@ -146,7 +182,9 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - check = CheckHWGroupResource('host', 'community', 1234, sensor=666, contact=None, output=None) + check = CheckHWGroupResource( + "host", "community", 1234, sensor=666, contact=None, output=None + ) with self.assertRaises(CheckHWGroupError): check._probe() @@ -156,11 +194,11 @@ def test_probe_poseidon_contact(self): def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'Poseidon 3', - '.1.3.6.1.4.1.21796.3.3.1.1.2.666': '333', # inpValue - '.1.3.6.1.4.1.21796.3.3.1.1.3.666': 'barfoo', # inpName - '.1.3.6.1.4.1.21796.3.3.1.1.4.666': '0', # AlarmSetup - '.1.3.6.1.4.1.21796.3.3.1.1.5.666': '0' # AlarmState + ".1.3.6.1.2.1.1.1.0": "Poseidon 3", + ".1.3.6.1.4.1.21796.3.3.1.1.2.666": "333", # inpValue + ".1.3.6.1.4.1.21796.3.3.1.1.3.666": "barfoo", # inpName + ".1.3.6.1.4.1.21796.3.3.1.1.4.666": "0", # AlarmSetup + ".1.3.6.1.4.1.21796.3.3.1.1.5.666": "0", # AlarmState } return values[arg] @@ -169,10 +207,12 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - check = CheckHWGroupResource('host', 'community', 1234, sensor=None, contact=666, output=None) + check = CheckHWGroupResource( + "host", "community", 1234, sensor=None, contact=666, output=None + ) actual = check._probe() - expected = ('barfoo [AlarmState: normal, AlarmSetup: inactive]', 333.0) + expected = ("barfoo [AlarmState: normal, AlarmSetup: inactive]", 333.0) self.assertEqual(actual, expected) def test_probe_poseidon_output(self): @@ -180,11 +220,11 @@ def test_probe_poseidon_output(self): def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'Poseidon 3', - '.1.3.6.1.4.1.21796.3.3.2.1.2.666': '333', # inpValue - '.1.3.6.1.4.1.21796.3.3.2.1.3.666': 'barfoo', # inpName - '.1.3.6.1.4.1.21796.3.3.2.1.4.666': '1', # AlarmSetup - '.1.3.6.1.4.1.21796.3.3.2.1.5.666': '2' # AlarmState + ".1.3.6.1.2.1.1.1.0": "Poseidon 3", + ".1.3.6.1.4.1.21796.3.3.2.1.2.666": "333", # inpValue + ".1.3.6.1.4.1.21796.3.3.2.1.3.666": "barfoo", # inpName + ".1.3.6.1.4.1.21796.3.3.2.1.4.666": "1", # AlarmSetup + ".1.3.6.1.4.1.21796.3.3.2.1.5.666": "2", # AlarmState } return values[arg] @@ -193,27 +233,28 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - check = CheckHWGroupResource('host', 'community', 1234, sensor=None, contact=None, output=666) + check = CheckHWGroupResource( + "host", "community", 1234, sensor=None, contact=None, output=666 + ) actual = check._probe() - expected = ('barfoo [Type: rts (-10V,+10V), Mode: autoTriggerEq]', 333.0) + expected = ("barfoo [Type: rts (-10V,+10V), Mode: autoTriggerEq]", 333.0) self.assertEqual(actual, expected) - def test_probe_ste2_sensor(self): m = mock.MagicMock() def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'STE2 r2, fw:1.5.4_2373', - '.1.3.6.1.4.1.21796.4.9.3.1.8.1': '14223', - '.1.3.6.1.4.1.21796.4.9.3.1.8.2': '24167', - '.1.3.6.1.4.1.21796.4.9.3.1.2.1': 'Sensor 14223', - '.1.3.6.1.4.1.21796.4.9.3.1.2.2': 'Sensor 24167', - '.1.3.6.1.4.1.21796.4.9.3.1.3.1': '1', - '.1.3.6.1.4.1.21796.4.9.3.1.3.2': '1', - '.1.3.6.1.4.1.21796.4.9.3.1.5.1': '428', - '.1.3.6.1.4.1.21796.4.9.3.1.5.2': '221' + ".1.3.6.1.2.1.1.1.0": "STE2 r2, fw:1.5.4_2373", + ".1.3.6.1.4.1.21796.4.9.3.1.8.1": "14223", + ".1.3.6.1.4.1.21796.4.9.3.1.8.2": "24167", + ".1.3.6.1.4.1.21796.4.9.3.1.2.1": "Sensor 14223", + ".1.3.6.1.4.1.21796.4.9.3.1.2.2": "Sensor 24167", + ".1.3.6.1.4.1.21796.4.9.3.1.3.1": "1", + ".1.3.6.1.4.1.21796.4.9.3.1.3.2": "1", + ".1.3.6.1.4.1.21796.4.9.3.1.5.1": "428", + ".1.3.6.1.4.1.21796.4.9.3.1.5.2": "221", } return values[arg] @@ -222,27 +263,28 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - check = CheckHWGroupResource('host', 'community', 1234, sensor=14223, contact=None, output=None) + check = CheckHWGroupResource( + "host", "community", 1234, sensor=14223, contact=None, output=None + ) actual = check._probe() - expected = ('Sensor 14223', 42.8) + expected = ("Sensor 14223", 42.8) self.assertEqual(actual, expected) - def test_probe_ste2_contact(self): m = mock.MagicMock() def side_effect(arg): values = { - '.1.3.6.1.2.1.1.1.0': 'STE2 r2, fw:1.5.4_2373', - '.1.3.6.1.4.1.21796.4.9.3.1.8.1': '14223', - '.1.3.6.1.4.1.21796.4.9.3.1.8.2': '24167', - '.1.3.6.1.4.1.21796.4.9.3.1.2.1': 'Sensor 14223', - '.1.3.6.1.4.1.21796.4.9.3.1.2.2': 'Sensor 24167', - '.1.3.6.1.4.1.21796.4.9.3.1.3.1': '1', - '.1.3.6.1.4.1.21796.4.9.3.1.3.2': '1', - '.1.3.6.1.4.1.21796.4.9.3.1.5.1': '428', - '.1.3.6.1.4.1.21796.4.9.3.1.5.2': '221' + ".1.3.6.1.2.1.1.1.0": "STE2 r2, fw:1.5.4_2373", + ".1.3.6.1.4.1.21796.4.9.3.1.8.1": "14223", + ".1.3.6.1.4.1.21796.4.9.3.1.8.2": "24167", + ".1.3.6.1.4.1.21796.4.9.3.1.2.1": "Sensor 14223", + ".1.3.6.1.4.1.21796.4.9.3.1.2.2": "Sensor 24167", + ".1.3.6.1.4.1.21796.4.9.3.1.3.1": "1", + ".1.3.6.1.4.1.21796.4.9.3.1.3.2": "1", + ".1.3.6.1.4.1.21796.4.9.3.1.5.1": "428", + ".1.3.6.1.4.1.21796.4.9.3.1.5.2": "221", } return values[arg] @@ -251,7 +293,9 @@ def side_effect(arg): # Monkey Patch request function with mock CheckHWGroupResource.SNMPReq = m - check = CheckHWGroupResource('host', 'community', 1234, sensor=None, contact=666, output=None) + check = CheckHWGroupResource( + "host", "community", 1234, sensor=None, contact=666, output=None + ) with self.assertRaises(CheckHWGroupError): check._probe()