test/test_contactdowntimes.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
  27 class TestContactDowntime(ShinkenTest):
  28
  29     def test_contact_downtime(self):
  30         self.print_header()
  31         # schedule a 2-minute downtime
  32         # downtime must be active
  33         # consume a good result, sleep for a minute
  34         # downtime must be active
  35         # consume a bad result
  36         # downtime must be active
  37         # no notification must be found in broks
  38         duration = 600
  39         now = time.time()
  40         # downtime valid for the next 2 minutes
  41         test_contact = self.sched.contacts.find_by_name('test_contact')
  42         cmd = "[%lu] SCHEDULE_CONTACT_DOWNTIME;test_contact;%d;%d;lausser;blablub" % (now, now, now + duration)
  43         self.sched.run_external_command(cmd)
  44
  45         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
  46         svc.checks_in_progress = []
  47         svc.act_depend_of = [] # no hostchecks on critical checkresults
  48
  49         # Change the notif interval, so we can notify as soon as we want
  50         svc.notification_interval = 0.001
  51
  52         host = self.sched.hosts.find_by_name("test_host_0")
  53         host.checks_in_progress = []
  54         host.act_depend_of = [] # ignore the router
  55
  56         #time.sleep(20)
  57         # We loop, the downtime wil be check and activate
  58         self.scheduler_loop(1, [[svc, 0, 'OK'], [host, 0, 'UP']])
  59
  60         self.assert_(self.any_log_match('CONTACT DOWNTIME ALERT.*;STARTED'))
  61         self.show_and_clear_logs()
  62
  63         print "downtime was scheduled. check its activity and the comment"
  64         self.assert_(len(self.sched.contact_downtimes) == 1)
  65         self.assert_(len(test_contact.downtimes) == 1)
  66         self.assert_(test_contact.downtimes[0] in self.sched.contact_downtimes.values())
  67
  68         self.assert_(test_contact.downtimes[0].is_in_effect)
  69         self.assert_(not test_contact.downtimes[0].can_be_deleted)
  70
  71         # Ok, we define the downtime like we should, now look at if it does the job : do not
  72         # raise notif during a downtime for this contact
  73         self.scheduler_loop(3, [[svc, 2, 'CRITICAL']])
  74
  75         # We should NOT see any service notification
  76         self.assert_(not self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
  77         self.show_and_clear_logs()
  78
  79         # Now we short the downtime a lot so it will be stop at now + 1 sec.
  80         test_contact.downtimes[0].end_time = time.time() + 1
  81
  82         time.sleep(2)
  83
  84         # We invalidate it with a scheduler loop
  85         self.scheduler_loop(1, [])
  86
  87         # So we should be out now, with a log
  88         self.assert_(self.any_log_match('CONTACT DOWNTIME ALERT.*;STOPPED'))
  89         self.show_and_clear_logs()
  90
  91         print "Downtime was ended. Check it is really stopped"
  92         self.assert_(len(self.sched.contact_downtimes) == 0)
  93         self.assert_(len(test_contact.downtimes) == 0)
  94
  95         # Now we want this contact to be really notify!
  96         # Ok, we define the downtime like we should, now look at if it does the job : do not
  97         # raise notif during a downtime for this contact
  98         self.scheduler_loop(3, [[svc, 2, 'CRITICAL']])
  99         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 100         self.show_and_clear_logs()
 101
 102
 103
 104     def test_contact_downtime_and_cancel(self):
 105         self.print_header()
 106         # schedule a 2-minute downtime
 107         # downtime must be active
 108         # consume a good result, sleep for a minute
 109         # downtime must be active
 110         # consume a bad result
 111         # downtime must be active
 112         # no notification must be found in broks
 113         duration = 600
 114         now = time.time()
 115         # downtime valid for the next 2 minutes
 116         test_contact = self.sched.contacts.find_by_name('test_contact')
 117         cmd = "[%lu] SCHEDULE_CONTACT_DOWNTIME;test_contact;%d;%d;lausser;blablub" % (now, now, now + duration)
 118         self.sched.run_external_command(cmd)
 119
 120         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 121         svc.checks_in_progress = []
 122         svc.act_depend_of = [] # no hostchecks on critical checkresults
 123
 124         # Change the notif interval, so we can notify as soon as we want
 125         svc.notification_interval = 0.001
 126
 127         host = self.sched.hosts.find_by_name("test_host_0")
 128         host.checks_in_progress = []
 129         host.act_depend_of = [] # ignore the router
 130
 131         #time.sleep(20)
 132         # We loop, the downtime wil be check and activate
 133         self.scheduler_loop(1, [[svc, 0, 'OK'], [host, 0, 'UP']])
 134
 135         self.assert_(self.any_log_match('CONTACT DOWNTIME ALERT.*;STARTED'))
 136         self.show_and_clear_logs()
 137
 138         print "downtime was scheduled. check its activity and the comment"
 139         self.assert_(len(self.sched.contact_downtimes) == 1)
 140         self.assert_(len(test_contact.downtimes) == 1)
 141         self.assert_(test_contact.downtimes[0] in self.sched.contact_downtimes.values())
 142
 143         self.assert_(test_contact.downtimes[0].is_in_effect)
 144         self.assert_(not test_contact.downtimes[0].can_be_deleted)
 145
 146         time.sleep(1)
 147         # Ok, we define the downtime like we should, now look at if it does the job : do not
 148         # raise notif during a downtime for this contact
 149         self.scheduler_loop(3, [[svc, 2, 'CRITICAL']])
 150
 151         # We should NOT see any service notification
 152         self.assert_(not self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 153         self.show_and_clear_logs()
 154
 155         downtime_id = test_contact.downtimes[0].id
 156         # OK, Now we cancel this downtime, we do not need it anymore
 157         cmd = "[%lu] DEL_CONTACT_DOWNTIME;%d" % (now, downtime_id)
 158         self.sched.run_external_command(cmd)
 159
 160         # We check if the downtime is tag as to remove
 161         self.assert_(test_contact.downtimes[0].can_be_deleted)
 162
 163         # We really delete it
 164         self.scheduler_loop(1, [])
 165
 166         # So we should be out now, with a log
 167         self.assert_(self.any_log_match('CONTACT DOWNTIME ALERT.*;CANCELLED'))
 168         self.show_and_clear_logs()
 169
 170         print "Downtime was cancelled"
 171         self.assert_(len(self.sched.contact_downtimes) == 0)
 172         self.assert_(len(test_contact.downtimes) == 0)
 173
 174         time.sleep(1)
 175         # Now we want this contact to be really notify!
 176         # Ok, we define the downtime like we should, now look at if it does the job : do not
 177         # raise notif during a downtime for this contact
 178         self.scheduler_loop(3, [[svc, 2, 'CRITICAL']])
 179         self.assert_(self.any_log_match('SERVICE NOTIFICATION.*;CRITICAL'))
 180         self.show_and_clear_logs()
 181
 182
 183
 184 if __name__ == '__main__':
 185     unittest.main()