icinga2/test/jenkins/ido_mysql.test

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

import sys

import utils
import ido_tests


def main():
    failures = False

    run_query = lambda q: utils.run_mysql_query(q, b'/usr/bin/mysql')

    if not ido_tests.validate_tables([d['Tables_in_icinga']
                                      for d in run_query('show tables')]):
        return 1 # Bail out as we cannot proceed without any data

    host_info = run_query('select * from icinga_hosts')
    if not ido_tests.verify_host_config(host_info):
        return 1 # Bail out as we cannot proceed without any data

    service_info = run_query(
        'select c2.alias, c1.* from icinga_services as c1 '
        'inner join icinga_hosts as c2'
        ' on c1.host_object_id = c2.host_object_id'
        )
    if not ido_tests.verify_service_config(service_info):
        return 1 # Bail out as we cannot proceed without any data

    hostchecks_data = run_query(
        'select c.alias, unix_timestamp(s.last_check) as last_check'
        ' from icinga_hoststatus as s '
        'inner join icinga_hosts as c'
        ' on s.host_object_id = c.host_object_id'
        )
    if not ido_tests.check_last_host_status_update(hostchecks_data):
        failures = True

    servicechecks_data = run_query(
        'select c2.alias, c1.display_name, unix_timestamp(s.last_check) as last_check'
        ' from icinga_servicestatus as s '
        'inner join icinga_services as c1'
        ' on s.service_object_id = c1.service_object_id '
        'inner join icinga_hosts as c2'
        ' on c1.host_object_id = c2.host_object_id'
        )
    if not ido_tests.check_last_service_status_update(servicechecks_data):
        failures = True

    logentry_info = run_query(
        'select hosts.alias,'
        '       max(unix_timestamp(logs.entry_time)) as entry_time,'
        '       max(unix_timestamp(hist.state_time)) as state_time'
        ' from icinga_logentries as logs '
        'inner join icinga_hosts as hosts'
        ' on logs.object_id = hosts.host_object_id and hosts.alias = "localhost" '
        'inner join icinga_statehistory as hist'
        ' on hist.object_id = hosts.host_object_id and hist.state_type = 1'
        )
    if not ido_tests.check_logentries(logentry_info):
        failures = True

    return 1 if failures else 0


if __name__ == '__main__':
    sys.exit(main())