Merge branch 'master' of ssh://naparuba@shinken.git.sourceforge.net/gitroot/shinken...
[shinken.git] / test / test_external_commands.py
blob40e7f55b4a63fff86ad6519e5e387a4b78e608d8
1 #!/usr/bin/env python2.6
2 #Copyright (C) 2009-2010 :
3 # Gabes Jean, naparuba@gmail.com
4 # Gerhard Lausser, Gerhard.Lausser@consol.de
6 #This file is part of Shinken.
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.
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.
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/>.
22 # This file is used to test reading and processing of config files
25 #It's ugly I know....
26 from shinken_test import *
27 import os
29 class TestConfig(ShinkenTest):
30 #setUp is in shinken_test
32 def send_cmd(self, line):
33 s = '[%d] %s\n' % (int(time.time()), line)
34 print "Writing %s in %s" % (s, self.conf.command_file)
35 fd = open(self.conf.command_file, 'wb')
36 fd.write(s)
37 fd.close()
40 #Change ME :)
41 def test_external_comand(self):
42 now = time.time()
43 host = self.sched.hosts.find_by_name("test_host_0")
44 router = self.sched.hosts.find_by_name("test_router_0")
45 svc = self.sched.services.find_srv_by_name_and_hostname("test_host_0", "test_ok_0")
46 self.scheduler_loop(2, [[host, 0, 'UP | value1=1 value2=2'], [router, 0, 'UP | rtt=10'], [svc, 2, 'BAD | value1=0 value2=0']])
47 self.assert_(host.state == 'UP')
48 self.assert_(host.state_type == 'HARD')
50 excmd = '[%d] PROCESS_HOST_CHECK_RESULT;test_host_0;2;Bob is not happy' % int(time.time())
52 self.sched.run_external_command(excmd)
53 self.scheduler_loop(1, [])
54 self.scheduler_loop(1, []) #Need 2 run for get then consume)
55 self.assert_(host.state == 'DOWN')
56 self.assert_(host.output == 'Bob is not happy')
59 #Clean the command_file
60 #try:
61 # os.unlink(self.conf.command_file)
62 #except :
63 # pass
66 if __name__ == '__main__':
67 unittest.main()