granicus.if.org Git - icinga2/blob - test/jenkins/ido_pgsql.test

   1 #!/usr/bin/env python
   2 from __future__ import unicode_literals
   3
   4 import sys
   5
   6 import utils
   7 import ido_tests
   8
   9
  10 def main():
  11     failures = False
  12
  13     run_query = lambda q: utils.run_pgsql_query(q, b'/usr/bin/psql')
  14
  15     if not ido_tests.validate_tables([d['Name'] for d in run_query('\\dt')
  16                                                 if d['Type'] == 'table']):
  17         return 1 # Bail out as we cannot proceed without any data
  18
  19     host_info = run_query('select * from icinga_hosts')
  20     if not ido_tests.verify_host_config(host_info):
  21         return 1 # Bail out as we cannot proceed without any data
  22
  23     service_info = run_query(
  24         'select c2.alias, c1.* from icinga_services as c1 '
  25         'inner join icinga_hosts as c2'
  26         ' on c1.host_object_id = c2.host_object_id'
  27         )
  28     if not ido_tests.verify_service_config(service_info):
  29         return 1 # Bail out as we cannot proceed without any data
  30
  31     hostchecks_data = run_query(
  32         'select c.alias, unix_timestamp(s.last_check) as last_check'
  33         ' from icinga_hoststatus as s '
  34         'inner join icinga_hosts as c'
  35         ' on s.host_object_id = c.host_object_id'
  36         )
  37     if not ido_tests.check_last_host_status_update(hostchecks_data):
  38         failures = True
  39
  40     servicechecks_data = run_query(
  41         'select c2.alias, c1.display_name, unix_timestamp(s.last_check) as last_check'
  42         ' from icinga_servicestatus as s '
  43         'inner join icinga_services as c1'
  44         ' on s.service_object_id = c1.service_object_id '
  45         'inner join icinga_hosts as c2'
  46         ' on c1.host_object_id = c2.host_object_id'
  47         )
  48     if not ido_tests.check_last_service_status_update(servicechecks_data):
  49         failures = True
  50
  51     logentry_info = run_query(
  52         'select hosts.alias,'
  53         '       max(unix_timestamp(logs.entry_time)) as entry_time,'
  54         '       max(unix_timestamp(hist.state_time)) as state_time'
  55         ' from icinga_logentries as logs '
  56         'inner join icinga_hosts as hosts'
  57         ' on logs.object_id = hosts.host_object_id '
  58         'inner join icinga_statehistory as hist'
  59         ' on hist.object_id = hosts.host_object_id '
  60         "where hosts.alias = 'localhost' and hist.state_type = 1 "
  61         'group by hosts.alias'
  62         )
  63     if not ido_tests.check_logentries(logentry_info):
  64         failures = True
  65
  66     return 1 if failures else 0
  67
  68
  69 if __name__ == '__main__':
  70     sys.exit(main())
  71