icinga2/test/jenkins/checkresult.test

96 lines
3.0 KiB
Plaintext
Raw Normal View History

2013-12-09 16:03:31 +01:00
#!/usr/bin/env python
from __future__ import unicode_literals
import os
import sys
import time
2013-12-10 14:55:33 +01:00
import utils
2013-12-09 16:03:31 +01:00
STATE_OK = 0
TYPE_PASSIVE_CHECK = 1
CHECK_INTERVAL = 300 # seconds
CHECKRESULT_READ_INTERVAL = 5 # seconds
CHECKRESULT_LOCATION = '/tmp/icinga2/checkresults'
CHECKRESULT_TEMPLATE = """
host_name=%(hostname)s
service_description=%(servicename)s
check_type=%(check_type)s
check_options=0
scheduled_check=0
reschedule_check=0
latency=0
start_time=%(start_time)s
finish_time=%(finish_time)s
early_timeout=0
exited_ok=%(excited_ok)s
return_code=%(return_code)s
output=%(output)s
"""
def main():
2013-12-10 14:55:33 +01:00
run_query = lambda q: utils.run_mysql_query(q, b'/usr/bin/mysql')
2013-12-09 16:03:31 +01:00
# We need to wait a bit first as Icinga processes a
# checkresult only if its newer than the last check
query = 'select unix_timestamp(s.last_check) as last_check ' \
'from icinga_servicestatus as s ' \
'inner join icinga_services as c ' \
'on s.service_object_id = c.service_object_id ' \
"where c.display_name = 'PassiveService1'"
state_time = float(next(iter(run_query(query)), {}).get('last_check', '0'))
if state_time == 0:
utils.Logger.fail('"PassiveService1" seems not'
' to have been checked yet\n')
2013-12-09 16:03:31 +01:00
return 1
if (state_time + CHECK_INTERVAL) - time.time() < 30:
time.sleep(45)
# Now pass the checkresult in
resultfile_path = os.path.join(CHECKRESULT_LOCATION, 'cfoobar')
with open(resultfile_path, 'w') as f:
2013-12-09 16:03:31 +01:00
f.write(CHECKRESULT_TEMPLATE % {
'hostname': 'nsca-ng',
'servicename': 'PassiveService1',
'check_type': TYPE_PASSIVE_CHECK,
'start_time': time.time(),
'finish_time': time.time(),
'excited_ok': '1',
'return_code': STATE_OK,
'output': 'Passing in CheckResult header files works!'
})
utils.Logger.debug('Written file: {0}\n'.format(resultfile_path))
2013-12-09 16:03:31 +01:00
# And notfiy Icinga that the file has been completely written...
resultfileok_path = os.path.join(CHECKRESULT_LOCATION, 'cfoobar.ok')
with open(resultfileok_path, 'w') as f:
2013-12-09 16:03:31 +01:00
pass
utils.Logger.debug('Written file: {0}\n'.format(resultfileok_path))
2013-12-09 16:03:31 +01:00
# Lastly check whether the service changed its state
time.sleep(CHECKRESULT_READ_INTERVAL * 2)
query = 'select s.output ' \
'from icinga_servicestatus as s ' \
'inner join icinga_services as c ' \
'on s.service_object_id = c.service_object_id ' \
"where c.display_name = 'PassiveService1'"
output = next(iter(run_query(query)), {}).get('output', '')
if output != 'Passing in CheckResult header files works!':
utils.Logger.fail('Checkresult header files seem '
'not to be processed properly\n')
2013-12-09 16:03:31 +01:00
return 1
utils.Logger.ok('Checkresult header files are processed properly\n')
2013-12-09 16:03:31 +01:00
return 0
if __name__ == '__main__':
sys.exit(main())