[wip] Add pylint and fix pylint issues

This commit is contained in:
Markus Opolka 2020-03-18 08:09:35 +01:00
parent c90b0323f5
commit 83ee5062f5
3 changed files with 43 additions and 33 deletions

View File

@ -12,6 +12,11 @@ jobs:
steps: steps:
- name: Checkout - name: Checkout
uses: actions/checkout@v2 uses: actions/checkout@v2
- name: Lint
run: |
pip3 install --upgrade pip wheel setuptools
pip3 install pylint
python3 -m pylint check_http_json.py
- name: Unit Test - name: Unit Test
run: | run: |
python3 -m unittest discover python3 -m unittest discover

5
.pylintrc Normal file
View File

@ -0,0 +1,5 @@
# pylint config
[MESSAGES CONTROL]
disable=line-too-long, redefined-outer-name, too-many-arguments, too-many-instance-attributes, fixme, invalid-name, superfluous-parens, missing-function-docstring, missing-module-docstring, multiple-imports, no-else-return, too-many-return-statements
[MASTER]
ignore-patterns=^test.*

View File

@ -1,14 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
plugin_description = \
"""
Check HTTP JSON Nagios Plugin
Generic Nagios plugin which checks json values from a given endpoint against
argument specified rules and determines the status and performance data for
that service.
"""
import urllib.request, urllib.error, urllib.parse import urllib.request, urllib.error, urllib.parse
import base64 import base64
import json import json
@ -19,13 +10,22 @@ from pprint import pprint
from urllib.error import HTTPError from urllib.error import HTTPError
from urllib.error import URLError from urllib.error import URLError
plugin_description = \
"""
Check HTTP JSON Nagios Plugin
Generic Nagios plugin which checks json values from a given endpoint against
argument specified rules and determines the status and performance data for
that service.
"""
OK_CODE = 0 OK_CODE = 0
WARNING_CODE = 1 WARNING_CODE = 1
CRITICAL_CODE = 2 CRITICAL_CODE = 2
UNKNOWN_CODE = 3 UNKNOWN_CODE = 3
__version__ = '2.0.0' __version__ = '2.0.0'
__version_date__ = '2022-02-16' __version_date__ = '2020-03-22'
class NagiosHelper: class NagiosHelper:
""" """
@ -104,7 +104,6 @@ class JsonHelper:
remainingKey = key[separatorIndex + 1:] remainingKey = key[separatorIndex + 1:]
if partialKey in data: if partialKey in data:
return self.get(remainingKey, data[partialKey]) return self.get(remainingKey, data[partialKey])
else:
return (None, 'not_found') return (None, 'not_found')
def getSubArrayElement(self, key, data): def getSubArrayElement(self, key, data):
@ -172,7 +171,7 @@ class JsonHelper:
if key.find(self.arrayOpener) != -1: if key.find(self.arrayOpener) != -1:
return self.getSubArrayElement(key, data) return self.getSubArrayElement(key, data)
else: else:
if type(data) == dict and key in data: if isinstance(data, dict) and key in data:
return data[key] return data[key]
else: else:
return (None, 'not_found') return (None, 'not_found')
@ -248,7 +247,7 @@ class JsonRuleProcessor:
def expandKeys(self, src): def expandKeys(self, src):
if src is None: if src is None:
return return []
dest = [] dest = []
for key in src: for key in src:
newKeys = self.helper.expandKey(key, []) newKeys = self.helper.expandKey(key, [])
@ -269,7 +268,7 @@ class JsonRuleProcessor:
for kv in equality_list: for kv in equality_list:
k, v = kv.split(',') k, v = kv.split(',')
key, alias = _getKeyAlias(k) key, alias = _getKeyAlias(k)
if (self.helper.equals(key, v) == False): if not self.helper.equals(key, v):
failure += " Key %s mismatch. %s != %s" % (alias, v, failure += " Key %s mismatch. %s != %s" % (alias, v,
self.helper.get(key)) self.helper.get(key))
return failure return failure
@ -279,7 +278,7 @@ class JsonRuleProcessor:
for kv in equality_list: for kv in equality_list:
k, v = kv.split(',') k, v = kv.split(',')
key, alias = _getKeyAlias(k) key, alias = _getKeyAlias(k)
if (self.helper.equals(key, v) == True): if self.helper.equals(key, v):
failure += " Key %s match found. %s == %s" % (alias, v, failure += " Key %s match found. %s == %s" % (alias, v,
self.helper.get(key)) self.helper.get(key))
return failure return failure
@ -415,7 +414,7 @@ def parseArgs(args):
""" """
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description = plugin_description + '\n\nVersion: %s (%s)' description=plugin_description + '\n\nVersion: %s (%s)'
%(__version__, __version_date__), %(__version__, __version_date__),
formatter_class=argparse.RawDescriptionHelpFormatter formatter_class=argparse.RawDescriptionHelpFormatter
) )
@ -525,7 +524,7 @@ def debugPrint(debug_flag, message, pretty_flag=False):
print(message) print(message)
"""Program entry point""" # Program entry point
if __name__ == "__main__": if __name__ == "__main__":
args = parseArgs(sys.argv[1:]) args = parseArgs(sys.argv[1:])
@ -534,7 +533,7 @@ if __name__ == "__main__":
if args.version: if args.version:
print('Version: %s - Date: %s' % (__version__, __version_date__)) print('Version: %s - Date: %s' % (__version__, __version_date__))
exit(0) sys.exit(0)
if args.ssl: if args.ssl:
url = "https://%s" % args.host url = "https://%s" % args.host
@ -552,25 +551,25 @@ if __name__ == "__main__":
context.load_verify_locations(args.cacert) context.load_verify_locations(args.cacert)
except ssl.SSLError: except ssl.SSLError:
nagios.append_unknown( nagios.append_unknown(
''' Error loading SSL CA cert "%s"!''' 'Error loading SSL CA cert "%s"!'
% args.cacert) % args.cacert)
if args.cert: if args.cert:
try: try:
context.load_cert_chain(args.cert,keyfile=args.key) context.load_cert_chain(args.cert, keyfile=args.key)
except ssl.SSLError: except ssl.SSLError:
if args.key: if args.key:
nagios.append_unknown( nagios.append_unknown(
''' Error loading SSL cert. Make sure key "%s" belongs to cert "%s"!''' 'Error loading SSL cert. Make sure key "%s" belongs to cert "%s"!'
% (args.key, args.cert)) % (args.key, args.cert))
else: else:
nagios.append_unknown( nagios.append_unknown(
''' Error loading SSL cert. Make sure "%s" contains the key as well!''' 'Error loading SSL cert. Make sure "%s" contains the key as well!'
% (args.cert)) % (args.cert))
if nagios.getCode() != OK_CODE: if nagios.getCode() != OK_CODE:
print(nagios.getMessage()) print(nagios.getMessage())
exit(nagios.getCode()) sys.exit(nagios.getCode())
else: else:
url = "http://%s" % args.host url = "http://%s" % args.host
@ -586,6 +585,7 @@ if __name__ == "__main__":
req = urllib.request.Request(url) req = urllib.request.Request(url)
req.add_header("User-Agent", "check_http_json") req.add_header("User-Agent", "check_http_json")
if args.auth: if args.auth:
# TODO: replace deprecated encodestring
base64str = base64.encodestring(args.auth).replace('\n', '') base64str = base64.encodestring(args.auth).replace('\n', '')
req.add_header('Authorization', 'Basic %s' % base64str) req.add_header('Authorization', 'Basic %s' % base64str)
if args.headers: if args.headers:
@ -628,6 +628,6 @@ if __name__ == "__main__":
# Print Nagios specific string and exit appropriately # Print Nagios specific string and exit appropriately
print(nagios.getMessage()) print(nagios.getMessage())
exit(nagios.getCode()) sys.exit(nagios.getCode())
#EOF #EOF