Add IDO tests

refs #5223
This commit is contained in:
Johannes Meyer 2013-12-06 10:07:55 +01:00
parent 3b04b4dcf5
commit 17e36fe1cc
4 changed files with 369 additions and 0 deletions

View File

@ -0,0 +1,184 @@
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

86
test/jenkins/ido_mysql.test Executable file
View File

@ -0,0 +1,86 @@
#!/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())

91
test/jenkins/ido_pgsql.test Executable file
View File

@ -0,0 +1,91 @@
#!/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())

View File

@ -8,5 +8,13 @@
"test_root": "/tmp" "test_root": "/tmp"
}, },
"setups": { "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"]
}
}
} }
} }