#!/usr/bin/python
|
#
|
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
|
# Use of this source code is governed by a BSD-style license that can be
|
# found in the LICENSE file.
|
|
"""Unit tests for server/cros/host_lock_manager.py."""
|
|
import mox
|
import unittest
|
import common
|
|
from autotest_lib.server import frontend
|
from autotest_lib.server.cros import host_lock_manager
|
|
class HostLockManagerTest(mox.MoxTestBase):
|
"""Unit tests for host_lock_manager.HostLockManager.
|
|
@attribute HOST1: a string, fake host.
|
@attribute HOST2: a string, fake host.
|
@attribute HOST3: a string, fake host.
|
"""
|
|
HOST1 = 'host1'
|
HOST2 = 'host2'
|
HOST3 = 'host3'
|
|
|
class FakeHost(object):
|
"""Fake version of Host object defiend in server/frontend.py.
|
|
@attribute locked: a boolean, True == host is locked.
|
@attribute locked_by: a string, fake user.
|
@attribute lock_time: a string, fake timestamp.
|
"""
|
|
def __init__(self, locked=False):
|
"""Initialize.
|
|
@param locked: a boolean, True == host is locked.
|
"""
|
self.locked = locked
|
self.locked_by = 'fake_user'
|
self.lock_time = 'fake time'
|
|
|
class MockHostLockManager(host_lock_manager.HostLockManager):
|
"""Mock out _host_modifier() in HostLockManager class..
|
|
@attribute locked: a boolean, True == host is locked.
|
@attribute locked_by: a string, fake user.
|
@attribute lock_time: a string, fake timestamp.
|
"""
|
|
def _host_modifier(self, hosts, operation, lock_reason=''):
|
"""Overwrites original _host_modifier().
|
|
Add hosts to self.locked_hosts for LOCK and remove hosts from
|
self.locked_hosts for UNLOCK.
|
|
@param a set of strings, host names.
|
@param operation: a string, LOCK or UNLOCK.
|
@param lock_reason: a string, a reason for locking the hosts
|
"""
|
if operation == self.LOCK:
|
assert lock_reason
|
self.locked_hosts = self.locked_hosts.union(hosts)
|
elif operation == self.UNLOCK:
|
self.locked_hosts = self.locked_hosts.difference(hosts)
|
|
|
def setUp(self):
|
super(HostLockManagerTest, self).setUp()
|
self.afe = self.mox.CreateMock(frontend.AFE)
|
self.manager = host_lock_manager.HostLockManager(self.afe)
|
|
|
def testCheckHost_SkipsUnknownHost(self):
|
"""Test that host unknown to AFE is skipped."""
|
self.afe.get_hosts(hostname=self.HOST1).AndReturn(None)
|
self.mox.ReplayAll()
|
actual = self.manager._check_host(self.HOST1, None)
|
self.assertEquals(None, actual)
|
|
|
def testCheckHost_DetectsLockedHost(self):
|
"""Test that a host which is already locked is skipped."""
|
host_info = [self.FakeHost(locked=True)]
|
self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info)
|
self.mox.ReplayAll()
|
actual = self.manager._check_host(self.HOST1, self.manager.LOCK)
|
self.assertEquals(None, actual)
|
|
|
def testCheckHost_DetectsUnlockedHost(self):
|
"""Test that a host which is already unlocked is skipped."""
|
host_info = [self.FakeHost()]
|
self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info)
|
self.mox.ReplayAll()
|
actual = self.manager._check_host(self.HOST1, self.manager.UNLOCK)
|
self.assertEquals(None, actual)
|
|
|
def testCheckHost_ReturnsHostToLock(self):
|
"""Test that a host which can be locked is returned."""
|
host_info = [self.FakeHost()]
|
self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info)
|
self.mox.ReplayAll()
|
host_with_dot = '.'.join([self.HOST1, 'cros'])
|
actual = self.manager._check_host(host_with_dot, self.manager.LOCK)
|
self.assertEquals(self.HOST1, actual)
|
|
|
def testCheckHost_ReturnsHostToUnlock(self):
|
"""Test that a host which can be unlocked is returned."""
|
host_info = [self.FakeHost(locked=True)]
|
self.afe.get_hosts(hostname=self.HOST1).AndReturn(host_info)
|
self.mox.ReplayAll()
|
host_with_dot = '.'.join([self.HOST1, 'cros'])
|
actual = self.manager._check_host(host_with_dot, self.manager.UNLOCK)
|
self.assertEquals(self.HOST1, actual)
|
|
|
def testLock_WithNonOverlappingHosts(self):
|
"""Tests host locking, all hosts not in self.locked_hosts."""
|
hosts = [self.HOST2]
|
manager = self.MockHostLockManager(self.afe)
|
manager.locked_hosts = set([self.HOST1])
|
manager.lock(hosts, lock_reason='Locking for test')
|
self.assertEquals(set([self.HOST1, self.HOST2]), manager.locked_hosts)
|
|
|
def testLock_WithPartialOverlappingHosts(self):
|
"""Tests host locking, some hosts not in self.locked_hosts."""
|
hosts = [self.HOST1, self.HOST2]
|
manager = self.MockHostLockManager(self.afe)
|
manager.locked_hosts = set([self.HOST1, self.HOST3])
|
manager.lock(hosts, lock_reason='Locking for test')
|
self.assertEquals(set([self.HOST1, self.HOST2, self.HOST3]),
|
manager.locked_hosts)
|
|
|
def testLock_WithFullyOverlappingHosts(self):
|
"""Tests host locking, all hosts in self.locked_hosts."""
|
hosts = [self.HOST1, self.HOST2]
|
self.manager.locked_hosts = set(hosts)
|
self.manager.lock(hosts)
|
self.assertEquals(set(hosts), self.manager.locked_hosts)
|
|
|
def testUnlock_WithNonOverlappingHosts(self):
|
"""Tests host unlocking, all hosts not in self.locked_hosts."""
|
hosts = [self.HOST2]
|
self.manager.locked_hosts = set([self.HOST1])
|
self.manager.unlock(hosts)
|
self.assertEquals(set([self.HOST1]), self.manager.locked_hosts)
|
|
|
def testUnlock_WithPartialOverlappingHosts(self):
|
"""Tests host locking, some hosts not in self.locked_hosts."""
|
hosts = [self.HOST1, self.HOST2]
|
manager = self.MockHostLockManager(self.afe)
|
manager.locked_hosts = set([self.HOST1, self.HOST3])
|
manager.unlock(hosts)
|
self.assertEquals(set([self.HOST3]), manager.locked_hosts)
|
|
|
def testUnlock_WithFullyOverlappingHosts(self):
|
"""Tests host locking, all hosts in self.locked_hosts."""
|
hosts = [self.HOST1, self.HOST2]
|
manager = self.MockHostLockManager(self.afe)
|
manager.locked_hosts = set([self.HOST1, self.HOST2, self.HOST3])
|
manager.unlock(hosts)
|
self.assertEquals(set([self.HOST3]), manager.locked_hosts)
|
|
|
def testHostModifier_WithHostsToLock(self):
|
"""Test host locking."""
|
hosts = set([self.HOST1])
|
self.manager.locked_hosts = set([self.HOST2])
|
self.mox.StubOutWithMock(self.manager, '_check_host')
|
self.manager._check_host(self.HOST1,
|
self.manager.LOCK).AndReturn(self.HOST1)
|
self.afe.run('modify_hosts',
|
host_filter_data={'hostname__in': [self.HOST1]},
|
update_data={'locked': True, 'lock_reason': 'Test'})
|
self.mox.ReplayAll()
|
self.manager._host_modifier(hosts, self.manager.LOCK,
|
lock_reason='Test')
|
self.assertEquals(set([self.HOST1, self.HOST2]),
|
self.manager.locked_hosts)
|
|
|
def testHostModifier_WithHostsToUnlock(self):
|
"""Test host unlocking."""
|
hosts = set([self.HOST1])
|
self.manager.locked_hosts = set([self.HOST1, self.HOST2])
|
self.mox.StubOutWithMock(self.manager, '_check_host')
|
self.manager._check_host(self.HOST1,
|
self.manager.UNLOCK).AndReturn(self.HOST1)
|
self.afe.run('modify_hosts',
|
host_filter_data={'hostname__in': [self.HOST1]},
|
update_data={'locked': False})
|
self.mox.ReplayAll()
|
self.manager._host_modifier(hosts, self.manager.UNLOCK)
|
self.assertEquals(set([self.HOST2]), self.manager.locked_hosts)
|
|
|
def testHostModifier_WithoutLockReason(self):
|
"""Test host locking without providing a lock reason."""
|
hosts = set([self.HOST1])
|
self.manager.locked_hosts = set([self.HOST2])
|
self.mox.StubOutWithMock(self.manager, '_check_host')
|
self.manager._check_host(self.HOST1,
|
self.manager.LOCK).AndReturn(self.HOST1)
|
self.afe.run('modify_hosts',
|
host_filter_data={'hostname__in': [self.HOST1]},
|
update_data={'locked': True,
|
'lock_reason': None})
|
self.mox.ReplayAll()
|
self.manager._host_modifier(hosts, self.manager.LOCK)
|
self.assertEquals(set([self.HOST2]), self.manager.locked_hosts)
|
|
|
if __name__ == '__main__':
|
unittest.main()
|
