icinga2/test/jenkins/eventhandler.test

#!/usr/bin/env python
from __future__ import unicode_literals

import os
import sys
import time

import utils


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 success(msg):
    print '[OK] {0}'.format(msg).encode('utf-8')


def fail(msg):
    print '[FAIL] {0}'.format(msg).encode('utf-8')


def info(msg):
    print '[INFO] {0}'.format(msg).encode('utf-8')


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':
        fail('Could not assign eventcommand "test_event" to host "localhost"')
        return 1
    success('Successfully assigned an eventcommand to host "localhost"')

    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;')
        fail('Could not trigger the eventcommand')
        return 1
    success('Successfully triggered the eventcommand')
    failure = False

    info('Checking host macros...')
    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
        fail('Escaping environment variables seems not to properly working')
        fail(' Expected: {0!r} Got: {1!r}'.format(host_info['name']*2,
                                                  event_output[0]))
    for i, column in enumerate(LS_HOST_COLUMNS[1:], 1):
        if column is not None:
            output_value = convert_output(event_output[i])
            if output_value != host_info[column]:
                failure = True
                fail('{0!r} is unequal to {1!r} ({2})'.format(output_value,
                                                              host_info[column],
                                                              column))

    info('Checking service macros...')
    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:
            output_value = convert_output(event_output[i])
            if output_value != svc_info[column]:
                failure = True
                fail('{0!r} is unequal to {1!r} ({2})'.format(output_value,
                                                              svc_info[column],
                                                              column))

    info('Checking global macros...')

    info('Checking command macros...')
    if convert_output(event_output[-1]) != 1337:
        failure = True
        fail('The command macro "custom_macro" is not being substituted')

    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
        fail('Could not disable the eventcommand')
    else:
        success('Successfully disabled the eventcommand')

    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
        fail('Could not re-enable the eventcommand')
    else:
        success('Successfully re-enabled the eventcommand')

    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
        fail('Could not remove eventcommand "test_event"'
             ' assigned to host "localhost"')
    else:
        success('Successfully removed the eventcommand'
                ' assigned to host "localhost"')

    return 1 if failure else 0


if __name__ == '__main__':
    with utils.LiveStatusSocket('/var/run/icinga2/cmd/livestatus') as LIVESTATUS:
        sys.exit(main())