test/test_business_correlator.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 import re
  26 #It's ugly I know....
  27 from shinken_test import *
  28
  29
  30 class TestConfig(ShinkenTest):
  31     # Uncomment this is you want to use a specific configuration
  32     # for your test
  33     def setUp(self):
  34         self.setup_with_file('etc/nagios_business_correlator.cfg')
  35
  36
  37     # We will try a simple bd1 OR db2
  38     def test_simple_or_business_correlator(self):
  39         #
  40         # Config is not correct because of a wrong relative path
  41         # in the main config file
  42         #
  43         print "Get the hosts and services"
  44         now = time.time()
  45         host = self.sched.hosts.find_by_name("test_host_0")
  46         host.checks_in_progress = []
  47         host.act_depend_of = [] # ignore the router
  48         router = self.sched.hosts.find_by_name("test_router_0")
  49         router.checks_in_progress = []
  50         router.act_depend_of = [] # ignore the router
  51         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
  52         svc.checks_in_progress = []
  53         svc.act_depend_of = [] # no hostchecks on critical checkresults
  54
  55         svc_bd1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db1")
  56         self.assert_(svc_bd1.got_business_rule == False)
  57         self.assert_(svc_bd1.business_rule is None)
  58         svc_bd2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db2")
  59         self.assert_(svc_bd2.got_business_rule == False)
  60         self.assert_(svc_bd2.business_rule is None)
  61         svc_cor = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "Simple_Or")
  62         self.assert_(svc_cor.got_business_rule == True)
  63         self.assert_(svc_cor.business_rule is not None)
  64         bp_rule = svc_cor.business_rule
  65         self.assert_(bp_rule.operand == '|')
  66
  67         # We check for good parent/childs links
  68         # So svc_cor should be a son of svc_bd1 and svc_bd2
  69         # and bd1 and bd2 should be parents of svc_cor
  70         self.assert_(svc_cor in svc_bd1.child_dependencies)
  71         self.assert_(svc_cor in svc_bd2.child_dependencies)
  72         self.assert_(svc_bd1 in svc_cor.parent_dependencies)
  73         self.assert_(svc_bd2 in svc_cor.parent_dependencies)
  74
  75         sons = bp_rule.sons
  76         print "Sons,", sons
  77         #We(ve got 2 sons, 2 services nodes
  78         self.assert_(len(sons) == 2)
  79         self.assert_(sons[0].operand == 'service')
  80         self.assert_(sons[0].sons[0] == svc_bd1)
  81         self.assert_(sons[1].operand == 'service')
  82         self.assert_(sons[1].sons[0] == svc_bd2)
  83
  84         # Now state working on the states
  85         self.scheduler_loop(1, [[svc_bd2, 0, 'OK | value1=1 value2=2'], [svc_bd1, 0, 'OK | rtt=10']])
  86         self.assert_(svc_bd1.state == 'OK')
  87         self.assert_(svc_bd1.state_type == 'HARD')
  88         self.assert_(svc_bd2.state == 'OK')
  89         self.assert_(svc_bd2.state_type == 'HARD')
  90
  91         state = bp_rule.get_state()
  92         self.assert_(state == 0)
  93
  94         # Now we set the bd1 as soft/CRITICAL
  95         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
  96         self.assert_(svc_bd1.state == 'CRITICAL')
  97         self.assert_(svc_bd1.state_type == 'SOFT')
  98         self.assert_(svc_bd1.last_hard_state_id == 0)
  99
 100         # The business rule must still be 0
 101         state = bp_rule.get_state()
 102         self.assert_(state == 0)
 103
 104         # Now we get bd1 CRITICAL/HARD
 105         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 106         self.assert_(svc_bd1.state == 'CRITICAL')
 107         self.assert_(svc_bd1.state_type == 'HARD')
 108         self.assert_(svc_bd1.last_hard_state_id == 2)
 109
 110         # The rule must still be a 0 (or inside)
 111         state = bp_rule.get_state()
 112         self.assert_(state == 0)
 113
 114         # Now we also set bd2 as CRITICAL/HARD... byebye 0 :)
 115         self.scheduler_loop(2, [[svc_bd2, 2, 'CRITICAL | value1=1 value2=2']])
 116         self.assert_(svc_bd2.state == 'CRITICAL')
 117         self.assert_(svc_bd2.state_type == 'HARD')
 118         self.assert_(svc_bd2.last_hard_state_id == 2)
 119
 120         # And now the state of the rule must be 2
 121         state = bp_rule.get_state()
 122         self.assert_(state == 2)
 123
 124         # And If we set one WARNING?
 125         self.scheduler_loop(2, [[svc_bd2, 1, 'WARNING | value1=1 value2=2']])
 126         self.assert_(svc_bd2.state == 'WARNING')
 127         self.assert_(svc_bd2.state_type == 'HARD')
 128         self.assert_(svc_bd2.last_hard_state_id == 1)
 129
 130         # Must be WARNING (better no 0 value)
 131         state = bp_rule.get_state()
 132         self.assert_(state == 1)
 133
 134
 135
 136     # We will try a simple bd1 AND db2
 137     def test_simple_and_business_correlator(self):
 138         #
 139         # Config is not correct because of a wrong relative path
 140         # in the main config file
 141         #
 142         print "Get the hosts and services"
 143         now = time.time()
 144         host = self.sched.hosts.find_by_name("test_host_0")
 145         host.checks_in_progress = []
 146         host.act_depend_of = [] # ignore the router
 147         router = self.sched.hosts.find_by_name("test_router_0")
 148         router.checks_in_progress = []
 149         router.act_depend_of = [] # ignore the router
 150         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 151         svc.checks_in_progress = []
 152         svc.act_depend_of = [] # no hostchecks on critical checkresults
 153
 154         svc_bd1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db1")
 155         self.assert_(svc_bd1.got_business_rule == False)
 156         self.assert_(svc_bd1.business_rule is None)
 157         svc_bd2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db2")
 158         self.assert_(svc_bd2.got_business_rule == False)
 159         self.assert_(svc_bd2.business_rule is None)
 160         svc_cor = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "Simple_And")
 161         self.assert_(svc_cor.got_business_rule == True)
 162         self.assert_(svc_cor.business_rule is not None)
 163         bp_rule = svc_cor.business_rule
 164         self.assert_(bp_rule.operand == '&')
 165
 166         sons = bp_rule.sons
 167         print "Sons,", sons
 168         #We(ve got 2 sons, 2 services nodes
 169         self.assert_(len(sons) == 2)
 170         self.assert_(sons[0].operand == 'service')
 171         self.assert_(sons[0].sons[0] == svc_bd1)
 172         self.assert_(sons[1].operand == 'service')
 173         self.assert_(sons[1].sons[0] == svc_bd2)
 174
 175         # Now state working on the states
 176         self.scheduler_loop(1, [[svc_bd2, 0, 'OK | value1=1 value2=2'], [svc_bd1, 0, 'OK | rtt=10']])
 177         self.assert_(svc_bd1.state == 'OK')
 178         self.assert_(svc_bd1.state_type == 'HARD')
 179         self.assert_(svc_bd2.state == 'OK')
 180         self.assert_(svc_bd2.state_type == 'HARD')
 181
 182         state = bp_rule.get_state()
 183         self.assert_(state == 0)
 184
 185         # Now we set the bd1 as soft/CRITICAL
 186         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 187         self.assert_(svc_bd1.state == 'CRITICAL')
 188         self.assert_(svc_bd1.state_type == 'SOFT')
 189         self.assert_(svc_bd1.last_hard_state_id == 0)
 190
 191         # The business rule must still be 0
 192         # becase we want HARD states
 193         state = bp_rule.get_state()
 194         self.assert_(state == 0)
 195
 196         # Now we get bd1 CRITICAL/HARD
 197         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 198         self.assert_(svc_bd1.state == 'CRITICAL')
 199         self.assert_(svc_bd1.state_type == 'HARD')
 200         self.assert_(svc_bd1.last_hard_state_id == 2)
 201
 202         # The rule must go CRITICAL
 203         state = bp_rule.get_state()
 204         self.assert_(state == 2)
 205
 206         # Now we also set bd2 as WARNING/HARD...
 207         self.scheduler_loop(2, [[svc_bd2, 1, 'WARNING | value1=1 value2=2']])
 208         self.assert_(svc_bd2.state == 'WARNING')
 209         self.assert_(svc_bd2.state_type == 'HARD')
 210         self.assert_(svc_bd2.last_hard_state_id == 1)
 211
 212         # And now the state of the rule must be 2
 213         state = bp_rule.get_state()
 214         self.assert_(state == 2)
 215
 216         # And If we set one WARNING too?
 217         self.scheduler_loop(2, [[svc_bd1, 1, 'WARNING | value1=1 value2=2']])
 218         self.assert_(svc_bd1.state == 'WARNING')
 219         self.assert_(svc_bd1.state_type == 'HARD')
 220         self.assert_(svc_bd1.last_hard_state_id == 1)
 221
 222         # Must be WARNING (worse no 0 value for both)
 223         state = bp_rule.get_state()
 224         self.assert_(state == 1)
 225
 226
 227
 228
 229     # We will try a simple 1of: bd1 OR/AND db2
 230     def test_simple_1of_business_correlator(self):
 231         #
 232         # Config is not correct because of a wrong relative path
 233         # in the main config file
 234         #
 235         print "Get the hosts and services"
 236         now = time.time()
 237         host = self.sched.hosts.find_by_name("test_host_0")
 238         host.checks_in_progress = []
 239         host.act_depend_of = [] # ignore the router
 240         router = self.sched.hosts.find_by_name("test_router_0")
 241         router.checks_in_progress = []
 242         router.act_depend_of = [] # ignore the router
 243         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 244         svc.checks_in_progress = []
 245         svc.act_depend_of = [] # no hostchecks on critical checkresults
 246
 247         svc_bd1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db1")
 248         self.assert_(svc_bd1.got_business_rule == False)
 249         self.assert_(svc_bd1.business_rule is None)
 250         svc_bd2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db2")
 251         self.assert_(svc_bd2.got_business_rule == False)
 252         self.assert_(svc_bd2.business_rule is None)
 253         svc_cor = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "Simple_1Of")
 254         self.assert_(svc_cor.got_business_rule == True)
 255         self.assert_(svc_cor.business_rule is not None)
 256         bp_rule = svc_cor.business_rule
 257         self.assert_(bp_rule.operand == 'of:')
 258         self.assert_(bp_rule.of_values == 1)
 259
 260
 261         sons = bp_rule.sons
 262         print "Sons,", sons
 263         # We've got 2 sons, 2 services nodes
 264         self.assert_(len(sons) == 2)
 265         self.assert_(sons[0].operand == 'service')
 266         self.assert_(sons[0].sons[0] == svc_bd1)
 267         self.assert_(sons[1].operand == 'service')
 268         self.assert_(sons[1].sons[0] == svc_bd2)
 269
 270         # Now state working on the states
 271         self.scheduler_loop(1, [[svc_bd2, 0, 'OK | value1=1 value2=2'], [svc_bd1, 0, 'OK | rtt=10']])
 272         self.assert_(svc_bd1.state == 'OK')
 273         self.assert_(svc_bd1.state_type == 'HARD')
 274         self.assert_(svc_bd2.state == 'OK')
 275         self.assert_(svc_bd2.state_type == 'HARD')
 276
 277         state = bp_rule.get_state()
 278         self.assert_(state == 0)
 279
 280         # Now we set the bd1 as soft/CRITICAL
 281         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 282         self.assert_(svc_bd1.state == 'CRITICAL')
 283         self.assert_(svc_bd1.state_type == 'SOFT')
 284         self.assert_(svc_bd1.last_hard_state_id == 0)
 285
 286         # The business rule must still be 0
 287         # becase we want HARD states
 288         state = bp_rule.get_state()
 289         self.assert_(state == 0)
 290
 291         # Now we get bd1 CRITICAL/HARD
 292         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 293         self.assert_(svc_bd1.state == 'CRITICAL')
 294         self.assert_(svc_bd1.state_type == 'HARD')
 295         self.assert_(svc_bd1.last_hard_state_id == 2)
 296
 297         # The rule still be OK
 298         state = bp_rule.get_state()
 299         self.assert_(state == 0)
 300
 301         # Now we also set bd2 as CRITICAL/HARD...
 302         self.scheduler_loop(2, [[svc_bd2, 2, 'CRITICAL | value1=1 value2=2']])
 303         self.assert_(svc_bd2.state == 'CRITICAL')
 304         self.assert_(svc_bd2.state_type == 'HARD')
 305         self.assert_(svc_bd2.last_hard_state_id == 2)
 306
 307         # And now the state of the rule must be 2 now
 308         state = bp_rule.get_state()
 309         self.assert_(state == 2)
 310
 311         # And If we set one WARNING now?
 312         self.scheduler_loop(2, [[svc_bd1, 1, 'WARNING | value1=1 value2=2']])
 313         self.assert_(svc_bd1.state == 'WARNING')
 314         self.assert_(svc_bd1.state_type == 'HARD')
 315         self.assert_(svc_bd1.last_hard_state_id == 1)
 316
 317         # Must be WARNING (worse no 0 value for both, like for AND rule)
 318         state = bp_rule.get_state()
 319         self.assert_(state == 1)
 320
 321
 322
 323     # We will try a simple bd1 OR db2, but this time we will
 324     # schedule a real check and see if it's good
 325     def test_simple_or_business_correlator_with_schedule(self):
 326         #
 327         # Config is not correct because of a wrong relative path
 328         # in the main config file
 329         #
 330         print "Get the hosts and services"
 331         now = time.time()
 332         host = self.sched.hosts.find_by_name("test_host_0")
 333         host.checks_in_progress = []
 334         host.act_depend_of = [] # ignore the router
 335         router = self.sched.hosts.find_by_name("test_router_0")
 336         router.checks_in_progress = []
 337         router.act_depend_of = [] # ignore the router
 338         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 339         svc.checks_in_progress = []
 340         svc.act_depend_of = [] # no hostchecks on critical checkresults
 341
 342         svc_bd1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db1")
 343         self.assert_(svc_bd1.got_business_rule == False)
 344         self.assert_(svc_bd1.business_rule is None)
 345         svc_bd2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db2")
 346         self.assert_(svc_bd2.got_business_rule == False)
 347         self.assert_(svc_bd2.business_rule is None)
 348         svc_cor = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "Simple_Or")
 349         self.assert_(svc_cor.got_business_rule == True)
 350         self.assert_(svc_cor.business_rule is not None)
 351         bp_rule = svc_cor.business_rule
 352         self.assert_(bp_rule.operand == '|')
 353
 354         sons = bp_rule.sons
 355         print "Sons,", sons
 356         #We(ve got 2 sons, 2 services nodes
 357         self.assert_(len(sons) == 2)
 358         self.assert_(sons[0].operand == 'service')
 359         self.assert_(sons[0].sons[0] == svc_bd1)
 360         self.assert_(sons[1].operand == 'service')
 361         self.assert_(sons[1].sons[0] == svc_bd2)
 362
 363         # Now state working on the states
 364         self.scheduler_loop(1, [[svc_bd2, 0, 'OK | value1=1 value2=2'], [svc_bd1, 0, 'OK | rtt=10']])
 365         self.assert_(svc_bd1.state == 'OK')
 366         self.assert_(svc_bd1.state_type == 'HARD')
 367         self.assert_(svc_bd2.state == 'OK')
 368         self.assert_(svc_bd2.state_type == 'HARD')
 369
 370         state = bp_rule.get_state()
 371         self.assert_(state == 0)
 372
 373         print "Launch internal check"
 374         svc_cor.launch_check(now-1)
 375         c = svc_cor.actions[0]
 376         self.assert_(c.internal == True)
 377         self.assert_(c.is_launchable(now))
 378
 379         #ask the scheduler to launch this check
 380         #and ask 2 loops: one for launch the check
 381         #and another to integer the result
 382         self.scheduler_loop(2, [])
 383
 384         # We should have no more the check
 385         self.assert_(len(svc_cor.actions) == 0)
 386
 387         print "Look at svc_cor state", svc_cor.state
 388         # What is the svc_cor state now?
 389         self.assert_(svc_cor.state == 'OK')
 390         self.assert_(svc_cor.state_type == 'HARD')
 391         self.assert_(svc_cor.last_hard_state_id == 0)
 392
 393
 394         # Now we set the bd1 as soft/CRITICAL
 395         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 396         self.assert_(svc_bd1.state == 'CRITICAL')
 397         self.assert_(svc_bd1.state_type == 'SOFT')
 398         self.assert_(svc_bd1.last_hard_state_id == 0)
 399
 400         # The business rule must still be 0
 401         state = bp_rule.get_state()
 402         self.assert_(state == 0)
 403
 404         print "Launch internal check"
 405         svc_cor.launch_check(now-1)
 406         c = svc_cor.actions[0]
 407         self.assert_(c.internal == True)
 408         self.assert_(c.is_launchable(now))
 409
 410         #ask the scheduler to launch this check
 411         #and ask 2 loops: one for launch the check
 412         #and another to integer the result
 413         self.scheduler_loop(2, [])
 414
 415         # We should have no more the check
 416         self.assert_(len(svc_cor.actions) == 0)
 417
 418         print "Look at svc_cor state", svc_cor.state
 419         # What is the svc_cor state now?
 420         self.assert_(svc_cor.state == 'OK')
 421         self.assert_(svc_cor.state_type == 'HARD')
 422         self.assert_(svc_cor.last_hard_state_id == 0)
 423
 424
 425         # Now we get bd1 CRITICAL/HARD
 426         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 427         self.assert_(svc_bd1.state == 'CRITICAL')
 428         self.assert_(svc_bd1.state_type == 'HARD')
 429         self.assert_(svc_bd1.last_hard_state_id == 2)
 430
 431         # The rule must still be a 0 (or inside)
 432         state = bp_rule.get_state()
 433         self.assert_(state == 0)
 434
 435         print "Launch internal check"
 436         svc_cor.launch_check(now-1)
 437         c = svc_cor.actions[0]
 438         self.assert_(c.internal == True)
 439         self.assert_(c.is_launchable(now))
 440
 441         #ask the scheduler to launch this check
 442         #and ask 2 loops: one for launch the check
 443         #and another to integer the result
 444         self.scheduler_loop(2, [])
 445
 446         # We should have no more the check
 447         self.assert_(len(svc_cor.actions) == 0)
 448
 449         print "Look at svc_cor state", svc_cor.state
 450         # What is the svc_cor state now?
 451         self.assert_(svc_cor.state == 'OK')
 452         self.assert_(svc_cor.state_type == 'HARD')
 453         self.assert_(svc_cor.last_hard_state_id == 0)
 454
 455
 456         # Now we also set bd2 as CRITICAL/HARD... byebye 0 :)
 457         self.scheduler_loop(2, [[svc_bd2, 2, 'CRITICAL | value1=1 value2=2']])
 458         self.assert_(svc_bd2.state == 'CRITICAL')
 459         self.assert_(svc_bd2.state_type == 'HARD')
 460         self.assert_(svc_bd2.last_hard_state_id == 2)
 461
 462         # And now the state of the rule must be 2
 463         state = bp_rule.get_state()
 464         self.assert_(state == 2)
 465
 466         # And now we must be CRITICAL/SOFT!
 467         print "Launch internal check"
 468         svc_cor.launch_check(now-1)
 469         c = svc_cor.actions[0]
 470         self.assert_(c.internal == True)
 471         self.assert_(c.is_launchable(now))
 472
 473         #ask the scheduler to launch this check
 474         #and ask 2 loops: one for launch the check
 475         #and another to integer the result
 476         self.scheduler_loop(2, [])
 477
 478         # We should have no more the check
 479         self.assert_(len(svc_cor.actions) == 0)
 480
 481         print "Look at svc_cor state", svc_cor.state
 482         # What is the svc_cor state now?
 483         self.assert_(svc_cor.state == 'CRITICAL')
 484         self.assert_(svc_cor.state_type == 'SOFT')
 485         self.assert_(svc_cor.last_hard_state_id == 0)
 486
 487         #OK, re recheck again, GO HARD!
 488         print "Launch internal check"
 489         svc_cor.launch_check(now-1)
 490         c = svc_cor.actions[0]
 491         self.assert_(c.internal == True)
 492         self.assert_(c.is_launchable(now))
 493
 494         #ask the scheduler to launch this check
 495         #and ask 2 loops: one for launch the check
 496         #and another to integer the result
 497         self.scheduler_loop(2, [])
 498
 499         # We should have no more the check
 500         self.assert_(len(svc_cor.actions) == 0)
 501
 502         print "Look at svc_cor state", svc_cor.state
 503         # What is the svc_cor state now?
 504         self.assert_(svc_cor.state == 'CRITICAL')
 505         self.assert_(svc_cor.state_type == 'HARD')
 506         self.assert_(svc_cor.last_hard_state_id == 2)
 507
 508
 509         # And If we set one WARNING?
 510         self.scheduler_loop(2, [[svc_bd2, 1, 'WARNING | value1=1 value2=2']])
 511         self.assert_(svc_bd2.state == 'WARNING')
 512         self.assert_(svc_bd2.state_type == 'HARD')
 513         self.assert_(svc_bd2.last_hard_state_id == 1)
 514
 515         # Must be WARNING (better no 0 value)
 516         state = bp_rule.get_state()
 517         self.assert_(state == 1)
 518
 519         # And in a HARD
 520         print "Launch internal check"
 521         svc_cor.launch_check(now-1)
 522         c = svc_cor.actions[0]
 523         self.assert_(c.internal == True)
 524         self.assert_(c.is_launchable(now))
 525
 526         #ask the scheduler to launch this check
 527         #and ask 2 loops: one for launch the check
 528         #and another to integer the result
 529         self.scheduler_loop(2, [])
 530
 531         # We should have no more the check
 532         self.assert_(len(svc_cor.actions) == 0)
 533
 534         print "Look at svc_cor state", svc_cor.state
 535         # What is the svc_cor state now?
 536         self.assert_(svc_cor.state == 'WARNING')
 537         self.assert_(svc_cor.state_type == 'HARD')
 538         self.assert_(svc_cor.last_hard_state_id == 1)
 539
 540         print "All elements", bp_rule.list_all_elements()
 541
 542         print "IMPACT:", svc_bd2.impacts
 543         for i in svc_bd2.impacts:
 544             print i.get_name()
 545
 546         # Assert that Simple_Or Is an impact of the problem bd2
 547         self.assert_(svc_cor in svc_bd2.impacts)
 548         # and bd1 too
 549         self.assert_(svc_cor in svc_bd1.impacts)
 550
 551
 552     def test_dep_node_list_elements(self):
 553         svc_bd1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db1")
 554         self.assert_(svc_bd1.got_business_rule == False)
 555         self.assert_(svc_bd1.business_rule is None)
 556         svc_bd2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db2")
 557         self.assert_(svc_bd2.got_business_rule == False)
 558         self.assert_(svc_bd2.business_rule is None)
 559         svc_cor = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "Simple_Or")
 560         self.assert_(svc_cor.got_business_rule == True)
 561         self.assert_(svc_cor.business_rule is not None)
 562         bp_rule = svc_cor.business_rule
 563         self.assert_(bp_rule.operand == '|')
 564
 565         print "All elements", bp_rule.list_all_elements()
 566         all_elt = bp_rule.list_all_elements()
 567
 568         self.assert_(svc_bd2 in all_elt)
 569         self.assert_(svc_bd1 in all_elt)
 570
 571         print "DBG: bd2 depend_on_me", svc_bd2.act_depend_of_me
 572
 573     # We will try a full ERP rule and
 574     # schedule a real check and see if it's good
 575     def test_full_erp_rule_with_schedule(self):
 576         #
 577         # Config is not correct because of a wrong relative path
 578         # in the main config file
 579         #
 580         print "Get the hosts and services"
 581         now = time.time()
 582         host = self.sched.hosts.find_by_name("test_host_0")
 583         host.checks_in_progress = []
 584         host.act_depend_of = [] # ignore the router
 585         router = self.sched.hosts.find_by_name("test_router_0")
 586         router.checks_in_progress = []
 587         router.act_depend_of = [] # ignore the router
 588         svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
 589         svc.checks_in_progress = []
 590         svc.act_depend_of = [] # no hostchecks on critical checkresults
 591
 592         svc_bd1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db1")
 593         self.assert_(svc_bd1.got_business_rule == False)
 594         self.assert_(svc_bd1.business_rule is None)
 595         svc_bd2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "db2")
 596         self.assert_(svc_bd2.got_business_rule == False)
 597         self.assert_(svc_bd2.business_rule is None)
 598         svc_web1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "web1")
 599         self.assert_(svc_web1.got_business_rule == False)
 600         self.assert_(svc_web1.business_rule is None)
 601         svc_web2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "web2")
 602         self.assert_(svc_web2.got_business_rule == False)
 603         self.assert_(svc_web2.business_rule is None)
 604         svc_lvs1 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "lvs1")
 605         self.assert_(svc_lvs1.got_business_rule == False)
 606         self.assert_(svc_lvs1.business_rule is None)
 607         svc_lvs2 = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "lvs2")
 608         self.assert_(svc_lvs2.got_business_rule == False)
 609         self.assert_(svc_lvs2.business_rule is None)
 610         svc_cor = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "ERP")
 611         self.assert_(svc_cor.got_business_rule == True)
 612         self.assert_(svc_cor.business_rule is not None)
 613         bp_rule = svc_cor.business_rule
 614         self.assert_(bp_rule.operand == '&')
 615
 616         sons = bp_rule.sons
 617         print "Sons,", sons
 618         #We've got 3 sons, each 3 rules
 619         self.assert_(len(sons) == 3)
 620         bd_node = sons[0]
 621         self.assert_(bd_node.operand == '|')
 622         self.assert_(bd_node.sons[0].sons[0] == svc_bd1)
 623         self.assert_(bd_node.sons[1].sons[0] == svc_bd2)
 624
 625         # Now state working on the states
 626         self.scheduler_loop(1, [[svc_bd2, 0, 'OK | value1=1 value2=2'], [svc_bd1, 0, 'OK | rtt=10']])
 627         self.assert_(svc_bd1.state == 'OK')
 628         self.assert_(svc_bd1.state_type == 'HARD')
 629         self.assert_(svc_bd2.state == 'OK')
 630         self.assert_(svc_bd2.state_type == 'HARD')
 631
 632         state = bp_rule.get_state()
 633         self.assert_(state == 0)
 634
 635         print "Launch internal check"
 636         svc_cor.launch_check(now-1)
 637         c = svc_cor.actions[0]
 638         self.assert_(c.internal == True)
 639         self.assert_(c.is_launchable(now))
 640
 641         #ask the scheduler to launch this check
 642         #and ask 2 loops: one for launch the check
 643         #and another to integer the result
 644         self.scheduler_loop(2, [])
 645
 646         # We should have no more the check
 647         self.assert_(len(svc_cor.actions) == 0)
 648
 649         print "Look at svc_cor state", svc_cor.state
 650         # What is the svc_cor state now?
 651         self.assert_(svc_cor.state == 'OK')
 652         self.assert_(svc_cor.state_type == 'HARD')
 653         self.assert_(svc_cor.last_hard_state_id == 0)
 654
 655
 656         # Now we set the bd1 as soft/CRITICAL
 657         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 658         self.assert_(svc_bd1.state == 'CRITICAL')
 659         self.assert_(svc_bd1.state_type == 'SOFT')
 660         self.assert_(svc_bd1.last_hard_state_id == 0)
 661
 662         # The business rule must still be 0
 663         state = bp_rule.get_state()
 664         self.assert_(state == 0)
 665
 666         print "Launch internal check"
 667         svc_cor.launch_check(now-1)
 668         c = svc_cor.actions[0]
 669         self.assert_(c.internal == True)
 670         self.assert_(c.is_launchable(now))
 671
 672         #ask the scheduler to launch this check
 673         #and ask 2 loops: one for launch the check
 674         #and another to integer the result
 675         self.scheduler_loop(2, [])
 676
 677         # We should have no more the check
 678         self.assert_(len(svc_cor.actions) == 0)
 679
 680         print "ERP: Look at svc_cor state", svc_cor.state
 681         # What is the svc_cor state now?
 682         self.assert_(svc_cor.state == 'OK')
 683         self.assert_(svc_cor.state_type == 'HARD')
 684         self.assert_(svc_cor.last_hard_state_id == 0)
 685
 686
 687         # Now we get bd1 CRITICAL/HARD
 688         self.scheduler_loop(1, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2']])
 689         self.assert_(svc_bd1.state == 'CRITICAL')
 690         self.assert_(svc_bd1.state_type == 'HARD')
 691         self.assert_(svc_bd1.last_hard_state_id == 2)
 692
 693         # The rule must still be a 0 (or inside)
 694         state = bp_rule.get_state()
 695         self.assert_(state == 0)
 696
 697         print "ERP: Launch internal check"
 698         svc_cor.launch_check(now-1)
 699         c = svc_cor.actions[0]
 700         self.assert_(c.internal == True)
 701         self.assert_(c.is_launchable(now))
 702
 703         #ask the scheduler to launch this check
 704         #and ask 2 loops: one for launch the check
 705         #and another to integer the result
 706         self.scheduler_loop(2, [])
 707
 708         # We should have no more the check
 709         self.assert_(len(svc_cor.actions) == 0)
 710
 711         print "ERP: Look at svc_cor state", svc_cor.state
 712         # What is the svc_cor state now?
 713         self.assert_(svc_cor.state == 'OK')
 714         self.assert_(svc_cor.state_type == 'HARD')
 715         self.assert_(svc_cor.last_hard_state_id == 0)
 716
 717
 718         # Now we also set bd2 as CRITICAL/HARD... byebye 0 :)
 719         self.scheduler_loop(2, [[svc_bd2, 2, 'CRITICAL | value1=1 value2=2']])
 720         self.assert_(svc_bd2.state == 'CRITICAL')
 721         self.assert_(svc_bd2.state_type == 'HARD')
 722         self.assert_(svc_bd2.last_hard_state_id == 2)
 723
 724         # And now the state of the rule must be 2
 725         state = bp_rule.get_state()
 726         self.assert_(state == 2)
 727
 728         # And now we must be CRITICAL/SOFT!
 729         print "ERP: Launch internal check"
 730         svc_cor.launch_check(now-1)
 731         c = svc_cor.actions[0]
 732         self.assert_(c.internal == True)
 733         self.assert_(c.is_launchable(now))
 734
 735         #ask the scheduler to launch this check
 736         #and ask 2 loops: one for launch the check
 737         #and another to integer the result
 738         self.scheduler_loop(2, [])
 739
 740         # We should have no more the check
 741         self.assert_(len(svc_cor.actions) == 0)
 742
 743         print "ERP: Look at svc_cor state", svc_cor.state
 744         # What is the svc_cor state now?
 745         self.assert_(svc_cor.state == 'CRITICAL')
 746         self.assert_(svc_cor.state_type == 'SOFT')
 747         self.assert_(svc_cor.last_hard_state_id == 0)
 748
 749         #OK, re recheck again, GO HARD!
 750         print "ERP: Launch internal check"
 751         svc_cor.launch_check(now-1)
 752         c = svc_cor.actions[0]
 753         self.assert_(c.internal == True)
 754         self.assert_(c.is_launchable(now))
 755
 756         #ask the scheduler to launch this check
 757         #and ask 2 loops: one for launch the check
 758         #and another to integer the result
 759         self.scheduler_loop(2, [])
 760
 761         # We should have no more the check
 762         self.assert_(len(svc_cor.actions) == 0)
 763
 764         print "ERP: Look at svc_cor state", svc_cor.state
 765         # What is the svc_cor state now?
 766         self.assert_(svc_cor.state == 'CRITICAL')
 767         self.assert_(svc_cor.state_type == 'HARD')
 768         self.assert_(svc_cor.last_hard_state_id == 2)
 769
 770
 771         # And If we set one WARNING?
 772         self.scheduler_loop(2, [[svc_bd2, 1, 'WARNING | value1=1 value2=2']])
 773         self.assert_(svc_bd2.state == 'WARNING')
 774         self.assert_(svc_bd2.state_type == 'HARD')
 775         self.assert_(svc_bd2.last_hard_state_id == 1)
 776
 777         # Must be WARNING (better no 0 value)
 778         state = bp_rule.get_state()
 779         self.assert_(state == 1)
 780
 781         # And in a HARD
 782         print "ERP: Launch internal check"
 783         svc_cor.launch_check(now-1)
 784         c = svc_cor.actions[0]
 785         self.assert_(c.internal == True)
 786         self.assert_(c.is_launchable(now))
 787
 788         #ask the scheduler to launch this check
 789         #and ask 2 loops: one for launch the check
 790         #and another to integer the result
 791         self.scheduler_loop(2, [])
 792
 793         # We should have no more the check
 794         self.assert_(len(svc_cor.actions) == 0)
 795
 796         print "ERP: Look at svc_cor state", svc_cor.state
 797         # What is the svc_cor state now?
 798         self.assert_(svc_cor.state == 'WARNING')
 799         self.assert_(svc_cor.state_type == 'HARD')
 800         self.assert_(svc_cor.last_hard_state_id == 1)
 801
 802         print "All elements", bp_rule.list_all_elements()
 803
 804         print "IMPACT:", svc_bd2.impacts
 805         for i in svc_bd2.impacts:
 806             print i.get_name()
 807
 808         # Assert that Simple_Or Is an impact of the problem bd2
 809         self.assert_(svc_cor in svc_bd2.impacts)
 810         # and bd1 too
 811         self.assert_(svc_cor in svc_bd1.impacts)
 812
 813         # And now all is green :)
 814         self.scheduler_loop(2, [[svc_bd2, 0, 'OK | value1=1 value2=2'], [svc_bd1, 0, 'OK | value1=1 value2=2']])
 815
 816         print "ERP: Launch internal check"
 817         svc_cor.launch_check(now-1)
 818         c = svc_cor.actions[0]
 819         self.assert_(c.internal == True)
 820         self.assert_(c.is_launchable(now))
 821
 822         #ask the scheduler to launch this check
 823         #and ask 2 loops: one for launch the check
 824         #and another to integer the result
 825         self.scheduler_loop(2, [])
 826
 827         # We should have no more the check
 828         self.assert_(len(svc_cor.actions) == 0)
 829
 830         print "ERP: Look at svc_cor state", svc_cor.state
 831         # What is the svc_cor state now?
 832         self.assert_(svc_cor.state == 'OK')
 833         self.assert_(svc_cor.state_type == 'HARD')
 834         self.assert_(svc_cor.last_hard_state_id == 0)
 835
 836         # And no more in impact
 837         self.assert_(svc_cor not in svc_bd2.impacts)
 838         self.assert_(svc_cor not in svc_bd1.impacts)
 839
 840         # And what if we set 2 service from distant rule CRITICAL?
 841         # ERP should be still OK
 842         # And now all is green :)
 843         self.scheduler_loop(2, [[svc_bd1, 2, 'CRITICAL | value1=1 value2=2'], [svc_web1, 2, 'CRITICAL | value1=1 value2=2']])
 844
 845         print "ERP: Launch internal check"
 846         svc_cor.launch_check(now-1)
 847         c = svc_cor.actions[0]
 848         self.assert_(c.internal == True)
 849         self.assert_(c.is_launchable(now))
 850
 851         #ask the scheduler to launch this check
 852         #and ask 2 loops: one for launch the check
 853         #and another to integer the result
 854         self.scheduler_loop(2, [])
 855
 856         # We should have no more the check
 857         self.assert_(len(svc_cor.actions) == 0)
 858
 859         print "ERP: Look at svc_cor state", svc_cor.state
 860         # What is the svc_cor state now?
 861         self.assert_(svc_cor.state == 'OK')
 862         self.assert_(svc_cor.state_type == 'HARD')
 863         self.assert_(svc_cor.last_hard_state_id == 0)
 864
 865
 866
 867 class TestConfigBroken(ShinkenTest):
 868     """A class with a broken configuration, where business rules reference unknown hosts/services"""
 869     def setUp(self):
 870         self.setup_with_file('etc/nagios_business_correlator_broken.cfg')
 871
 872
 873     def test_conf_is_correct(self):
 874         #
 875         # Business rules use services which don't exist. We want
 876         # the arbiter to output an error message and exit
 877         # in a controlled manner.
 878         #
 879         print "conf_is_correct", self.conf.conf_is_correct
 880         self.assert_(not self.conf.conf_is_correct)
 881
 882         # Get the arbiter's log broks
 883         logs = [b.data['log'] for b in self.broks.values() if b.type == 'log']
 884
 885         # Simple_1Of_1unk_svc : my business rule is invalid
 886         # Simple_1Of_1unk_svc : Business rule uses unknown service test_host_0/db3
 887         self.assert_(len([log for log in logs if re.search('Simple_1Of_1unk_svc', log)]) == 2)
 888         self.assert_(len([log for log in logs if re.search('service test_host_0/db3', log)]) == 1)
 889         # ERP_unk_svc : my business rule is invalid
 890         # ERP_unk_svc : Business rule uses unknown service test_host_0/web100
 891         # ERP_unk_svc : Business rule uses unknown service test_host_0/lvs100
 892         self.assert_(len([log for log in logs if re.search('ERP_unk_svc', log)]) == 3)
 893         self.assert_(len([log for log in logs if re.search('service test_host_0/web100', log)]) == 1)
 894         self.assert_(len([log for log in logs if re.search('service test_host_0/lvs100', log)]) == 1)
 895         # Simple_1Of_1unk_host : my business rule is invalid
 896         # Simple_1Of_1unk_host : Business rule uses unknown host test_host_9
 897         self.assert_(len([log for log in logs if re.search('Simple_1Of_1unk_host', log)]) == 2)
 898         self.assert_(len([log for log in logs if re.search('host test_host_9', log)]) == 1)
 899
 900         # Now the number of all failed business rules.
 901         self.assert_(len([log for log in logs if re.search('my business rule is invalid', log)]) == 3)
 902
 903
 904
 905 if __name__ == '__main__':
 906     unittest.main()
 907