<commit_before>#!/usr/bin/env python
# Copyright (C) 2009-2014:
#    <NAME>, <EMAIL>
#    <NAME>, <EMAIL>
#
# This file is part of Shinken.
#
# Shinken is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Shinken is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Shinken.  If not, see <http://www.gnu.org/licenses/>.


#
# This file is used to test host- and service-downtimes.
#

import time

from shinken_test import unittest, ShinkenTest


class TestNotif(ShinkenTest):

    def test_continuous_notifications(self):
        self.print_header()
        # retry_interval 2
        # critical notification
        # run loop -> another notification
        now = time.time()
        host = self.sched.hosts.find_by_name("test_host_0")
        host.checks_in_progress = []
        host.act_depend_of = []  # ignore the router
        svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")

        # To make tests quicker we make notifications send very quickly
        svc.notification_interval = 0.001

        svc.checks_in_progress = []
        svc.act_depend_of = []  # no hostchecks on critical checkresults
        #--------------------------------------------------------------
        # initialize host/service state
        #--------------------------------------------------------------
        self.scheduler_loop(1, [[host, 0, 'UP']], do_sleep=True, sleep_time=0.1)
        print "- 1 x OK -------------------------------------"
        self.scheduler_loop(1, [[svc, 0, 'OK']], do_sleep=True, sleep_time=0.1)

        self.assertEqual(0, svc.current_notification_number)
        #--------------------------------------------------------------
        # service reaches soft;1
        # there must not be any notification
        #--------------------------------------------------------------
        print "- 1 x BAD get soft -------------------------------------"
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        # check_notification: not (soft)
        print "---current_notification_number", svc.current_notification_number
        #--------------------------------------------------------------
        # service reaches hard;2
        # a notification must have been created
        # notification number must be 1
        #--------------------------------------------------------------
        print "- 1 x BAD get hard -------------------------------------"
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        #self.show_and_clear_actions()
        self.show_actions()
        print svc.notifications_in_progress
        for n in svc.notifications_in_progress.values():
            print n
        # check_notification: yes (hard)
        print "---current_notification_number", svc.current_notification_number
        # notification_number is already sent. the next one has been scheduled
        # and is waiting for notification_interval to pass. so the current
        # number is 2
        self.assertEqual(1, svc.current_notification_number)
        print "---------------------------------1st round with a hard"
        print "find a way to get the number of the last reaction"
        cnn = svc.current_notification_number
        print "- 5 x BAD repeat -------------------------------------"
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "cnn and cur", cnn, svc.current_notification_number
        self.assertGreater(svc.current_notification_number, cnn)
        cnn = svc.current_notification_number
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "svc.current_notification_number, cnn", svc.current_notification_number, cnn
        self.assertGreater(svc.current_notification_number, cnn)
        #--------------------------------------------------------------
        # 2 cycles = 2 minutes = 2 new notifications
        #--------------------------------------------------------------
        cnn = svc.current_notification_number
        self.scheduler_loop(2, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "svc.current_notification_number, cnn", svc.current_notification_number, cnn
        self.assertGreater(svc.current_notification_number, cnn)
        #--------------------------------------------------------------
        # 2 cycles = 2 minutes = 2 new notifications (theoretically)
        # BUT: test_contact filters notifications
        # we do not raise current_notification_number if no mail was sent
        #--------------------------------------------------------------
        now = time.time()
        cmd = "[%lu] DISABLE_CONTACT_SVC_NOTIFICATIONS;test_contact" % now
        self.sched.run_external_command(cmd)
        cnn = svc.current_notification_number
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        self.assertEqual(cnn, svc.current_notification_number)
        #--------------------------------------------------------------
        # again a normal cycle
        # test_contact receives his mail
        #--------------------------------------------------------------
        now = time.time()
        cmd = "[%lu] ENABLE_CONTACT_SVC_NOTIFICATIONS;test_contact" % now
        self.sched.run_external_command(cmd)
        #cnn = svc.current_notification_number
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "svc.current_notification_number, cnn", svc.current_notification_number, cnn
        self.assertEqual(cnn + 1, svc.current_notification_number)
        #--------------------------------------------------------------
        # now recover. there must be no scheduled/inpoller notification
        #--------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 0, 'GOOD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_and_clear_actions()
        self.assertEqual(0, svc.current_notification_number)

    def test_continuous_notifications_delayed(self):
        self.print_header()
        # retry_interval 2
        # critical notification
        # run loop -> another notification
        now = time.time()
        host = self.sched.hosts.find_by_name("test_host_0")
        host.checks_in_progress = []
        host.act_depend_of = []  # ignore the router
        svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")

        # To make tests quicker we make notifications send very quickly
        svc.notification_interval = 0.001  # and send imediatly then

        svc.first_notification_delay = 0.1  # set 6s for first notif delay
        svc.checks_in_progress = []
        svc.act_depend_of = []  # no hostchecks on critical checkresults
        self.scheduler_loop(1, [[host, 0, 'UP']], do_sleep=True, sleep_time=1)
        #-----------------------------------------------------------------
        # initialize with a good check. there must be no pending notification
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 0, 'OK']], do_sleep=True, sleep_time=1)
        self.show_and_clear_logs()
        self.show_and_clear_actions()
        self.assertEqual(0, svc.current_notification_number)
        #-----------------------------------------------------------------
        # check fails and enters soft state.
        # there must be no notification, only the event handler
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 1, 'BAD']], do_sleep=True, sleep_time=1)
        self.assertEqual(1, self.count_actions())
        print time.time()
        print  svc.last_time_warning, svc.last_time_critical, svc.last_time_unknown, svc.last_time_ok
        last_time_not_ok = svc.last_time_non_ok_or_up()
        deadline = svc.last_time_non_ok_or_up() + svc.first_notification_delay * svc.__class__.interval_length
        #-----------------------------------------------------------------
        # check fails again and enters hard state.
        # now there is a (scheduled for later) notification and an event handler
        # current_notification_number is still 0, until notifications
        # have actually been sent
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        self.assertEqual(0, svc.current_notification_number)
        #-----------------------------------------------------------------
        # repeat bad checks during the delay time
        # there is 1 action which is the scheduled notification
        #-----------------------------------------------------------------
        loop = 0
        while deadline > time.time():
            loop += 1
            self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
            self.show_and_clear_logs()
            self.show_actions()
            print deadline - time.time()
            ###self.assertEqual(1, self.count_actions())
        #-----------------------------------------------------------------
        # now the delay period is over and the notification can be sent
        # with the next bad check
        # there is 1 action, the notification (
        # 1 notification was sent, so current_notification_number is 1
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=1)
        print "Counted actions", self.count_actions()
        self.assertEqual(2, self.count_actions())
        # 1 master, 1 child
        self.assertEqual(1, svc.current_notification_number)
        self.show_actions()
        self.assertEqual(1, len(svc.notifications_in_progress))  # master is zombieand removed_from_in_progress
        self.show_logs()
        self.assert_log_match(1, 'SERVICE NOTIFICATION.*;CRITICAL;')
        self.show_and_clear_logs()<commit_msg>Fix: fixed unstable assert in test_continuous_notifications_delayed<commit_after>#!/usr/bin/env python
# Copyright (C) 2009-2014:
#    <NAME>, <EMAIL>
#    <NAME>, <EMAIL>
#
# This file is part of Shinken.
#
# Shinken is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Shinken is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Shinken.  If not, see <http://www.gnu.org/licenses/>.


#
# This file is used to test host- and service-downtimes.
#

import time

from shinken_test import unittest, ShinkenTest
from shinken_test import time_hacker


class TestNotif(ShinkenTest):

    def test_continuous_notifications(self):
        self.print_header()
        # retry_interval 2
        # critical notification
        # run loop -> another notification
        now = time.time()
        host = self.sched.hosts.find_by_name("test_host_0")
        host.checks_in_progress = []
        host.act_depend_of = []  # ignore the router
        svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")

        # To make tests quicker we make notifications send very quickly
        svc.notification_interval = 0.001

        svc.checks_in_progress = []
        svc.act_depend_of = []  # no hostchecks on critical checkresults
        #--------------------------------------------------------------
        # initialize host/service state
        #--------------------------------------------------------------
        self.scheduler_loop(1, [[host, 0, 'UP']], do_sleep=True, sleep_time=0.1)
        print "- 1 x OK -------------------------------------"
        self.scheduler_loop(1, [[svc, 0, 'OK']], do_sleep=True, sleep_time=0.1)

        self.assertEqual(0, svc.current_notification_number)
        #--------------------------------------------------------------
        # service reaches soft;1
        # there must not be any notification
        #--------------------------------------------------------------
        print "- 1 x BAD get soft -------------------------------------"
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        # check_notification: not (soft)
        print "---current_notification_number", svc.current_notification_number
        #--------------------------------------------------------------
        # service reaches hard;2
        # a notification must have been created
        # notification number must be 1
        #--------------------------------------------------------------
        print "- 1 x BAD get hard -------------------------------------"
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        #self.show_and_clear_actions()
        self.show_actions()
        print svc.notifications_in_progress
        for n in svc.notifications_in_progress.values():
            print n
        # check_notification: yes (hard)
        print "---current_notification_number", svc.current_notification_number
        # notification_number is already sent. the next one has been scheduled
        # and is waiting for notification_interval to pass. so the current
        # number is 2
        self.assertEqual(1, svc.current_notification_number)
        print "---------------------------------1st round with a hard"
        print "find a way to get the number of the last reaction"
        cnn = svc.current_notification_number
        print "- 5 x BAD repeat -------------------------------------"
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "cnn and cur", cnn, svc.current_notification_number
        self.assertGreater(svc.current_notification_number, cnn)
        cnn = svc.current_notification_number
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "svc.current_notification_number, cnn", svc.current_notification_number, cnn
        self.assertGreater(svc.current_notification_number, cnn)
        #--------------------------------------------------------------
        # 2 cycles = 2 minutes = 2 new notifications
        #--------------------------------------------------------------
        cnn = svc.current_notification_number
        self.scheduler_loop(2, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "svc.current_notification_number, cnn", svc.current_notification_number, cnn
        self.assertGreater(svc.current_notification_number, cnn)
        #--------------------------------------------------------------
        # 2 cycles = 2 minutes = 2 new notifications (theoretically)
        # BUT: test_contact filters notifications
        # we do not raise current_notification_number if no mail was sent
        #--------------------------------------------------------------
        now = time.time()
        cmd = "[%lu] DISABLE_CONTACT_SVC_NOTIFICATIONS;test_contact" % now
        self.sched.run_external_command(cmd)
        cnn = svc.current_notification_number
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        self.assertEqual(cnn, svc.current_notification_number)
        #--------------------------------------------------------------
        # again a normal cycle
        # test_contact receives his mail
        #--------------------------------------------------------------
        now = time.time()
        cmd = "[%lu] ENABLE_CONTACT_SVC_NOTIFICATIONS;test_contact" % now
        self.sched.run_external_command(cmd)
        #cnn = svc.current_notification_number
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        print "svc.current_notification_number, cnn", svc.current_notification_number, cnn
        self.assertEqual(cnn + 1, svc.current_notification_number)
        #--------------------------------------------------------------
        # now recover. there must be no scheduled/inpoller notification
        #--------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 0, 'GOOD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_and_clear_actions()
        self.assertEqual(0, svc.current_notification_number)

    def test_continuous_notifications_delayed(self):
        self.print_header()
        # retry_interval 2
        # critical notification
        # run loop -> another notification
        now = time.time()
        host = self.sched.hosts.find_by_name("test_host_0")
        host.checks_in_progress = []
        host.act_depend_of = []  # ignore the router
        svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")

        # To make tests quicker we make notifications send very quickly
        svc.notification_interval = 0.001  # and send imediatly then

        svc.first_notification_delay = 0.1  # set 6s for first notif delay
        svc.checks_in_progress = []
        svc.act_depend_of = []  # no hostchecks on critical checkresults
        self.scheduler_loop(1, [[host, 0, 'UP']], do_sleep=True, sleep_time=1)
        #-----------------------------------------------------------------
        # initialize with a good check. there must be no pending notification
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 0, 'OK']], do_sleep=True, sleep_time=1)
        self.show_and_clear_logs()
        self.show_and_clear_actions()
        self.assertEqual(0, svc.current_notification_number)
        #-----------------------------------------------------------------
        # check fails and enters soft state.
        # there must be no notification, only the event handler
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 1, 'BAD']], do_sleep=True, sleep_time=1)
        self.assertEqual(1, self.count_actions())
        now = time.time()
        print svc.last_time_warning, svc.last_time_critical, svc.last_time_unknown, svc.last_time_ok
        last_time_not_ok = svc.last_time_non_ok_or_up()
        deadline = svc.last_time_non_ok_or_up() + svc.first_notification_delay * svc.__class__.interval_length
        print("deadline is in %s secs" % (deadline - now))
        #-----------------------------------------------------------------
        # check fails again and enters hard state.
        # now there is a (scheduled for later) notification and an event handler
        # current_notification_number is still 0, until notifications
        # have actually been sent
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=0.1)
        self.show_and_clear_logs()
        self.show_actions()
        self.assertEqual(0, svc.current_notification_number)
        # sleep up to deadline:
        time_hacker.time_warp(deadline - now)
        # even if time_hacker is used here, we still call time.sleep()
        # to show that we must wait the necessary delay time:
        time.sleep(deadline - now)
        #-----------------------------------------------------------------
        # now the delay period is over and the notification can be sent
        # with the next bad check
        # there is 1 action, the notification (
        # 1 notification was sent, so current_notification_number is 1
        #-----------------------------------------------------------------
        self.scheduler_loop(1, [[svc, 2, 'BAD']], do_sleep=True, sleep_time=1)
        print "Counted actions", self.count_actions()
        self.assertEqual(2, self.count_actions())
        # 1 master, 1 child
        self.assertEqual(1, svc.current_notification_number)
        self.show_actions()
        self.assertEqual(1, len(svc.notifications_in_progress))  # master is zombieand removed_from_in_progress
        self.show_logs()
        self.assert_log_match(1, 'SERVICE NOTIFICATION.*;CRITICAL;')
        self.show_and_clear_logs()
{'fraction_non_alphanumeric': 0.22887518409425625, 'fraction_numerical': 0.011045655375552283, 'mean_word_length': 1.6392566500668044, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2821480', 'n_tokens_mistral': 5793, 'n_tokens_neox': 5626, 'n_words': 1441}