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

   1 #!/usr/bin/env python
   2 from __future__ import unicode_literals
   3
   4 import os
   5 import sys
   6 import time
   7
   8 from ido_mysql import run_query
   9
  10
  11 STATE_OK = 0
  12 TYPE_PASSIVE_CHECK = 1
  13
  14 CHECK_INTERVAL = 300 # seconds
  15 CHECKRESULT_READ_INTERVAL = 5 # seconds
  16 CHECKRESULT_LOCATION = '/tmp/icinga2/checkresults'
  17 CHECKRESULT_TEMPLATE = """
  18 host_name=%(hostname)s
  19 service_description=%(servicename)s
  20 check_type=%(check_type)s
  21 check_options=0
  22 scheduled_check=0
  23 reschedule_check=0
  24 latency=0
  25 start_time=%(start_time)s
  26 finish_time=%(finish_time)s
  27 early_timeout=0
  28 exited_ok=%(excited_ok)s
  29 return_code=%(return_code)s
  30 output=%(output)s
  31 """
  32
  33
  34 def main():
  35     # We need to wait a bit first as Icinga processes a
  36     # checkresult only if its newer than the last check
  37     query = 'select unix_timestamp(s.last_check) as last_check ' \
  38             'from icinga_servicestatus as s ' \
  39             'inner join icinga_services as c ' \
  40             'on s.service_object_id = c.service_object_id ' \
  41             "where c.display_name = 'PassiveService1'"
  42     state_time = float(next(iter(run_query(query)), {}).get('last_check', '0'))
  43     if state_time == 0:
  44         print '"PassiveService1" seems not to have been checked yet'
  45         return 1
  46
  47     if (state_time + CHECK_INTERVAL) - time.time() < 30:
  48         time.sleep(45)
  49
  50     # Now pass the checkresult in
  51     with open(os.path.join(CHECKRESULT_LOCATION, 'cfoobar'), 'w') as f:
  52         f.write(CHECKRESULT_TEMPLATE % {
  53             'hostname': 'nsca-ng',
  54             'servicename': 'PassiveService1',
  55             'check_type': TYPE_PASSIVE_CHECK,
  56             'start_time': time.time(),
  57             'finish_time': time.time(),
  58             'excited_ok': '1',
  59             'return_code': STATE_OK,
  60             'output': 'Passing in CheckResult header files works!'
  61             })
  62
  63     # And notfiy Icinga that the file has been completely written...
  64     with open(os.path.join(CHECKRESULT_LOCATION, 'cfoobar.ok'), 'w') as f:
  65         pass
  66
  67     # Lastly check whether the service changed its state
  68     time.sleep(CHECKRESULT_READ_INTERVAL * 2)
  69
  70     query = 'select s.output ' \
  71             'from icinga_servicestatus as s ' \
  72             'inner join icinga_services as c ' \
  73             'on s.service_object_id = c.service_object_id ' \
  74             "where c.display_name = 'PassiveService1'"
  75     output = next(iter(run_query(query)), {}).get('output', '')
  76     if output != 'Passing in CheckResult header files works!':
  77         print 'Checkresult header files seem not to be processed properly'
  78         return 1
  79
  80     print 'Checkresult header files are processed properly'
  81     return 0
  82
  83
  84 if __name__ == '__main__':
  85     sys.exit(main())