#!/usr/bin/env python from __future__ import unicode_literals import os import sys import time import socket import utils LIVESTATUS_PATH = '/var/run/icinga2/cmd/livestatus' LS_HOST_COLUMNS = [ 'name', 'name', 'display_name', 'display_name', None, 'state', 'state_type', 'current_attempt', 'max_check_attempts', None, 'last_state', None, 'last_state_change', None, 'latency', 'execution_time', 'plugin_output', None, 'last_check', 'address', 'address6' ] LS_SVC_COLUMNS = [ 'description', 'display_name', 'display_name', None, 'state', 'state_type', 'current_attempt', 'max_check_attempts', None, 'last_state', None, 'last_state_change', None, 'latency', 'execution_time', 'plugin_output', 'perf_data', 'last_check', 'host_num_services', 'host_num_services_ok', 'host_num_services_warn', 'host_num_services_unknown', 'host_num_services_crit' ] STATE_MAP = { 'SOFT': 0, 'HARD': 1 } def send_command(command): try: return send_query('COMMAND [{0}] {1}'.format(int(time.time()), command)) except utils.LiveStatusError, error: sys.stderr.write('Failed to execute command: {0}\n\n{1}\n' ''.format(command, error)) def send_query(query): response = LIVESTATUS.query(query + '\nColumnHeaders: on') if response: header, result = response.pop(0), {} return [dict((header[i], v) for i, v in enumerate(r)) for r in response] return [] def get_one(query): return next(iter(send_query(query)), {}) def get_event_output(): try: with open('/tmp/test_event.out') as f: remove = True return f.read().rstrip().split('|') except (IOError, OSError): remove = False finally: if remove: os.system('sudo rm -f /tmp/test_event.out') def convert_output(value): try: return int(value) except ValueError: try: return float(value) except ValueError: return STATE_MAP.get(value, value) def validate_time_format(inputstr, formatstr): try: time.strptime(inputstr, formatstr) except ValueError: return False else: return True def main(): send_command('CHANGE_HOST_EVENT_HANDLER;localhost;test_event') host_info = get_one('GET hosts\nFilter: name = localhost' '\nColumns: event_handler') if host_info.get('event_handler') != 'test_event': utils.Logger.fail('Could not assign eventcommand "test_event"' ' to host "localhost"\n') return 1 utils.Logger.ok('Successfully assigned an eventcommand' ' to host "localhost"\n') send_command('PROCESS_HOST_CHECK_RESULT;localhost;1;A negative result to' ' trigger an eventhandler|some interesting perfdata!') event_output = get_event_output() if not event_output: send_command('CHANGE_HOST_EVENT_HANDLER;localhost;') utils.Logger.fail('Could not trigger the eventcommand\n') return 1 utils.Logger.ok('Successfully triggered the eventcommand\n') failure = False utils.Logger.info('Checking host macros...\n') host_info = get_one('GET hosts\nFilter: name = localhost\nColumns: {0}' ''.format(' '.join(c for c in LS_HOST_COLUMNS if c))) if event_output[0] != host_info['name']*2: failure = True utils.Logger.fail('Escaping environment variables ' 'seems not to properly working\n') utils.Logger.fail(' Expected: {0!r} Got: {1!r}\n' ''.format(host_info['name']*2, event_output[0])) else: utils.Logger.ok('Escaped environment variables' ' are properly processed\n') for i, column in enumerate(LS_HOST_COLUMNS[1:], 1): if column is not None: macro_name, _, macro_value = event_output[i].partition('=') output_value = convert_output(macro_value) if output_value != host_info[column]: failure = True utils.Logger.fail('Macro "{0}" returns an incorrect value. ' 'Expected "{2}" but got "{1}"\n' ''.format(macro_name, output_value, host_info[column])) else: utils.Logger.ok('Macro "{0}" returns the correct value\n' ''.format(macro_name)) utils.Logger.info('Checking service macros...\n') svc_info = get_one('GET services\nFilter: description = ping4\nColumns: {0}' ''.format(' '.join(c for c in LS_SVC_COLUMNS if c))) for i, column in enumerate(LS_SVC_COLUMNS, len(LS_HOST_COLUMNS)): if column is not None: macro_name, _, macro_value = event_output[i].partition('=') output_value = convert_output(macro_value) if output_value != svc_info[column]: failure = True utils.Logger.fail('Macro "{0}" returns an incorrect value. ' 'Expected "{2}" but got "{1}"\n' ''.format(macro_name, output_value, svc_info[column])) else: utils.Logger.ok('Macro "{0}" returns the correct value\n' ''.format(macro_name)) utils.Logger.info('Checking global macros...\n') timet = convert_output(event_output[-6].partition('=')[2]) if not isinstance(timet, int): failure = True utils.Logger.fail('Macro "TIMET" does not return a timestamp. ' 'Expected int but got: {0!r}\n'.format(timet)) else: utils.Logger.ok('Macro "TIMET" returns the correct value\n') longdatetime = event_output[-5].partition('=')[2] longdatetime_format = '%Y-%m-%d %H:%M:%S +0000' if not validate_time_format(longdatetime, longdatetime_format): failure = True utils.Logger.fail('Macro "LONGDATETIME" returns an incorrect value.' ' Expected value of format "{0}" but got "{1}"\n' ''.format(longdatetime_format, longdatetime)) else: utils.Logger.ok('Macro "LONGDATETIME" returns the correct value\n') shortdatetime = event_output[-4].partition('=')[2] shortdatetime_format = '%Y-%m-%d %H:%M:%S' if not validate_time_format(shortdatetime, shortdatetime_format): failure = True utils.Logger.fail('Macro "SHORTDATETIME" returns an incorrect value.' ' Expected value of format "{0}" but got "{1}"\n' ''.format(shortdatetime_format, shortdatetime)) else: utils.Logger.ok('Macro "SHORTDATETIME" returns the correct value\n') m_date = event_output[-3].partition('=')[2] m_date_format = '%Y-%m-%d' if not validate_time_format(m_date, m_date_format): failure = True utils.Logger.fail('Macro "DATE" returns an incorrect value. ' 'Expected value of format "{0}" but got "{1}"\n' ''.format(m_date_format, m_date)) else: utils.Logger.ok('Macro "DATE" returns the correct value\n') m_time = event_output[-2].partition('=')[2] m_time_format = '%H:%M:%S +0000' if not validate_time_format(m_time, m_time_format): failure = True utils.Logger.fail('Macro "TIME" returns an incorrect value. ' 'Expected value of format "{0}" but got "{1}"\n' ''.format(m_time_format, m_time)) else: utils.Logger.ok('Macro "TIME" returns the correct value\n') utils.Logger.info('Checking command macros...\n') if convert_output(event_output[-1].partition('=')[2]) != 1337: failure = True utils.Logger.fail('The command macro "custom_macro"' ' is not being substituted\n') else: utils.Logger.ok('The command macro "custom_macro"' ' is correctly substituted\n') send_command('DISABLE_HOST_EVENT_HANDLER;localhost') send_command('PROCESS_HOST_CHECK_RESULT;localhost;0;A positive result that' ' should not trigger an eventhandler') if get_event_output(): failure = True utils.Logger.fail('Could not disable the eventcommand\n') else: utils.Logger.ok('Successfully disabled the eventcommand\n') send_command('ENABLE_HOST_EVENT_HANDLER;localhost') host_info = get_one('GET hosts\nFilter: name = localhost' '\nColumns: event_handler_enabled') if host_info['event_handler_enabled'] != 1: failure = True utils.Logger.fail('Could not re-enable the eventcommand\n') else: utils.Logger.ok('Successfully re-enabled the eventcommand\n') send_command('CHANGE_HOST_EVENT_HANDLER;localhost;') host_info = get_one('GET hosts\nFilter: name = localhost' '\nColumns: event_handler') if host_info['event_handler']: failure = True utils.Logger.fail('Could not remove eventcommand "test_event"' ' assigned to host "localhost"\n') else: utils.Logger.ok('Successfully removed the eventcommand' ' assigned to host "localhost"\n') return 1 if failure else 0 if __name__ == '__main__': try: with utils.LiveStatusSocket(LIVESTATUS_PATH) as LIVESTATUS: sys.exit(main()) except (OSError, IOError, socket.error), e: utils.Logger.error('Could not connect to Livestatus socket: {0} ({1})' '\n'.format(LIVESTATUS_PATH, unicode(e)))