OLD | NEW |
| (Empty) |
1 #!/usr/bin/python | |
2 | |
3 # Copyright (c) 2010 The Chromium OS Authors. All rights reserved. | |
4 # Use of this source code is governed by a BSD-style license that can be | |
5 # found in the LICENSE file. | |
6 | |
7 """Units tests for cbuildbot_comm commands.""" | |
8 | |
9 import cbuildbot_comm | |
10 import sys | |
11 import threading | |
12 import time | |
13 import unittest | |
14 | |
15 _TEST_CONFIG = {'test_slave' : | |
16 {'master' : False, | |
17 'hostname' : 'localhost', | |
18 'important' : True}, | |
19 'test_master' : | |
20 {'master' : True, | |
21 'important' : False | |
22 } | |
23 } | |
24 | |
25 # Reduce timeouts. | |
26 cbuildbot_comm._HEARTBEAT_TIMEOUT = 2 | |
27 cbuildbot_comm._MAX_TIMEOUT = 6 | |
28 | |
29 class _MasterSendBadStatus(threading.Thread): | |
30 | |
31 def __init__(self, test_class): | |
32 threading.Thread.__init__(self) | |
33 self.test_class = test_class | |
34 | |
35 def run(self): | |
36 # Sleep for heartbeat timeout to let slave start up. | |
37 time.sleep(2) | |
38 return_value = cbuildbot_comm._SendCommand('localhost', 'bad-command', | |
39 'args') | |
40 self.test_class.assertEqual(return_value, | |
41 cbuildbot_comm._STATUS_COMMAND_REJECTED) | |
42 | |
43 class _MasterCheckStatusThread(threading.Thread): | |
44 | |
45 def __init__(self, config, expected_return, test_class): | |
46 threading.Thread.__init__(self) | |
47 self.config = config | |
48 self.expected_return = expected_return | |
49 self.test_class = test_class | |
50 | |
51 def run(self): | |
52 return_value = cbuildbot_comm.HaveSlavesCompleted(self.config) | |
53 self.test_class.assertEqual(return_value, self.expected_return) | |
54 | |
55 | |
56 class CBuildBotCommTest(unittest.TestCase): | |
57 | |
58 def testSlaveComplete(self): | |
59 print >> sys.stderr, '\n>>> Running testSlaveComplete\n' | |
60 # Master should check statuses in another thread. | |
61 master_thread = _MasterCheckStatusThread(_TEST_CONFIG, True, self) | |
62 master_thread.start() | |
63 | |
64 return_value = cbuildbot_comm.PublishStatus( | |
65 cbuildbot_comm.STATUS_BUILD_COMPLETE) | |
66 self.assertEqual(return_value, True) | |
67 | |
68 def testMasterTimeout(self): | |
69 print >> sys.stderr, '\n>>> Running testMasterTimeout\n' | |
70 return_value = cbuildbot_comm.HaveSlavesCompleted(_TEST_CONFIG) | |
71 self.assertEqual(return_value, False) | |
72 | |
73 def testSlaveTimeout(self): | |
74 print >> sys.stderr, '\n>>> Running testSlaveTimeout\n' | |
75 return_value = cbuildbot_comm.PublishStatus( | |
76 cbuildbot_comm.STATUS_BUILD_COMPLETE) | |
77 self.assertEqual(return_value, False) | |
78 | |
79 def testSlaveFail(self): | |
80 print >> sys.stderr, '\n>>> Running testSlaveFail\n' | |
81 # Master should check statuses in another thread. | |
82 master_thread = _MasterCheckStatusThread(_TEST_CONFIG, False, self) | |
83 master_thread.start() | |
84 | |
85 return_value = cbuildbot_comm.PublishStatus( | |
86 cbuildbot_comm.STATUS_BUILD_FAILED) | |
87 self.assertEqual(return_value, True) | |
88 | |
89 def testBadCommand(self): | |
90 print >> sys.stderr, '\n>>> Running testSendBadCommand\n' | |
91 # Master should check statuses in another thread. | |
92 master_thread = _MasterSendBadStatus(self) | |
93 master_thread.start() | |
94 | |
95 return_value = cbuildbot_comm.PublishStatus( | |
96 cbuildbot_comm.STATUS_BUILD_COMPLETE) | |
97 self.assertEqual(return_value, False) | |
98 | |
99 | |
100 if __name__ == '__main__': | |
101 unittest.main() | |
OLD | NEW |