gitorious/nagios-http-json
1
0
Fork 0
mirror of https://github.com/drewkerrigan/nagios-http-json.git synced 2024-11-22 02:13:47 +01:00
Code Issues Releases Wiki Activity
fix/improve-error-handling
nagios-http-json/test/test_check_http_json.py
aro-lew b920a65afd
Feature: Add Timestamp checks (#87)
2024-06-28 14:39:57 +02:00

451 lines
23 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
import json
import unittest
from unittest.mock import patch
import sys
sys.path.append('..')
from check_http_json import *
OK_CODE = 0
WARNING_CODE = 1
CRITICAL_CODE = 2
UNKNOWN_CODE = 3
class RulesHelper:
separator = '.'
value_separator = ':'
debug = False
key_threshold_warning = None
key_value_list = None
key_value_list_not = None
key_list = None
key_threshold_critical = None
key_value_list_critical = None
key_value_list_not_critical = None
key_time_list = None
key_time_list_critical = None
key_value_list_unknown = None
key_list_critical = None
metric_list = None
def dash_m(self, data):
self.metric_list = data
return self
def dash_e(self, data):
self.key_list = data
return self
def dash_E(self, data):
self.key_list_critical = data
return self
def dash_q(self, data):
self.key_value_list = data
return self
def dash_Q(self, data):
self.key_value_list_critical = data
return self
def dash_y(self, data):
self.key_value_list_not = data
return self
def dash_Y(self, data):
self.key_value_list_not_critical = data
return self
def dash_U(self, data):
self.key_value_list_unknown = data
return self
def dash_w(self, data):
self.key_threshold_warning = data
return self
def dash_c(self, data):
self.key_threshold_critical = data
return self
def dash_dash_key_time(self, data):
self.key_time_list = data
return self
def dash_dash_key_time_critical(self, data):
self.key_time_list_critical = data
return self
class UtilTest(unittest.TestCase):
"""
Tests for the util fucntions
"""
rules = RulesHelper()
def check_data(self, args, jsondata, code):
data = json.loads(jsondata)
nagios = NagiosHelper()
processor = JsonRuleProcessor(data, args)
nagios.append_message(WARNING_CODE, processor.checkWarning())
nagios.append_message(CRITICAL_CODE, processor.checkCritical())
nagios.append_metrics(processor.checkMetrics())
nagios.append_message(UNKNOWN_CODE, processor.checkUnknown())
self.assertEqual(code, nagios.getCode())
def test_metrics(self):
self.check_data(RulesHelper().dash_m(['metric,,1:4,1:5']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_m(['metric,,1:5,1:4']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_m(['metric,,1:5,1:5,6,10']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_m(['metric,,1:5,1:5,1,4']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_m(['metric,s,@1:4,@6:10,1,10']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_m(['(*).value,s,1:5,1:5']),
'[{"value": 5},{"value": 100}]', CRITICAL_CODE)
self.check_data(RulesHelper().dash_m(['metric>foobar,,1:4,1:5']),
'{"metric": 5}', WARNING_CODE)
def test_unknown(self):
self.check_data(RulesHelper().dash_U(['metric,0']),
'{"metric": 3}', UNKNOWN_CODE)
def test_array(self):
self.check_data(RulesHelper().dash_q(['foo(0),bar']),
'{"foo": ["bar"]}', OK_CODE)
self.check_data(RulesHelper().dash_q(['foo(0),foo']),
'{"foo": ["bar"]}', WARNING_CODE)
self.check_data(RulesHelper().dash_Q(['foo(1),bar']),
'{"foo": ["bar"]}', CRITICAL_CODE)
def test_exists(self):
self.check_data(RulesHelper().dash_e(['nothere']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_E(['nothere']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_e(['metric']),
'{"metric": 5}', OK_CODE)
def test_equality(self):
self.check_data(RulesHelper().dash_q(['metric,6']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_Q(['metric,6']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_q(['metric,5']),
'{"metric": 5}', OK_CODE)
def test_equality_colon(self):
"""
See https://github.com/drewkerrigan/nagios-http-json/issues/43
"""
rules = RulesHelper()
rules.value_separator = '_'
# This should not fail
self.check_data(rules.dash_q(['metric,foo:bar']),
'{"metric": "foo:bar"}', OK_CODE)
def test_non_equality(self):
self.check_data(RulesHelper().dash_y(['metric,6']),
'{"metric": 6}', WARNING_CODE)
self.check_data(RulesHelper().dash_Y(['metric,6']),
'{"metric": 6}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_y(['metric,5']),
'{"metric": 6}', OK_CODE)
def test_warning_thresholds(self):
self.check_data(RulesHelper().dash_w(['metric,5']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,5:']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,~:5']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,1:5']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,@5']),
'{"metric": 6}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,@5:']),
'{"metric": 4}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,@~:5']),
'{"metric": 6}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,@1:5']),
'{"metric": 6}', OK_CODE)
self.check_data(RulesHelper().dash_w(['metric,5']),
'{"metric": 6}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,5:']),
'{"metric": 4}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,~:5']),
'{"metric": 6}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,1:5']),
'{"metric": 6}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,@5']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,@5:']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,@~:5']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['metric,@1:5']),
'{"metric": 5}', WARNING_CODE)
self.check_data(RulesHelper().dash_w(['(*).value,@1:5']),
'[{"value": 5},{"value": 1000}]', WARNING_CODE)
def test_critical_thresholds(self):
self.check_data(RulesHelper().dash_c(['metric,5']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,5:']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,~:5']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,1:5']),
'{"metric": 5}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,@5']),
'{"metric": 6}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,@5:']),
'{"metric": 4}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,@~:5']),
'{"metric": 6}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,@1:5']),
'{"metric": 6}', OK_CODE)
self.check_data(RulesHelper().dash_c(['metric,5']),
'{"metric": 6}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,5:']),
'{"metric": 4}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,~:5']),
'{"metric": 6}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,1:5']),
'{"metric": 6}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,@5']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,@5:']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,@~:5']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['metric,@1:5']),
'{"metric": 5}', CRITICAL_CODE)
self.check_data(RulesHelper().dash_c(['(*).value,@1:5']),
'[{"value": 5},{"value": 1000}]', CRITICAL_CODE)
def test_separator(self):
rules = RulesHelper()
rules.separator = '_'
self.check_data(
rules.dash_q(
['(0)_gauges_jvm.buffers.direct.capacity(1)_value,1234']),
'''[{ "gauges": { "jvm.buffers.direct.capacity": [
{"value": 215415},{"value": 1234}]}}]''',
OK_CODE)
self.check_data(
rules.dash_q(
['(*)_gauges_jvm.buffers.direct.capacity(1)_value,1234']),
'''[{ "gauges": { "jvm.buffers.direct.capacity": [
{"value": 215415},{"value": 1234}]}},
{ "gauges": { "jvm.buffers.direct.capacity": [
{"value": 215415},{"value": 1235}]}}]''',
WARNING_CODE)
def test_array_with_missing_element(self):
"""
See https://github.com/drewkerrigan/nagios-http-json/issues/34
"""
rules = RulesHelper()
# This should simply work
data = '[{"Node": "there"}]'
self.check_data(rules.dash_q(['(0).Node,there']), data, OK_CODE)
# This should warn us
data = '[{"Node": "othervalue"}]'
self.check_data(rules.dash_q(['(0).Node,there']), data, WARNING_CODE)
# # This should not throw an IndexError
data = '[{"Node": "foobar"}]'
self.check_data(rules.dash_q(['(0).Node,foobar', '(1).Node,missing']), data, WARNING_CODE)
self.check_data(rules.dash_q(['(0).Node,foobar', '(1).Node,missing', '(2).Node,alsomissing']), data, WARNING_CODE)
# This should not throw a KeyError
data = '{}'
self.check_data(rules.dash_q(['(0).Node,foobar', '(1).Node,missing']), data, CRITICAL_CODE)
def test_subelem(self):
rules = RulesHelper()
data = '{"foo": {"foo": {"foo": "bar"}}}'
self.check_data(rules.dash_E(['foo.foo.foo.foo.foo']), data, CRITICAL_CODE)
def test_subarrayelem_missing_elem(self):
rules = RulesHelper()
data = '[{"capacity": {"value": 1000}},{"capacity": {"value": 2200}}]'
self.check_data(rules.dash_E(['(*).capacity.value']), data, OK_CODE)
self.check_data(rules.dash_E(['(*).capacity.value.too_deep']), data, CRITICAL_CODE)
# Should not throw keyerror
self.check_data(rules.dash_E(['foo']), data, CRITICAL_CODE)
def test_empty_key_value_array(self):
"""
https://github.com/drewkerrigan/nagios-http-json/issues/61
"""
rules = RulesHelper()
# This should simply work
data = '[{"update_status": "finished"},{"update_status": "finished"}]'
self.check_data(rules.dash_q(['(*).update_status,finished']), data, OK_CODE)
# This should warn us
data = '[{"update_status": "finished"},{"update_status": "failure"}]'
self.check_data(rules.dash_q(['(*).update_status,finished']), data, WARNING_CODE)
# This should throw an error
data = '[]'
self.check_data(rules.dash_q(['(*).update_status,warn_me']), data, CRITICAL_CODE)
def test_key_time(self):
if sys.version_info[1] >= 11:
# Test current timestamp.
now = datetime.now(timezone.utc)
data = "{\"timestamp\": \"%s\",\"timestamp2\": \"%s\"}" % (now, now)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30s', 'timestamp2,30s']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@2d']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-2d']), data, CRITICAL_CODE)
# Test 31 minute in the past.
data = "{\"timestamp\": \"%s\"}" % (now - timedelta(minutes=31))
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@2d']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-2d']), data, CRITICAL_CODE)
# Test two hours and one minute in the past.
data = "{\"timestamp\": \"%s\"}" % (now - timedelta(hours=2) - timedelta(minutes=1))
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@2d']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-2d']), data, CRITICAL_CODE)
# Test one day and one minute in the past.
data = "{\"timestamp\": \"%s\"}" % (now - timedelta(days=1) - timedelta(minutes=1))
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@2d']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-2d']), data, CRITICAL_CODE)
# Test two hours and one minute in the future.
data = "{\"timestamp\": \"%s\"}" % (now + timedelta(hours=2) + timedelta(minutes=1))
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@2d']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-2d']), data, CRITICAL_CODE)
else:
data = "{\"timestamp\": \"2020-01-01\"}"
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,2d']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-30m']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-1h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,-3h']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,-2d']), data, OK_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-30m']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-1h']), data, CRITICAL_CODE)
self.check_data(RulesHelper().dash_dash_key_time(['timestamp,@-3h']), data, WARNING_CODE)
self.check_data(RulesHelper().dash_dash_key_time_critical(['timestamp,@-2d']), data, CRITICAL_CODE)
Reference in New Issue View Git Blame Copy Permalink