• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python2
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."""
8import mock
9import unittest
10import common
11
12from autotest_lib.server.cros import host_lock_manager
13from autotest_lib.server.cros.chaos_lib import chaos_datastore_utils
14
15
16class HostLockManagerTest(unittest.TestCase):
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 MockHostLockManager(host_lock_manager.HostLockManager):
30        """Mock out _host_modifier() in HostLockManager class..
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 _host_modifier(self, hosts, operation, lock_reason=''):
38            """Overwrites original _host_modifier().
39
40            Add hosts to self.locked_hosts for LOCK and remove hosts from
41            self.locked_hosts for UNLOCK.
42
43            @param a set of strings, host names.
44            @param operation: a string, LOCK or UNLOCK.
45            @param lock_reason: a string, a reason for locking the hosts
46            """
47            if operation == self.LOCK:
48                assert lock_reason
49                self.locked_hosts = self.locked_hosts.union(hosts)
50            elif operation == self.UNLOCK:
51                self.locked_hosts = self.locked_hosts.difference(hosts)
52
53
54    def setUp(self):
55        super(HostLockManagerTest, self).setUp()
56        self.manager = host_lock_manager.HostLockManager()
57
58
59    # Patch mock object to return host as unknown from DataStore
60    @mock.patch.object(chaos_datastore_utils.ChaosDataStoreUtils, 'show_device',
61        return_value=False)
62    def testCheckHost_SkipsUnknownHost(self, get_mock):
63        actual = self.manager._check_host('host1', None)
64        self.assertEquals(None, actual)
65
66
67    @mock.patch.object(chaos_datastore_utils.ChaosDataStoreUtils, 'show_device',
68        return_value={'lock_status': True, 'locked_by': 'Mock',
69        'lock_status_updated': 'fake_time'})
70    def testCheckHost_DetectsLockedHost(self, get_mock):
71        """Test that a host which is already locked is skipped."""
72        actual = self.manager._check_host(self.HOST1, self.manager.LOCK)
73        self.assertEquals(None, actual)
74
75
76    @mock.patch.object(chaos_datastore_utils.ChaosDataStoreUtils, 'show_device',
77        return_value={'lock_status': False, 'locked_by': 'Mock',
78        'lock_status_updated': 'fake_time'})
79    def testCheckHost_DetectsUnlockedHost(self, get_mock):
80        """Test that a host which is already unlocked is skipped."""
81        actual = self.manager._check_host(self.HOST1, self.manager.UNLOCK)
82        self.assertEquals(None, actual)
83
84
85    @mock.patch.object(chaos_datastore_utils.ChaosDataStoreUtils, 'show_device',
86        return_value={'lock_status': False, 'locked_by': 'Mock',
87        'lock_status_updated': 'fake_time'})
88    def testCheckHost_ReturnsHostToLock(self, get_mock):
89        """Test that a host which can be locked is returned."""
90        actual = self.manager._check_host(self.HOST1, self.manager.LOCK)
91        self.assertEquals(self.HOST1, actual)
92
93
94    @mock.patch.object(chaos_datastore_utils.ChaosDataStoreUtils, 'show_device',
95        return_value={'lock_status': True, 'locked_by': 'Mock',
96        'lock_status_updated': 'fake_time'})
97    def testCheckHost_ReturnsHostToUnlock(self, get_mock):
98        """Test that a host which can be unlocked is returned."""
99        actual = self.manager._check_host(self.HOST1, self.manager.UNLOCK)
100        self.assertEquals(self.HOST1, actual)
101
102
103    def testLock_WithNonOverlappingHosts(self):
104        """Tests host locking, all hosts not in self.locked_hosts."""
105        hosts = [self.HOST2]
106        manager = self.MockHostLockManager()
107        manager.locked_hosts = set([self.HOST1])
108        manager.lock(hosts, lock_reason='Locking for test')
109        self.assertEquals(set([self.HOST1, self.HOST2]), manager.locked_hosts)
110
111
112    def testLock_WithPartialOverlappingHosts(self):
113        """Tests host locking, some hosts not in self.locked_hosts."""
114        hosts = [self.HOST1, self.HOST2]
115        manager = self.MockHostLockManager()
116        manager.locked_hosts = set([self.HOST1, self.HOST3])
117        manager.lock(hosts, lock_reason='Locking for test')
118        self.assertEquals(set([self.HOST1, self.HOST2, self.HOST3]),
119                          manager.locked_hosts)
120
121
122    def testLock_WithFullyOverlappingHosts(self):
123        """Tests host locking, all hosts in self.locked_hosts."""
124        hosts = [self.HOST1, self.HOST2]
125        self.manager.locked_hosts = set(hosts)
126        self.manager.lock(hosts)
127        self.assertEquals(set(hosts), self.manager.locked_hosts)
128
129
130    def testUnlock_WithNonOverlappingHosts(self):
131        """Tests host unlocking, all hosts not in self.locked_hosts."""
132        hosts = [self.HOST2]
133        self.manager.locked_hosts = set([self.HOST1])
134        self.manager.unlock(hosts)
135        self.assertEquals(set([self.HOST1]), self.manager.locked_hosts)
136
137
138if __name__ == '__main__':
139    unittest.main()
140