test/test_module_status_dat.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 #
  23 # This file is used to test reading and processing of config files
  24 #
  25
  26 import os
  27
  28 from shinken_test import unittest, ShinkenTest
  29
  30 from shinken.modules.status_dat_broker import get_instance
  31
  32
  33 class TestConfig(ShinkenTest):
  34     #setUp is in shinken_test
  35
  36     def nb_of_string(self, buf, s):
  37         nb_s = 0
  38         for line in buf.splitlines():
  39             if line == s:
  40                 print "Find string"
  41                 nb_s += 1
  42         return nb_s
  43
  44
  45     #Change ME :)
  46     def test_simplelog(self):
  47         print self.conf.modules
  48         #get our modules
  49         mod = None
  50         for m in self.conf.modules:
  51             if m.module_type == 'status_dat':
  52                 mod = m
  53         self.assert_(mod is not None)
  54         self.assert_(mod.status_file == '/usr/local/shinken/var/status.data')
  55         self.assert_(mod.module_name == 'Status-Dat')
  56         self.assert_(mod.object_cache_file == '/usr/local/shinken/var/objects.cache')
  57
  58         try :
  59             os.unlink(mod.status_file)
  60             os.unlink(mod.module_name)
  61         except :
  62             pass
  63
  64         sl = get_instance(mod)
  65         print sl
  66         #Hack here :(
  67         sl.properties = {}
  68         sl.properties['to_queue'] = None
  69         self.sched.fill_initial_broks()
  70         print self.sched.broks
  71         sl.init()
  72         for b in self.sched.broks.values():
  73             b.instance_id = 0
  74             sl.manage_brok(b)
  75
  76         #Now verify the objects.dat file
  77         sl.objects_cache.create_or_update()
  78         obj = open(mod.object_cache_file)
  79         buf = obj.read()
  80         obj.close()
  81         #Check for 1 service and only one
  82         nb_services = self.nb_of_string(buf, "define service {")
  83         self.assert_(nb_services == 1)
  84         #2 hosts : host and a router
  85         nb_hosts = self.nb_of_string(buf, "define host {")
  86         self.assert_(nb_hosts == 2)
  87
  88         #Same for status.dat.
  89         sl.status.create_or_update()
  90         status = open(mod.status_file)
  91         buf = status.read()
  92         obj.close()
  93
  94         nb_prog = self.nb_of_string(buf, "programstatus {")
  95         self.assert_(nb_prog == 1)
  96
  97         nb_hosts = self.nb_of_string(buf, "hoststatus {")
  98         self.assert_(nb_hosts == 2)
  99
 100         nb_services = self.nb_of_string(buf, "servicestatus {")
 101         self.assert_(nb_services  == 1)
 102
 103
 104         #now check if after a resend we still got the good number
 105         self.sched.broks.clear()
 106         self.sched.fill_initial_broks()
 107         for b in self.sched.broks.values():
 108             b.instance_id = 0
 109             sl.manage_brok(b)
 110
 111         #Now verify the objects.dat file
 112         sl.objects_cache.create_or_update()
 113         obj = open(mod.object_cache_file)
 114         buf = obj.read()
 115         obj.close()
 116         #Check for 1 service and only one
 117         nb_services = self.nb_of_string(buf, "define service {")
 118         self.assert_(nb_services == 1)
 119         #2 hosts : host and a router
 120         nb_hosts = self.nb_of_string(buf, "define host {")
 121         self.assert_(nb_hosts == 2)
 122
 123         #Same for status.dat.
 124         sl.status.create_or_update()
 125         status = open(mod.status_file)
 126         buf = status.read()
 127         obj.close()
 128
 129         nb_prog = self.nb_of_string(buf, "programstatus {")
 130         self.assert_(nb_prog == 1)
 131
 132         nb_hosts = self.nb_of_string(buf, "hoststatus {")
 133         self.assert_(nb_hosts == 2)
 134
 135         nb_services = self.nb_of_string(buf, "servicestatus {")
 136         self.assert_(nb_services  == 1)
 137
 138
 139         os.unlink(mod.status_file)
 140         os.unlink(mod.object_cache_file)
 141
 142
 143 if __name__ == '__main__':
 144     unittest.main()
 145