test/test_problem_impact.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
  26 from shinken_test import *
  27
  28
  29 class TestConfig(ShinkenTest):
  30     def setUp(self):
  31         self.setup_with_file('etc/nagios_problem_impact.cfg')
  32
  33
  34     def test_problems_impacts(self):
  35         self.print_header()
  36         # retry_interval 2
  37         # critical notification
  38         # run loop -> another notification
  39
  40         #First initialize routers 0 and 1
  41         now = time.time()
  42
  43         #The problem_impact_state change should be enabled in the configuration
  44         self.assert_(self.conf.enable_problem_impacts_states_change == True)
  45
  46         host_router_0 = self.sched.hosts.find_by_name("test_router_0")
  47         host_router_0.checks_in_progress = []
  48         host_router_1 = self.sched.hosts.find_by_name("test_router_1")
  49         host_router_1.checks_in_progress = []
  50
  51         #Then initialize host under theses routers
  52         host_0 = self.sched.hosts.find_by_name("test_host_0")
  53         host_0.checks_in_progress = []
  54         host_1 = self.sched.hosts.find_by_name("test_host_1")
  55         host_1.checks_in_progress = []
  56
  57         all_hosts = [host_router_0, host_router_1, host_0, host_1]
  58         all_routers = [host_router_0, host_router_1]
  59         all_servers = [host_0, host_1]
  60
  61         #--------------------------------------------------------------
  62         # initialize host states as UP
  63         #--------------------------------------------------------------
  64         print "- 4 x UP -------------------------------------"
  65         self.scheduler_loop(1, [[host_router_0, 0, 'UP'], [host_router_1, 0, 'UP'], [host_0, 0, 'UP'], [host_1, 0, 'UP']], do_sleep=False)
  66
  67         for h in all_hosts:
  68             self.assert_(h.state == 'UP')
  69             self.assert_(h.state_type == 'HARD')
  70
  71         #--------------------------------------------------------------
  72         # Now we add some problems to routers
  73         #--------------------------------------------------------------
  74         print "- routers get DOWN /SOFT-------------------------------------"
  75         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  76         #Max attempt is at 5, should be soft now
  77         for h in all_routers:
  78             self.assert_(h.state == 'DOWN')
  79             self.assert_(h.state_type == 'SOFT')
  80
  81         print "- routers get DOWN /HARD-------------------------------------"
  82         #Now put 4 more checks so we get DOWN/HARD
  83         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  84         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  85         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  86         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  87
  88         #Max attempt is reach, should be HARD now
  89         for h in all_routers:
  90             self.assert_(h.state == 'DOWN')
  91             self.assert_(h.state_type == 'HARD')
  92
  93
  94         #--------------------------------------------------------------
  95         # Routers get HARD/DOWN
  96         # should be problems now!
  97         #--------------------------------------------------------------
  98         #Now check in the brok generation too
  99         host_router_0_brok = host_router_0.get_update_status_brok()
 100         host_router_1_brok = host_router_1.get_update_status_brok()
 101
 102         #Should be problems and have sub servers as impacts
 103         for h in all_routers:
 104             self.assert_(h.is_problem == True)
 105             for s in all_servers:
 106                 self.assert_(s in h.impacts)
 107                 self.assert_(s.get_dbg_name() in host_router_0_brok.data['impacts']['hosts'])
 108                 self.assert_(s.get_dbg_name() in host_router_1_brok.data['impacts']['hosts'])
 109
 110         #Now impacts should really be .. impacts :)
 111         for s in all_servers:
 112             self.assert_(s.is_impact == True)
 113             self.assert_(s.state == 'UNREACHABLE')
 114             #And check the services are impacted too
 115             for svc in s.services:
 116                 print "Service state", svc.state
 117                 self.assert_(svc.state == 'UNKNOWN')
 118                 self.assert_(svc.get_dbg_name() in host_router_0_brok.data['impacts']['services'])
 119                 self.assert_(svc.get_dbg_name() in host_router_1_brok.data['impacts']['services'])
 120                 brk_svc = svc.get_update_status_brok()
 121                 self.assert_(brk_svc.data['source_problems']['hosts'] == ['test_router_0', 'test_router_1'])
 122             for h in all_routers:
 123                 self.assert_(h in s.source_problems)
 124                 brk_hst = s.get_update_status_brok()
 125                 self.assert_(h.get_dbg_name() in brk_hst.data['source_problems']['hosts'])
 126
 127
 128         #--------------------------------------------------------------
 129         # One router get UP now
 130         #--------------------------------------------------------------
 131         print "- 1 X UP for a router ------------------------------"
 132         #Ok here the problem/impact propagation is Checked. Now what
 133         #if one router get back? :)
 134         self.scheduler_loop(1, [[host_router_0, 0, 'UP']], do_sleep=False)
 135
 136         #should be UP/HARD now
 137         self.assert_(host_router_0.state == 'UP')
 138         self.assert_(host_router_0.state_type == 'HARD')
 139
 140         #And should not be a problem any more!
 141         self.assert_(host_router_0.is_problem == False)
 142         self.assert_(host_router_0.impacts == [])
 143
 144         #And check if it's no more in sources problems of others servers
 145         for s in all_servers:
 146             #Still impacted by the other server
 147             self.assert_(s.is_impact == True)
 148             self.assert_(s.source_problems == [host_router_1])
 149
 150
 151         #--------------------------------------------------------------
 152         # The other router get UP :)
 153         #--------------------------------------------------------------
 154         print "- 1 X UP for the last router ------------------------------"
 155         #What is the last router get back? :)
 156         self.scheduler_loop(1, [[host_router_1, 0, 'UP']], do_sleep=False)
 157
 158         #should be UP/HARD now
 159         self.assert_(host_router_1.state == 'UP')
 160         self.assert_(host_router_1.state_type == 'HARD')
 161
 162         #And should not be a problem any more!
 163         self.assert_(host_router_1.is_problem == False)
 164         self.assert_(host_router_1.impacts == [])
 165
 166         #And check if it's no more in sources problems of others servers
 167         for s in all_servers:
 168             #Still impacted by the other server
 169             self.assert_(s.is_impact == False)
 170             self.assert_(s.state == 'UP')
 171             self.assert_(s.source_problems == [])
 172
 173         #It's done :)
 174
 175
 176 if __name__ == '__main__':
 177     unittest.main()