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