test/test_dependencies.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 host- and service-downtimes.
  23 #
  24
  25 from shinken_test import *
  26 sys.setcheckinterval(10000)
  27
  28
  29
  30 class TestConfig(ShinkenTest):
  31     def setUp(self):
  32         self.setup_with_file('etc/nagios_dependencies.cfg')
  33
  34
  35     def test_service_dependencies(self):
  36         self.print_header()
  37         now = time.time()
  38         test_host_0 = self.sched.hosts.find_by_name("test_host_0")
  39         test_host_1 = self.sched.hosts.find_by_name("test_host_1")
  40         test_host_0.checks_in_progress = []
  41         test_host_1.checks_in_progress = []
  42         test_host_0.act_depend_of = [] # ignore the router
  43         test_host_1.act_depend_of = [] # ignore the router
  44         router = self.sched.hosts.find_by_name("test_router_0")
  45         router.checks_in_progress = []
  46         router.act_depend_of = [] # ignore other routers
  47         test_host_0_test_ok_0 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
  48         test_host_0_test_ok_1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_1")
  49         test_host_1_test_ok_0 = self.sched.services.find_srv_by_name_and_hostname("test_host_1", "test_ok_0")
  50         test_host_1_test_ok_1 = self.sched.services.find_srv_by_name_and_hostname("test_host_1", "test_ok_1")
  51         # the most important: test_ok_0 is in the chk_depend_of-list of test_ok_1
  52         self.assert_(test_host_0_test_ok_0 in [x[0] for x in test_host_0_test_ok_1.chk_depend_of])
  53         self.assert_(test_host_1_test_ok_0 in [x[0] for x in test_host_1_test_ok_1.chk_depend_of])
  54
  55         # and not vice versa
  56         self.assert_(test_host_0_test_ok_1 not in [x[0] for x in test_host_0_test_ok_0.chk_depend_of])
  57         self.assert_(test_host_1_test_ok_1 not in [x[0] for x in test_host_1_test_ok_0.chk_depend_of])
  58
  59         # test_ok_0 is also in the act_depend_of-list of test_ok_1
  60         self.assert_(test_host_0_test_ok_0 in [x[0] for x in test_host_0_test_ok_1.chk_depend_of])
  61         self.assert_(test_host_1_test_ok_0 in [x[0] for x in test_host_1_test_ok_1.chk_depend_of])
  62
  63         # check the criteria
  64         # execution_failure_criteria      u,c
  65         # notification_failure_criteria   u,c,w
  66         self.assert_([['u', 'c']] == [x[1] for x in test_host_0_test_ok_1.chk_depend_of if x[0] is test_host_0_test_ok_0])
  67         self.assert_([['u', 'c']] == [x[1] for x in test_host_1_test_ok_1.chk_depend_of if x[0] is test_host_1_test_ok_0])
  68         self.assert_([['u', 'c', 'w']] == [x[1] for x in test_host_0_test_ok_1.act_depend_of if x[0] is test_host_0_test_ok_0])
  69         self.assert_([['u', 'c', 'w']] == [x[1] for x in test_host_1_test_ok_1.act_depend_of if x[0] is test_host_1_test_ok_0])
  70
  71         # and every service has the host in it's act_depend_of-list
  72         self.assert_(test_host_0 in [x[0] for x in test_host_0_test_ok_0.act_depend_of])
  73         self.assert_(test_host_0 in [x[0] for x in test_host_0_test_ok_1.act_depend_of])
  74         self.assert_(test_host_1 in [x[0] for x in test_host_1_test_ok_0.act_depend_of])
  75         self.assert_(test_host_1 in [x[0] for x in test_host_1_test_ok_1.act_depend_of])
  76
  77         # and final count the masters
  78         self.assert_(len(test_host_0_test_ok_0.chk_depend_of) == 0)
  79         self.assert_(len(test_host_0_test_ok_1.chk_depend_of) == 1)
  80         self.assert_(len(test_host_1_test_ok_0.chk_depend_of) == 0)
  81         self.assert_(len(test_host_1_test_ok_1.chk_depend_of) == 1)
  82         self.assert_(len(test_host_0_test_ok_0.act_depend_of) == 1) # same, plus the host
  83         self.assert_(len(test_host_0_test_ok_1.act_depend_of) == 2)
  84         self.assert_(len(test_host_1_test_ok_0.act_depend_of) == 1)
  85         self.assert_(len(test_host_1_test_ok_1.act_depend_of) == 2)
  86
  87
  88     def test_host_dependencies(self):
  89         self.print_header()
  90         now = time.time()
  91         #
  92         #   A  <------  B  <--
  93         #   ^                 \---  C
  94         #   |---------------------
  95         #
  96         host_A = self.sched.hosts.find_by_name("test_host_A")
  97         host_B = self.sched.hosts.find_by_name("test_host_B")
  98         host_C = self.sched.hosts.find_by_name("test_host_C")
  99         host_D = self.sched.hosts.find_by_name("test_host_D")
 100
 101
 102         # the most important: test_ok_0 is in the chk_depend_of-list of test_ok_1
 103         #self.assert_(host_A in [x[0] for x in host_C.chk_depend_of])
 104         print host_C.act_depend_of
 105         print host_C.chk_depend_of
 106         print host_C.chk_depend_of_me
 107         self.assert_(host_B in [x[0] for x in host_C.act_depend_of])
 108         self.assert_(host_A in [x[0] for x in host_C.act_depend_of])
 109         self.assert_(host_A in [x[0] for x in host_B.act_depend_of])
 110         self.assert_(host_A.act_depend_of == [])
 111         self.assert_(host_B in [x[0] for x in host_C.chk_depend_of])
 112         self.assert_(host_A in [x[0] for x in host_C.chk_depend_of])
 113         self.assert_(host_A in [x[0] for x in host_B.chk_depend_of])
 114         self.assert_(host_A.act_depend_of == [])
 115         self.assert_(host_B in [x[0] for x in host_A.act_depend_of_me])
 116         self.assert_(host_C in [x[0] for x in host_A.act_depend_of_me])
 117         self.assert_(host_C in [x[0] for x in host_B.act_depend_of_me])
 118         #self.assert_(host_C.act_depend_of_me == []) # D in here
 119         self.assert_(host_B in [x[0] for x in host_A.chk_depend_of_me])
 120         self.assert_(host_C in [x[0] for x in host_A.chk_depend_of_me])
 121         self.assert_(host_C in [x[0] for x in host_B.chk_depend_of_me])
 122         self.assert_(host_D in [x[0] for x in host_C.chk_depend_of_me])
 123
 124         # check the notification/execution criteria
 125         self.assert_([['d', 'u']] == [x[1] for x in host_C.act_depend_of if x[0] is host_B])
 126         self.assert_([['d']] == [x[1] for x in host_C.chk_depend_of if x[0] is host_B])
 127         self.assert_([['d', 'u']] == [x[1] for x in host_C.act_depend_of if x[0] is host_A])
 128         self.assert_([['d']] == [x[1] for x in host_C.chk_depend_of if x[0] is host_A])
 129         self.assert_([['d', 'u']] == [x[1] for x in host_B.act_depend_of if x[0] is host_A])
 130         self.assert_([['n']] == [x[1] for x in host_B.chk_depend_of if x[0] is host_A])
 131
 132
 133     def test_host_inherits_dependencies(self):
 134         self.print_header()
 135         now = time.time()
 136         #
 137         #   A  <------  B  <--
 138         #   ^                 \---  C   <--  D
 139         #   |---------------------
 140         #
 141         host_A = self.sched.hosts.find_by_name("test_host_A")
 142         host_B = self.sched.hosts.find_by_name("test_host_B")
 143         host_C = self.sched.hosts.find_by_name("test_host_C")
 144         host_D = self.sched.hosts.find_by_name("test_host_D")
 145
 146         print "A depends on", ",".join([x[0].get_name() for x in host_A.chk_depend_of])
 147         print "B depends on", ",".join([x[0].get_name() for x in host_B.chk_depend_of])
 148         print "C depends on", ",".join([x[0].get_name() for x in host_C.chk_depend_of])
 149         print "D depends on", ",".join([x[0].get_name() for x in host_D.chk_depend_of])
 150
 151         self.assert_(host_A.act_depend_of == [])
 152         self.assert_(host_A in [x[0] for x in host_B.act_depend_of])
 153         self.assert_(host_A in [x[0] for x in host_C.act_depend_of])
 154         self.assert_(host_B in [x[0] for x in host_C.act_depend_of])
 155         self.assert_(host_C in [x[0] for x in host_D.act_depend_of])
 156         # and through inherits_parent....
 157         #self.assert_(host_A in [x[0] for x in host_D.act_depend_of])
 158         #self.assert_(host_B in [x[0] for x in host_D.act_depend_of])
 159
 160
 161     #Now test a in service service_dep definition. More easierto use than create a full new object
 162     def test_in_servicedef_dep(self):
 163         svc_parent = self.sched.services.find_srv_by_name_and_hostname("test_host_1", "test_parent_svc")
 164         svc_son = self.sched.services.find_srv_by_name_and_hostname("test_host_1", "test_son_svc")
 165
 166         print "DumP", self.conf.servicedependencies
 167
 168         # the most important: test_parent is in the chk_depend_of-list of test_son
 169         print "Dep : ", svc_son.act_depend_of
 170         self.assert_([['u', 'c', 'w']] == [x[1] for x in svc_son.act_depend_of if x[0] is svc_parent])
 171
 172
 173
 174     def test_host_non_inherits_dependencies(self):
 175         #
 176         #   A  <------  B  <--
 177         #   ^                 \NOT/---  C   <--  D
 178         #   |---------------------
 179         #
 180         host_A = self.sched.hosts.find_by_name("test_host_A")
 181         host_B = self.sched.hosts.find_by_name("test_host_B")
 182         host_C = self.sched.hosts.find_by_name("test_host_C")
 183         host_D = self.sched.hosts.find_by_name("test_host_D")
 184         host_E = self.sched.hosts.find_by_name("test_host_E")
 185
 186
 187         print "A depends on", ",".join([x[0].get_name() for x in host_A.chk_depend_of])
 188         print "B depends on", ",".join([x[0].get_name() for x in host_B.chk_depend_of])
 189         print "C depends on", ",".join([x[0].get_name() for x in host_C.chk_depend_of])
 190         print "D depends on", ",".join([x[0].get_name() for x in host_D.chk_depend_of])
 191         print "E depends on", ",".join([x[0].get_name() for x in host_E.chk_depend_of])
 192
 193         host_C.state = 'DOWN'
 194         print "D state", host_D.state
 195         print "E dep", host_E.chk_depend_of
 196         print "I raise?", host_D.do_i_raise_dependency('d', inherit_parents=False)
 197         #If I ask D for dep, he should raise Nothing if we do not want parents.
 198         self.assert_(host_D.do_i_raise_dependency('d', inherit_parents=False) == False)
 199         #But he should raise a problem (C here) of we ask for its parents
 200         self.assert_(host_D.do_i_raise_dependency('d', inherit_parents=True) == True)
 201
 202
 203
 204
 205 if __name__ == '__main__':
 206     import cProfile
 207     command = """unittest.main()"""
 208     unittest.main()
 209     #cProfile.runctx( command, globals(), locals(), filename="Thruk.profile" )
 210