• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
9import mox
10import unittest
11import common
12
13from autotest_lib.server import frontend
14from autotest_lib.server.cros import host_lock_manager
15
16class 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
225if __name__ == '__main__':
226    unittest.main()
227