test/test_services.py

   1 #!/usr/bin/env python2.6
   2 #Copyright (C) 2009-2010 :
   3 #    Gabes Jean, naparuba@gmail.com
   4 #    Gerhard Lausser, Gerhard.Lausser@consol.de
   5 #
   6 #This file is part of Shinken.
   7 #
   8 #Shinken is free software: you can redistribute it and/or modify
   9 #it under the terms of the GNU Affero General Public License as published by
  10 #the Free Software Foundation, either version 3 of the License, or
  11 #(at your option) any later version.
  12 #
  13 #Shinken is distributed in the hope that it will be useful,
  14 #but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 #MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 #GNU Affero General Public License for more details.
  17 #
  18 #You should have received a copy of the GNU Affero General Public License
  19 #along with Shinken.  If not, see <http://www.gnu.org/licenses/>.
  20
  21 #
  22 # This file is used to test reading and processing of config files
  23 #
  24
  25 import copy
  26 #It's ugly I know....
  27 from shinken_test import *
  28
  29
  30 class TestConfig(ShinkenTest):
  31     #setUp is in shinken_test
  32
  33     def get_svc(self):
  34         return self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
  35
  36
  37     #Look if get_*_name return the good result
  38     def test_get_name(self):
  39         svc = self.get_svc()
  40         print svc.get_dbg_name()
  41         self.assert_(svc.get_name() == 'test_ok_0')
  42         self.assert_(svc.get_dbg_name() == 'test_host_0/test_ok_0')
  43
  44     #getstate should be with all properties in dict class + id
  45     #check also the setstate
  46     def test___getstate__(self):
  47         svc = self.get_svc()
  48         cls = svc.__class__
  49         #We get teh state
  50         state = svc.__getstate__()
  51         #Check it's the good lenght
  52         self.assert_(len(state) == len(cls.properties) + len(cls.running_properties) + 1)
  53         #we copy the service
  54         svc_copy = copy.copy(svc)
  55         #reset the state in the original service
  56         svc.__setstate__(state)
  57         #And it should be the same :then before :)
  58         for p in cls.properties:
  59 #            print getattr(svc_copy, p)
  60 #            print getattr(svc, p)
  61             self.assert_(getattr(svc_copy, p) == getattr(svc, p))
  62
  63
  64     #Look if it can detect all incorrect cases
  65     def test_is_correct(self):
  66         svc = self.get_svc()
  67
  68         #first it's ok
  69         self.assert_(svc.is_correct() == True)
  70
  71         #Now try to delete a required property
  72         max_check_attempts = svc.max_check_attempts
  73         del svc.max_check_attempts
  74         self.assert_(svc.is_correct() == False)
  75         svc.max_check_attempts = max_check_attempts
  76
  77         ###
  78         ### Now special cases
  79         ###
  80
  81         #no check command
  82         check_command = svc.check_command
  83         del svc.check_command
  84         self.assert_(svc.is_correct() == False)
  85         svc.check_command = check_command
  86         self.assert_(svc.is_correct() == True)
  87
  88         #no notification_interval
  89         notification_interval = svc.notification_interval
  90         del svc.notification_interval
  91         self.assert_(svc.is_correct() == False)
  92         svc.notification_interval = notification_interval
  93         self.assert_(svc.is_correct() == True)
  94
  95
  96     #Look for set/unset impacted states (unknown)
  97     def test_impact_state(self):
  98         svc = self.get_svc()
  99         ori_state = svc.state
 100         ori_state_id = svc.state_id
 101         svc.set_impact_state()
 102         self.assert_(svc.state == 'UNKNOWN')
 103         self.assert_(svc.state_id == 3)
 104         svc.unset_impact_state()
 105         self.assert_(svc.state == ori_state)
 106         self.assert_(svc.state_id == ori_state_id)
 107
 108     def test_set_state_from_exit_status(self):
 109         svc = self.get_svc()
 110         #First OK
 111         svc.set_state_from_exit_status(0)
 112         self.assert_(svc.state == 'OK')
 113         self.assert_(svc.state_id == 0)
 114         self.assert_(svc.is_state('OK') == True)
 115         self.assert_(svc.is_state('o') == True)
 116         #Then warning
 117         svc.set_state_from_exit_status(1)
 118         self.assert_(svc.state == 'WARNING')
 119         self.assert_(svc.state_id == 1)
 120         self.assert_(svc.is_state('WARNING') == True)
 121         self.assert_(svc.is_state('w') == True)
 122         #Then Critical
 123         svc.set_state_from_exit_status(2)
 124         self.assert_(svc.state == 'CRITICAL')
 125         self.assert_(svc.state_id == 2)
 126         self.assert_(svc.is_state('CRITICAL') == True)
 127         self.assert_(svc.is_state('c') == True)
 128         #And unknown
 129         svc.set_state_from_exit_status(3)
 130         self.assert_(svc.state == 'UNKNOWN')
 131         self.assert_(svc.state_id == 3)
 132         self.assert_(svc.is_state('UNKNOWN') == True)
 133         self.assert_(svc.is_state('u') == True)
 134
 135         #And something else :)
 136         svc.set_state_from_exit_status(99)
 137         self.assert_(svc.state == 'CRITICAL')
 138         self.assert_(svc.state_id == 2)
 139         self.assert_(svc.is_state('CRITICAL') == True)
 140         self.assert_(svc.is_state('c') == True)
 141
 142     #Check if the check_freshnes is doing it's job
 143     def test_check_freshness(self):
 144         self.print_header()
 145         #We want an eventhandelr (the perfdata command) to be put in the actions dict
 146         #after we got a service check
 147         now = time.time()
 148         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 149         svc.checks_in_progress = []
 150         svc.act_depend_of = [] # no hostchecks on critical checkresults
 151         #--------------------------------------------------------------
 152         # initialize host/service state
 153         #--------------------------------------------------------------
 154         #We do not want to be just a string but a real command
 155         print "Additonal freshness latency", svc.__class__.additional_freshness_latency
 156         self.scheduler_loop(1, [[svc, 0, 'OK | bibi=99%']])
 157         print "Addi :", svc.last_state_update, svc.freshness_threshold , svc.check_freshness
 158         #By default check fresh ness is set at false, so no new checks
 159         self.assert_(len(svc.actions) == 0)
 160         svc.do_check_freshness()
 161         self.assert_(len(svc.actions) == 0)
 162
 163         #We make it 10s less than it was
 164         svc.last_state_update = svc.last_state_update - 10
 165
 166
 167         #Now we active it, with a too small value (now - 10s is still higer than now - (1 - 15, the addition time)
 168         #So still no check
 169         svc.check_freshness = True
 170         svc.freshness_threshold = 1
 171         print "Addi:", svc.last_state_update, svc.freshness_threshold , svc.check_freshness
 172         svc.do_check_freshness()
 173         self.assert_(len(svc.actions) == 0)
 174
 175         #Ok, now, we remove again 10s. Here we will saw the new entry
 176         svc.last_state_update = svc.last_state_update - 10
 177         svc.do_check_freshness()
 178         self.assert_(len(svc.actions) == 1)
 179         #And we check for the message in the log too
 180         self.assert_(self.log_match(1, 'Warning: The results of service'))
 181
 182
 183     def test_criticity_value(self):
 184         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 185         #This service inherit the improtance value from his father, 5
 186         self.assert_(svc.criticity == 5)
 187
 188
 189     #Look if the service is in the servicegroup
 190     def test_servicegroup(self):
 191         sg = self.sched.servicegroups.find_by_name("servicegroup_01")
 192         self.assert_(sg is not None)
 193         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 194         self.assert_(svc in sg.members)
 195         self.assert_(sg in svc.servicegroups)
 196
 197     #Look at the good of the last_hard_state_change
 198     def test_service_last_hard_state(self):
 199         self.print_header()
 200         #We want an eventhandelr (the perfdata command) to be put in the actions dict
 201         #after we got a service check
 202         now = time.time()
 203         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 204         svc.checks_in_progress = []
 205         svc.act_depend_of = [] # no hostchecks on critical checkresults
 206         #--------------------------------------------------------------
 207         # initialize host/service state
 208         #--------------------------------------------------------------
 209         #We do not want to be just a string but a real command
 210         self.scheduler_loop(1, [[svc, 0, 'OK | bibi=99%']])
 211         print "FUCK", svc.last_hard_state_change
 212         orig = svc.last_hard_state_change
 213         self.assert_(svc.last_hard_state == 'OK')
 214
 215         #now still ok
 216         self.scheduler_loop(1, [[svc, 0, 'OK | bibi=99%']])
 217         self.assert_(svc.last_hard_state_change == orig)
 218         self.assert_(svc.last_hard_state == 'OK')
 219
 220         #now error but still SOFT
 221         self.scheduler_loop(1, [[svc, 2, 'CRITICAL | bibi=99%']])
 222         print "FUCK", svc.state_type
 223         self.assert_(svc.last_hard_state_change == orig)
 224         self.assert_(svc.last_hard_state == 'OK')
 225
 226         #now go hard!
 227         now = int(time.time())
 228         self.scheduler_loop(1, [[svc, 2, 'CRITICAL | bibi=99%']])
 229         print "FUCK", svc.state_type
 230         self.assert_(svc.last_hard_state_change == now)
 231         self.assert_(svc.last_hard_state == 'CRITICAL')
 232         print "Last hard state id", svc.last_hard_state_id
 233         self.assert_(svc.last_hard_state_id == 2)
 234
 235
 236     # Check if the autoslots are fill like it should
 237     def test_autoslots(self):
 238         svc = self.get_svc()
 239         self.assert_("check_period" not in svc.__dict__)
 240
 241
 242     # Check if the parent/childs dependencies are fill like it should
 243     def test_parent_child_dep_list(self):
 244         svc = self.get_svc()
 245         # Look if our host is a parent
 246         self.assert_(svc.host in svc.parent_dependencies)
 247         # and if we are a child of it
 248         self.assert_(svc in svc.host.child_dependencies)
 249
 250
 251 if __name__ == '__main__':
 252     unittest.main()
 253
 254