1 #!/usr/bin/python 2 # 3 # Copyright (c) 2012 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 """Unit tests for server/cros/host_lock_manager.py.""" 8 9 import mox 10 import unittest 11 import common 12 13 from autotest_lib.server import frontend 14 from autotest_lib.server.cros import host_lock_manager 15 16 class HostLockManagerTest(mox.MoxTestBase): 17 """Unit tests for host_lock_manager.HostLockManager. 18 19 @attribute HOST1: a string, fake host. 20 @attribute HOST2: a string, fake host. 21 @attribute HOST3: a string, fake host. 22 """ 23 24 HOST1 = 'host1' 25 HOST2 = 'host2' 26 HOST3 = 'host3' 27 28 29 class FakeHost(object): 30 """Fake version of Host object defiend in server/frontend.py. 31 32 @attribute locked: a boolean, True == host is locked. 33 @attribute locked_by: a string, fake user. 34 @attribute lock_time: a string, fake timestamp. 35 """ 36 37 def __init__(self, locked=False): 38 """Initialize. 39 40 @param locked: a boolean, True == host is locked. 41 """ 42 self.locked = locked 43 self.locked_by = 'fake_user' 44 self.lock_time = 'fake time' 45 46 47 class MockHostLockManager(host_lock_manager.HostLockManager): 48 """Mock out _host_modifier() in HostLockManager class.. 49 50 @attribute locked: a boolean, True == host is locked. 51 @attribute locked_by: a string, fake user. 52 @attribute lock_time: a string, fake timestamp. 53 """ 54 55 def _host_modifier(self, hosts, operation, lock_reason=''): 56 """Overwrites original _host_modifier(). 57 58 Add hosts to self.locked_hosts for LOCK and remove hosts from 59 self.locked_hosts for UNLOCK. 60 61 @param a set of strings, host names. 62 @param operation: a string, LOCK or UNLOCK. 63 @param lock_reason: a string, a reason for locking the hosts 64 """ 65 if operation == self.LOCK: 66 assert lock_reason 67 self.locked_hosts = self.locked_hosts.union(hosts) 68 elif operation == self.UNLOCK: 69 self.locked_hosts = self.locked_hosts.difference(hosts) 70 71 72 def setUp(self): 73 super(HostLockManagerTest, self).setUp() 74 self.afe = self.mox.CreateMock(frontend.AFE) 75 self.manager = host_lock_manager.HostLockManager(self.afe) 76 77 78 def testCheckHost_SkipsUnknownHost(self): 79 """Test that host unknown to AFE is skipped.""" 80 self.afe.get_hosts(hostname=self.HOST1).AndReturn(None) 81 self.mox.ReplayAll() 82 actual = self.manager._check_host(self.HOST1, None) 83 self.assertEquals(None, actual) 84 85 86 def testCheckHost_DetectsLockedHost(self): 87 """Test that a host which is already locked is skipped.""" 88 host_info = [self.FakeHost(locked=True)] 89 self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info) 90 self.mox.ReplayAll() 91 actual = self.manager._check_host(self.HOST1, self.manager.LOCK) 92 self.assertEquals(None, actual) 93 94 95 def testCheckHost_DetectsUnlockedHost(self): 96 """Test that a host which is already unlocked is skipped.""" 97 host_info = [self.FakeHost()] 98 self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info) 99 self.mox.ReplayAll() 100 actual = self.manager._check_host(self.HOST1, self.manager.UNLOCK) 101 self.assertEquals(None, actual) 102 103 104 def testCheckHost_ReturnsHostToLock(self): 105 """Test that a host which can be locked is returned.""" 106 host_info = [self.FakeHost()] 107 self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info) 108 self.mox.ReplayAll() 109 host_with_dot = '.'.join([self.HOST1, 'cros']) 110 actual = self.manager._check_host(host_with_dot, self.manager.LOCK) 111 self.assertEquals(self.HOST1, actual) 112 113 114 def testCheckHost_ReturnsHostToUnlock(self): 115 """Test that a host which can be unlocked is returned.""" 116 host_info = [self.FakeHost(locked=True)] 117 self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info) 118 self.mox.ReplayAll() 119 host_with_dot = '.'.join([self.HOST1, 'cros']) 120 actual = self.manager._check_host(host_with_dot, self.manager.UNLOCK) 121 self.assertEquals(self.HOST1, actual) 122 123 124 def testLock_WithNonOverlappingHosts(self): 125 """Tests host locking, all hosts not in self.locked_hosts.""" 126 hosts = [self.HOST2] 127 manager = self.MockHostLockManager(self.afe) 128 manager.locked_hosts = set([self.HOST1]) 129 manager.lock(hosts, lock_reason='Locking for test') 130 self.assertEquals(set([self.HOST1, self.HOST2]), manager.locked_hosts) 131 132 133 def testLock_WithPartialOverlappingHosts(self): 134 """Tests host locking, some hosts not in self.locked_hosts.""" 135 hosts = [self.HOST1, self.HOST2] 136 manager = self.MockHostLockManager(self.afe) 137 manager.locked_hosts = set([self.HOST1, self.HOST3]) 138 manager.lock(hosts, lock_reason='Locking for test') 139 self.assertEquals(set([self.HOST1, self.HOST2, self.HOST3]), 140 manager.locked_hosts) 141 142 143 def testLock_WithFullyOverlappingHosts(self): 144 """Tests host locking, all hosts in self.locked_hosts.""" 145 hosts = [self.HOST1, self.HOST2] 146 self.manager.locked_hosts = set(hosts) 147 self.manager.lock(hosts) 148 self.assertEquals(set(hosts), self.manager.locked_hosts) 149 150 151 def testUnlock_WithNonOverlappingHosts(self): 152 """Tests host unlocking, all hosts not in self.locked_hosts.""" 153 hosts = [self.HOST2] 154 self.manager.locked_hosts = set([self.HOST1]) 155 self.manager.unlock(hosts) 156 self.assertEquals(set([self.HOST1]), self.manager.locked_hosts) 157 158 159 def testUnlock_WithPartialOverlappingHosts(self): 160 """Tests host locking, some hosts not in self.locked_hosts.""" 161 hosts = [self.HOST1, self.HOST2] 162 manager = self.MockHostLockManager(self.afe) 163 manager.locked_hosts = set([self.HOST1, self.HOST3]) 164 manager.unlock(hosts) 165 self.assertEquals(set([self.HOST3]), manager.locked_hosts) 166 167 168 def testUnlock_WithFullyOverlappingHosts(self): 169 """Tests host locking, all hosts in self.locked_hosts.""" 170 hosts = [self.HOST1, self.HOST2] 171 manager = self.MockHostLockManager(self.afe) 172 manager.locked_hosts = set([self.HOST1, self.HOST2, self.HOST3]) 173 manager.unlock(hosts) 174 self.assertEquals(set([self.HOST3]), manager.locked_hosts) 175 176 177 def testHostModifier_WithHostsToLock(self): 178 """Test host locking.""" 179 hosts = set([self.HOST1]) 180 self.manager.locked_hosts = set([self.HOST2]) 181 self.mox.StubOutWithMock(self.manager, '_check_host') 182 self.manager._check_host(self.HOST1, 183 self.manager.LOCK).AndReturn(self.HOST1) 184 self.afe.run('modify_hosts', 185 host_filter_data={'hostname__in': [self.HOST1]}, 186 update_data={'locked': True, 'lock_reason': 'Test'}) 187 self.mox.ReplayAll() 188 self.manager._host_modifier(hosts, self.manager.LOCK, 189 lock_reason='Test') 190 self.assertEquals(set([self.HOST1, self.HOST2]), 191 self.manager.locked_hosts) 192 193 194 def testHostModifier_WithHostsToUnlock(self): 195 """Test host unlocking.""" 196 hosts = set([self.HOST1]) 197 self.manager.locked_hosts = set([self.HOST1, self.HOST2]) 198 self.mox.StubOutWithMock(self.manager, '_check_host') 199 self.manager._check_host(self.HOST1, 200 self.manager.UNLOCK).AndReturn(self.HOST1) 201 self.afe.run('modify_hosts', 202 host_filter_data={'hostname__in': [self.HOST1]}, 203 update_data={'locked': False}) 204 self.mox.ReplayAll() 205 self.manager._host_modifier(hosts, self.manager.UNLOCK) 206 self.assertEquals(set([self.HOST2]), self.manager.locked_hosts) 207 208 209 def testHostModifier_WithoutLockReason(self): 210 """Test host locking without providing a lock reason.""" 211 hosts = set([self.HOST1]) 212 self.manager.locked_hosts = set([self.HOST2]) 213 self.mox.StubOutWithMock(self.manager, '_check_host') 214 self.manager._check_host(self.HOST1, 215 self.manager.LOCK).AndReturn(self.HOST1) 216 self.afe.run('modify_hosts', 217 host_filter_data={'hostname__in': [self.HOST1]}, 218 update_data={'locked': True, 219 'lock_reason': None}) 220 self.mox.ReplayAll() 221 self.manager._host_modifier(hosts, self.manager.LOCK) 222 self.assertEquals(set([self.HOST2]), self.manager.locked_hosts) 223 224 225 if __name__ == '__main__': 226 unittest.main() 227
