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         self.assert_(host_router_0.criticity == 3)
  49         host_router_1 = self.sched.hosts.find_by_name("test_router_1")
  50         host_router_1.checks_in_progress = []
  51         self.assert_(host_router_1.criticity == 3)
  52
  53         #Then initialize host under theses routers
  54         host_0 = self.sched.hosts.find_by_name("test_host_0")
  55         host_0.checks_in_progress = []
  56         host_1 = self.sched.hosts.find_by_name("test_host_1")
  57         host_1.checks_in_progress = []
  58
  59         all_hosts = [host_router_0, host_router_1, host_0, host_1]
  60         all_routers = [host_router_0, host_router_1]
  61         all_servers = [host_0, host_1]
  62
  63         #--------------------------------------------------------------
  64         # initialize host states as UP
  65         #--------------------------------------------------------------
  66         print "- 4 x UP -------------------------------------"
  67         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)
  68
  69         for h in all_hosts:
  70             self.assert_(h.state == 'UP')
  71             self.assert_(h.state_type == 'HARD')
  72
  73         #--------------------------------------------------------------
  74         # Now we add some problems to routers
  75         #--------------------------------------------------------------
  76         print "- routers get DOWN /SOFT-------------------------------------"
  77         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  78         #Max attempt is at 5, should be soft now
  79         for h in all_routers:
  80             self.assert_(h.state == 'DOWN')
  81             self.assert_(h.state_type == 'SOFT')
  82
  83         print "- routers get DOWN /HARD-------------------------------------"
  84         #Now put 4 more checks so we get DOWN/HARD
  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         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  88         self.scheduler_loop(1, [[host_router_0, 1, 'DOWN'], [host_router_1, 2, 'DOWN']], do_sleep=False)
  89
  90         #Max attempt is reach, should be HARD now
  91         for h in all_routers:
  92             self.assert_(h.state == 'DOWN')
  93             self.assert_(h.state_type == 'HARD')
  94
  95
  96         #--------------------------------------------------------------
  97         # Routers get HARD/DOWN
  98         # should be problems now!
  99         #--------------------------------------------------------------
 100         #Now check in the brok generation too
 101         host_router_0_brok = host_router_0.get_update_status_brok()
 102         host_router_1_brok = host_router_1.get_update_status_brok()
 103
 104         #Should be problems and have sub servers as impacts
 105         for h in all_routers:
 106             self.assert_(h.is_problem == True)
 107             # Now routers are problems, they should have take the max
 108             # criticity value ofthe impacts, so here 5
 109             self.assert_(h.criticity == 5)
 110             for s in all_servers:
 111                 self.assert_(s in h.impacts)
 112                 self.assert_(s.get_dbg_name() in host_router_0_brok.data['impacts']['hosts'])
 113                 self.assert_(s.get_dbg_name() in host_router_1_brok.data['impacts']['hosts'])
 114
 115         # Should have host notification, but it's not so simple:
 116         # our contact say : not under 5, and our hosts are 3. But
 117         # the impacts have huge criticity, so the hosts gain such criticity
 118         self.assert_(self.any_log_match('HOST NOTIFICATION.*;'))
 119         self.show_and_clear_logs()
 120
 121
 122         #Now impacts should really be .. impacts :)
 123         for s in all_servers:
 124             self.assert_(s.is_impact == True)
 125             self.assert_(s.state == 'UNREACHABLE')
 126             #And check the services are impacted too
 127             for svc in s.services:
 128                 print "Service state", svc.state
 129                 self.assert_(svc.state == 'UNKNOWN')
 130                 self.assert_(svc.get_dbg_name() in host_router_0_brok.data['impacts']['services'])
 131                 self.assert_(svc.get_dbg_name() in host_router_1_brok.data['impacts']['services'])
 132                 brk_svc = svc.get_update_status_brok()
 133                 self.assert_(brk_svc.data['source_problems']['hosts'] == ['test_router_0', 'test_router_1'])
 134             for h in all_routers:
 135                 self.assert_(h in s.source_problems)
 136                 brk_hst = s.get_update_status_brok()
 137                 self.assert_(h.get_dbg_name() in brk_hst.data['source_problems']['hosts'])
 138
 139
 140         #--------------------------------------------------------------
 141         # One router get UP now
 142         #--------------------------------------------------------------
 143         print "- 1 X UP for a router ------------------------------"
 144         #Ok here the problem/impact propagation is Checked. Now what
 145         #if one router get back? :)
 146         self.scheduler_loop(1, [[host_router_0, 0, 'UP']], do_sleep=False)
 147
 148         #should be UP/HARD now
 149         self.assert_(host_router_0.state == 'UP')
 150         self.assert_(host_router_0.state_type == 'HARD')
 151
 152         #And should not be a problem any more!
 153         self.assert_(host_router_0.is_problem == False)
 154         self.assert_(host_router_0.impacts == [])
 155
 156         #And check if it's no more in sources problems of others servers
 157         for s in all_servers:
 158             #Still impacted by the other server
 159             self.assert_(s.is_impact == True)
 160             self.assert_(s.source_problems == [host_router_1])
 161
 162
 163         #--------------------------------------------------------------
 164         # The other router get UP :)
 165         #--------------------------------------------------------------
 166         print "- 1 X UP for the last router ------------------------------"
 167         #What is the last router get back? :)
 168         self.scheduler_loop(1, [[host_router_1, 0, 'UP']], do_sleep=False)
 169
 170         #should be UP/HARD now
 171         self.assert_(host_router_1.state == 'UP')
 172         self.assert_(host_router_1.state_type == 'HARD')
 173
 174         #And should not be a problem any more!
 175         self.assert_(host_router_1.is_problem == False)
 176         self.assert_(host_router_1.impacts == [])
 177
 178         #And check if it's no more in sources problems of others servers
 179         for s in all_servers:
 180             #Still impacted by the other server
 181             self.assert_(s.is_impact == False)
 182             self.assert_(s.state == 'UP')
 183             self.assert_(s.source_problems == [])
 184
 185         # And our "criticity" should have failed back to our
 186         # conf value, so 3
 187         self.assert_(host_router_0.criticity == 3)
 188         self.assert_(host_router_1.criticity == 3)
 189         #It's done :)
 190
 191
 192 if __name__ == '__main__':
 193     unittest.main()