test/test_unknown_do_not_change.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 #It's ugly I know....
  26 from shinken_test import *
  27
  28
  29 class TestUnknownNotChangeState(ShinkenTest):
  30     #Uncomment this is you want to use a specific configuration
  31     #for your test
  32     #def setUp(self):
  33     #    self.setup_with_file('etc/nagios_1r_1h_1s.cfg')
  34
  35
  36     # We got problem with unknown results on bad connexions
  37     # for critical services and host : if it was in a notification pass
  38     # then the notification is restarted, but it's just a missing data,
  39     # not a reason to warn about it
  40     def test_unknown_do_not_change_state(self):
  41         host = self.sched.hosts.find_by_name("test_host_0")
  42         host.checks_in_progress = []
  43         host.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 the router
  47         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
  48         svc.checks_in_progress = []
  49         svc.act_depend_of = [] # no hostchecks on critical checkresults
  50
  51         print "GO OK"*10
  52         self.scheduler_loop(2, [[host, 0, 'UP | value1=1 value2=2'], [router, 0, 'UP | rtt=10'], [svc, 0, 'OK | value1=0 value2=0']])
  53         self.assert_(svc.state == 'OK')
  54         self.assert_(svc.state_type == 'HARD')
  55
  56         print "GO CRITICAL SOFT"*10
  57         # Ok we are UP, now we seach to go in trouble
  58         self.scheduler_loop(1, [[svc, 2, 'PROBLEM | value1=1 value2=2']])
  59         # CRITICAL/SOFT
  60         self.assert_(svc.state == 'CRITICAL')
  61         self.assert_(svc.state_type == 'SOFT')
  62         # And again and again :)
  63         print "GO CRITICAL HARD"*10
  64         self.scheduler_loop(2, [[svc, 2, 'PROBLEM | value1=1 value2=2']])
  65         # CRITICAL/HARD
  66         self.assert_(svc.state == 'CRITICAL')
  67         self.assert_(svc.state_type == 'HARD')
  68
  69         # Should have a notification about it
  70         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
  71         self.show_and_clear_logs()
  72
  73         print "GO UNKNOWN HARD"*10
  74         # Then we make it as a unknown state
  75         self.scheduler_loop(1, [[svc, 3, 'Unknown | value1=1 value2=2']])
  76         # And we DO NOT WANT A NOTIF HERE
  77         self.assert_(not self.any_log_match('SERVICE NOTIFICATION.*;UNKNOWN'))
  78         self.show_and_clear_logs()
  79
  80         print "Return CRITICAL HARD"*10
  81         # Then we came back as CRITICAL
  82         self.scheduler_loop(1, [[svc, 2, 'CRITICAL | value1=1 value2=2']])
  83         print svc.state, svc.state_type
  84         self.assert_(not self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
  85         self.show_and_clear_logs()
  86
  87         print "Still CRITICAL HARD"*10
  88         # Then we came back as CRITICAL
  89         self.scheduler_loop(1, [[svc, 2, 'CRITICAL | value1=1 value2=2']])
  90         print svc.state, svc.state_type
  91         self.assert_(not self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
  92         self.show_and_clear_logs()
  93
  94         # We check if we can still have new notifications of course
  95         # And we speedup the notification
  96         for n in svc.notifications_in_progress.values():
  97             n.t_to_go = time.time()
  98         self.scheduler_loop(1, [[svc, 2, 'CRITICAL | value1=1 value2=2']])
  99         print svc.state, svc.state_type
 100         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 101         self.show_and_clear_logs()
 102
 103
 104     # We got problem with unknown results on bad connexions
 105     # for critical services and host : if it was in a notification pass
 106     # then the notification is restarted, but it's just a missing data,
 107     # not a reason to warn about it
 108     def test_unknown_do_not_change_state_with_different_exit_status_phase(self):
 109         host = self.sched.hosts.find_by_name("test_host_0")
 110         host.checks_in_progress = []
 111         host.act_depend_of = [] # ignore the router
 112         router = self.sched.hosts.find_by_name("test_router_0")
 113         router.checks_in_progress = []
 114         router.act_depend_of = [] # ignore the router
 115         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 116         svc.checks_in_progress = []
 117         svc.act_depend_of = [] # no hostchecks on critical checkresults
 118
 119         print "GO OK"*10
 120         self.scheduler_loop(2, [[host, 0, 'UP | value1=1 value2=2'], [router, 0, 'UP | rtt=10'], [svc, 0, 'OK | value1=0 value2=0']])
 121         self.assert_(svc.state == 'OK')
 122         self.assert_(svc.state_type == 'HARD')
 123
 124         print "GO CRITICAL SOFT"*10
 125         # Ok we are UP, now we seach to go in trouble
 126         self.scheduler_loop(1, [[svc, 2, 'PROBLEM | value1=1 value2=2']])
 127         # CRITICAL/SOFT
 128         self.assert_(svc.state == 'CRITICAL')
 129         self.assert_(svc.state_type == 'SOFT')
 130         # And again and again :)
 131         print "GO CRITICAL HARD"*10
 132         self.scheduler_loop(2, [[svc, 2, 'PROBLEM | value1=1 value2=2']])
 133         # CRITICAL/HARD
 134         self.assert_(svc.state == 'CRITICAL')
 135         self.assert_(svc.state_type == 'HARD')
 136
 137         # Should have a notification about it
 138         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 139         self.show_and_clear_logs()
 140
 141         print "GO UNKNOWN HARD"*10
 142         # Then we make it as a unknown state
 143         self.scheduler_loop(1, [[svc, 3, 'Unknown | value1=1 value2=2']])
 144         # And we DO NOT WANT A NOTIF HERE
 145         self.assert_(not self.any_log_match('SERVICE NOTIFICATION.*;UNKNOWN'))
 146         self.show_and_clear_logs()
 147
 148         print "Return CRITICAL HARD"*10
 149         # Then we came back as WARNING here, so a different than we came in the phase!
 150         self.scheduler_loop(1, [[svc, 1, 'WARNING | value1=1 value2=2']])
 151         print svc.state, svc.state_type
 152         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;WARNING'))
 153         self.show_and_clear_logs()
 154
 155         # We check if we can still have new notifications of course
 156         # And we speedup the notification
 157         for n in svc.notifications_in_progress.values():
 158             n.t_to_go = time.time()
 159         self.scheduler_loop(1, [[svc, 1, 'WARNING | value1=1 value2=2']])
 160         print svc.state, svc.state_type
 161         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;WARNING'))
 162         self.show_and_clear_logs()
 163
 164         # And what if we came back as critical so? :)
 165         self.scheduler_loop(1, [[svc, 2, 'CRITICAL | value1=1 value2=2']])
 166         print svc.state, svc.state_type
 167         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 168         self.show_and_clear_logs()
 169
 170
 171
 172
 173     # But we want to still raise notif as unknown if we first met this state
 174     def test_unknown_still_raise_notif(self):
 175         host = self.sched.hosts.find_by_name("test_host_0")
 176         host.checks_in_progress = []
 177         host.act_depend_of = [] # ignore the router
 178         router = self.sched.hosts.find_by_name("test_router_0")
 179         router.checks_in_progress = []
 180         router.act_depend_of = [] # ignore the router
 181         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 182         svc.checks_in_progress = []
 183         svc.act_depend_of = [] # no hostchecks on critical checkresults
 184
 185         self.scheduler_loop(2, [[host, 0, 'UP | value1=1 value2=2'], [router, 0, 'UP | rtt=10'], [svc, 0, 'OK | value1=0 value2=0']])
 186         self.assert_(svc.state == 'OK')
 187         self.assert_(svc.state_type == 'HARD')
 188
 189         # Ok we are UP, now we seach to go in trouble
 190         self.scheduler_loop(1, [[svc, 3, 'PROBLEM | value1=1 value2=2']])
 191         # UNKOWN/SOFT
 192         self.assert_(svc.state == 'UNKNOWN')
 193         self.assert_(svc.state_type == 'SOFT')
 194         # And again and again :)
 195         self.scheduler_loop(2, [[svc, 3, 'PROBLEM | value1=1 value2=2']])
 196         # UNKNOWN/HARD
 197         self.assert_(svc.state == 'UNKNOWN')
 198         self.assert_(svc.state_type == 'HARD')
 199
 200         # Should have a notification about it !
 201         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;UNKNOWN'))
 202         self.show_and_clear_logs()
 203
 204         # Then we make it as a critical state
 205         # and we want a notif too
 206         self.scheduler_loop(1, [[svc, 2, 'critical | value1=1 value2=2']])
 207         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 208         self.show_and_clear_logs()
 209
 210
 211
 212
 213     # We got problem with unknown results on bad connexions
 214     # for critical services and host : if it was in a notification pass
 215     # then the notification is restarted, but it's just a missing data,
 216     # not a reason to warn about it
 217     def test_unreach_do_not_change_state(self):
 218         host = self.sched.hosts.find_by_name("test_host_0")
 219         host.checks_in_progress = []
 220         router = self.sched.hosts.find_by_name("test_router_0")
 221         router.checks_in_progress = []
 222         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 223         svc.checks_in_progress = []
 224         svc.act_depend_of = [] # no hostchecks on critical checkresults
 225
 226
 227         print "GO OK"*10
 228         self.scheduler_loop(2, [[host, 0, 'UP | value1=1 value2=2'], [router, 0, 'UP | rtt=10'], [svc, 0, 'OK | value1=0 value2=0']])
 229         self.assert_(svc.state == 'OK')
 230         self.assert_(svc.state_type == 'HARD')
 231
 232         print "GO DOWN SOFT"*10
 233         # Ok we are UP, now we seach to go in trouble
 234         self.scheduler_loop(1, [[host, 2, 'PROBLEM | value1=1 value2=2']])
 235         # CRITICAL/SOFT
 236         self.assert_(host.state == 'DOWN')
 237         self.assert_(host.state_type == 'SOFT')
 238         # And again and again :)
 239         print "GO CRITICAL HARD"*10
 240         self.scheduler_loop(2, [[host, 2, 'PROBLEM | value1=1 value2=2']])
 241         # CRITICAL/HARD
 242         self.assert_(host.state == 'DOWN')
 243         self.assert_(host.state_type == 'HARD')
 244
 245         # Should have a notification about it
 246         self.assert_(self.any_log_match('HOST NOTIFICATION.*;DOWN'))
 247         self.show_and_clear_logs()
 248
 249         print "GO UNREACH HARD"*10
 250         # Then we make it as a unknown state
 251         self.scheduler_loop(3, [[router, 2, 'Bad router | value1=1 value2=2']])
 252         # so we warn about the router, not the host
 253         self.assert_(self.any_log_match('HOST NOTIFICATION.*;DOWN'))
 254         self.show_and_clear_logs()
 255
 256         print "BIBI"*100
 257         for n in host.notifications_in_progress.values():
 258             print n.__dict__
 259
 260         # the we go in UNREACH
 261         self.scheduler_loop(1, [[host, 2, 'CRITICAL | value1=1 value2=2']])
 262         print host.state, host.state_type
 263         self.show_and_clear_logs()
 264         self.assert_(host.state == 'UNREACHABLE')
 265         self.assert_(host.state_type == 'HARD')
 266
 267         # The the router came back :)
 268         print "Router is back from Hell"*10
 269         self.scheduler_loop(1, [[router, 0, 'Ok, I am back guys | value1=1 value2=2']])
 270         self.assert_(self.any_log_match('HOST NOTIFICATION.*;UP'))
 271         self.show_and_clear_logs()
 272
 273         # But how the host will say now?
 274         self.scheduler_loop(1, [[host, 2, 'CRITICAL | value1=1 value2=2']])
 275         print host.state, host.state_type
 276         # And here we DO NOT WANT new notification
 277         # If you folow, it THE important point of this test!
 278         self.assert_(not self.any_log_match('HOST NOTIFICATION.*;DOWN'))
 279         self.show_and_clear_logs()
 280
 281         print "Now go in the future, I want a notification"
 282         # Check if we still got the next notification for this of course
 283
 284         # Hack so the notification will raise now if it can
 285         for n in host.notifications_in_progress.values():
 286             n.t_to_go = time.time()
 287         self.scheduler_loop(1, [[host, 2, 'CRITICAL | value1=1 value2=2']])
 288         print host.state, host.state_type
 289         # And here we DO NOT WANT new notification
 290         self.assert_(self.any_log_match('HOST NOTIFICATION.*;DOWN'))
 291         self.show_and_clear_logs()
 292
 293
 294
 295
 296
 297 if __name__ == '__main__':
 298     unittest.main()
 299