--- /dev/null
+from __future__ import unicode_literals
+
+from datetime import datetime, timedelta
+
+
+TABLE_PREFIX = 'icinga_'
+TABLES = [
+ # Central tables
+ 'instances',
+ 'objects',
+ # Debugging tables
+ 'conninfo',
+ # Historical tables
+ 'acknowledgements',
+ 'commenthistory',
+ 'contactnotifications',
+ 'dbversion',
+ 'downtimehistory',
+ 'eventhandlers',
+ 'externalcommands',
+ 'flappinghistory',
+ 'hostchecks',
+ 'logentries',
+ 'notifications',
+ 'processevents',
+ 'servicechecks',
+ 'statehistory',
+ 'systemcommands',
+ # Current status tables
+ 'comments',
+ 'customvariablestatus',
+ 'hoststatus',
+ 'programstatus',
+ 'runtimevariables',
+ 'scheduleddowntime',
+ 'servicestatus',
+ 'contactstatus',
+ # Configuration tables
+ 'commands',
+ 'configfiles',
+ 'configfilevariables',
+ 'contact_addresses',
+ 'contact_notificationcommands',
+ 'contactgroup_members',
+ 'contactgroups',
+ 'contactnotificationmethods',
+ 'contacts',
+ 'customvariables',
+ 'host_contactgroups',
+ 'host_contacts',
+ 'host_parenthosts',
+ 'hostdependencies',
+ 'hostescalation_contactgroups',
+ 'hostescalation_contacts',
+ 'hostescalations',
+ 'hostgroup_members',
+ 'hostgroups',
+ 'hosts',
+ 'service_contactgroups',
+ 'service_contacts',
+ 'servicedependencies',
+ 'serviceescalation_contactgroups',
+ 'serviceescalation_contacts',
+ 'serviceescalations',
+ 'servicegroup_members',
+ 'servicegroups',
+ 'services',
+ 'timeperiod_timeranges',
+ 'timeperiods'
+ ]
+EXAMPLE_CONFIG = {
+ 'localhost': ['disk', 'http', 'icinga', 'load', 'ping4',
+ 'ping6', 'processes', 'ssh', 'users'],
+ 'nsca-ng': ['PassiveService1', 'PassiveService2']
+}
+
+
+def validate_tables(tables):
+ """
+ Return whether all tables of the IDO database scheme exist in
+ the given table listing
+
+ """
+ missing = [n for n in TABLES if TABLE_PREFIX + n not in tables]
+ if missing:
+ print 'Some tables are missing in the IDO'
+ print 'Missing tables: ' + ', '.join(missing)
+ return False
+
+ print 'All tables were found in the IDO'
+ return True
+
+
+def verify_host_config(config_data):
+ """
+ Return whether the example hosts exist in the given "hosts" table
+
+ """
+ if len([1 for e in config_data if e['alias'] in EXAMPLE_CONFIG]) == 2:
+ print 'All example hosts are stored in the IDO'
+ return True
+
+ print 'Some example hosts are missing in the IDO'
+ return False
+
+
+def verify_service_config(config_data):
+ """
+ Return whether the example services exist in the given "services" table
+
+ """
+ for hostname, servicename in ((h, s) for h, ss in EXAMPLE_CONFIG.iteritems()
+ for s in ss):
+ # Not very efficient, but suitable for just two hosts...
+ if not any(1 for c in config_data
+ if c['alias'] == hostname and
+ c['display_name'] == servicename):
+ print 'The config stored in the IDO is missing some services'
+ return False
+
+ print 'The service config stored in the IDO is correct'
+ return True
+
+
+def check_last_host_status_update(check_info):
+ """
+ Return whether the example hosts are checked as scheduled
+
+ """
+ for info in check_info:
+ if info['alias'] == 'localhost':
+ last_check = datetime.fromtimestamp(float(info['last_check']))
+ if datetime.now() - last_check > timedelta(minutes=5, seconds=10):
+ print 'The last status update of host "localhost"' \
+ ' was more than 5 minutes ago'
+ return False
+ elif info['alias'] == 'nsca-ng':
+ if float(info['last_check']) > 0:
+ print 'The host "nsca-ng" was checked even though' \
+ ' it should not be actively checked'
+ return False
+
+ print 'The updates of both example hosts are processed as configured'
+ return True
+
+
+def check_last_service_status_update(check_info):
+ """
+ Return whether the example services are checked as scheduled
+
+ """
+ for info in check_info:
+ if info['display_name'] in EXAMPLE_CONFIG.get(info['alias'], []):
+ last_check = datetime.fromtimestamp(float(info['last_check']))
+ if datetime.now() - last_check > timedelta(minutes=5, seconds=10):
+ print 'The last status update of service "{0}" of' \
+ ' host "{1}" was more than 5 minutes ago' \
+ ''.format(info['display_name'], info['alias'])
+ return False
+
+ print 'The updates of all example services are processed as configured'
+ return True
+
+
+def check_logentries(logentry_info):
+ """
+ Return whether the given logentry originates from host "localhost"
+ and refers to its very last hard status change
+
+ """
+ if logentry_info and logentry_info[0]['alias'] == 'localhost':
+ entry_time = datetime.fromtimestamp(float(logentry_info[0]['entry_time']))
+ state_time = datetime.fromtimestamp(float(logentry_info[0]['state_time']))
+ if entry_time - state_time > timedelta(seconds=10):
+ print 'The last hard state of host "localhost"' \
+ ' seems not to have been logged'
+ return False
+ else:
+ print 'No logs found in the IDO for host "localhost"'
+ return False
+
+ print 'The last hard state of host "localhost" was properly logged'
+ return True
+
--- /dev/null
+#!/usr/bin/env python
+from __future__ import unicode_literals
+
+import sys
+import subprocess
+
+import ido_tests
+
+MYSQL = b"/usr/bin/mysql"
+PARAMS = b"-t -D icinga -u icinga --password=icinga -e".split()
+SEPARATOR = '|'
+
+
+def run_query(query):
+ p = subprocess.Popen([MYSQL] + PARAMS + [query.encode('utf-8')],
+ stdout=subprocess.PIPE)
+ return parse_result([l.decode('utf-8') for l in p.stdout.readlines()])
+
+
+def parse_result(resultset):
+ result, header = [], None
+ for line in (l for l in resultset if SEPARATOR in l):
+ columns = [c.strip() for c in line[1:-3].split(SEPARATOR)]
+ if header is None:
+ header = columns
+ else:
+ result.append(dict((header[i], v) for i, v in enumerate(columns)))
+ return result
+
+
+def main():
+ if not ido_tests.validate_tables([d['Tables_in_icinga']
+ for d in run_query('show tables')]):
+ return 1
+
+ host_info = run_query('select * from icinga_hosts')
+ if not ido_tests.verify_host_config(host_info):
+ return 1
+
+ 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
+
+ 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):
+ return 1
+
+ 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):
+ return 1
+
+ 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):
+ return 1
+
+ return 0
+
+
+if __name__ == '__main__':
+ sys.exit(main())
+
--- /dev/null
+#!/usr/bin/env python
+from __future__ import unicode_literals
+
+import sys
+import subprocess
+
+import ido_tests
+
+PSQL = b"/usr/bin/psql"
+PARAMS = b"-nq -U icinga -d icinga -c".split()
+SEPARATOR = '|'
+ENVIRONMENT = {
+ b'PGPASSWORD': b'icinga'
+ }
+
+
+def run_query(query):
+ p = subprocess.Popen([PSQL] + PARAMS + [query.encode('utf-8')],
+ stdout=subprocess.PIPE, env=ENVIRONMENT)
+ return parse_result([l.decode('utf-8') for l in p.stdout.readlines()])
+
+
+def parse_result(resultset):
+ result, header = [], None
+ for line in (l for l in resultset if SEPARATOR in l):
+ columns = [c.strip() for c in line.split(SEPARATOR)]
+ if header is None:
+ header = columns
+ else:
+ result.append(dict((header[i], v) for i, v in enumerate(columns)))
+ return result
+
+
+def main():
+ if not ido_tests.validate_tables([d['Name'] for d in run_query('\\dt')
+ if d['Type'] == 'table']):
+ return 1
+
+ host_info = run_query('select * from icinga_hosts')
+ if not ido_tests.verify_host_config(host_info):
+ return 1
+
+ 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
+
+ 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):
+ return 1
+
+ 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):
+ return 1
+
+ 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 '
+ 'inner join icinga_statehistory as hist'
+ ' on hist.object_id = hosts.host_object_id '
+ "where hosts.alias = 'localhost' and hist.state_type = 1 "
+ 'group by hosts.alias'
+ )
+ if not ido_tests.check_logentries(logentry_info):
+ return 1
+
+ return 0
+
+
+if __name__ == '__main__':
+ sys.exit(main())
+
"test_root": "/tmp"
},
"setups": {
+ "^ido_[a-z]{2}sql.test$": {
+ "setup": {
+ "copy": ["files/ido_tests.py >> /tmp/ido_tests.py"]
+ },
+ "teardown": {
+ "clean": ["/tmp/ido_tests.py", "/tmp/ido_tests.pyc"]
+ }
+ }
}
}
projects / icinga2 / commitdiff