server/cros/chaos_ap_configurators/ap_batch_locker_unittest.py

#!/usr/bin/python
#
# Copyright (c) 2013 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/chaos_ap_configurators/ap_batch_locker.py."""

import mox

from autotest_lib.server.cros.chaos_ap_configurators import ap_batch_locker
from autotest_lib.server.cros.chaos_ap_configurators import \
    ap_configurator_factory


class ConstructApLockersTest(mox.MoxTestBase):
    """Unit tests for ap_batch_locker.construct_ap_lockers()."""


    def setUp(self):
        """Initialize."""
        super(ConstructApLockersTest, self).setUp()
        self.mox.StubOutWithMock(ap_configurator_factory,
                                 'APConfiguratorFactory')
        self.mox.StubOutWithMock(ap_batch_locker, 'ApLocker')
        self.mock_factory = self.mox.CreateMockAnything()
        self.retries = 1


    def testConstructApLockers_withEmptyApSpec(self):
        """Tests an empty list is returned by default."""
        ap_configurator_factory.APConfiguratorFactory().AndReturn(
                self.mock_factory)
        self.mock_factory.get_ap_configurators({}).AndReturn([])
        self.mox.ReplayAll()
        actual = ap_batch_locker.construct_ap_lockers({}, self.retries)
        self.assertEquals([], actual)


    def testConstructApLockers_withValidApSpec(self):
        """Tests proper mocks are invoked with a valid ap_spec."""
        ap_spec = {'security': ''}
        mock_ap1 = 'mock_ap1'
        mock_ap2 = 'mock_ap2'
        mock_ap_list = [mock_ap1, mock_ap2]
        ap_configurator_factory.APConfiguratorFactory().AndReturn(
                self.mock_factory)
        self.mock_factory.get_ap_configurators(ap_spec).AndReturn(mock_ap_list)
        ap_batch_locker.ApLocker(mock_ap1, self.retries).AndReturn(mock_ap1)
        ap_batch_locker.ApLocker(mock_ap2, self.retries).AndReturn(mock_ap2)
        self.mox.ReplayAll()
        actual = ap_batch_locker.construct_ap_lockers(ap_spec, self.retries)
        self.assertEquals(mock_ap_list, actual)


# host name of a mock APConfigurator.
MOCK_AP = 'mock_ap'


class MockApConfigurator(object):
    """Mock of an APConfigurator object.

    @attribute host_name: a string, ap host name.
    """


    def __init__(self):
        """Initialize.

        @attribute host_name: a string, ap host name.
        """
        self.host_name = MOCK_AP


class ApBatchLockerLockApInAfeTest(mox.MoxTestBase):
    """Unit tests for ap_batch_locker.ApBatchLocker.lock_ap_in_afe()."""


    class MockApBatchLocker(ap_batch_locker.ApBatchLocker):
        """Mock of ap_batch_locker.ApBatchLocker().

        @attribute aps_to_lock: a list of ApLocker objects.
        """

        def __init__(self, mox_obj):
            """Initialize."""
            self.aps_to_lock = []
            self.manager = mox_obj.CreateMockAnything()


    def setUp(self):
        """Initialize."""
        super(ApBatchLockerLockApInAfeTest, self).setUp()
        self.mock_batch_locker = self.MockApBatchLocker(self.mox)
        self.retries = 2
        self.mock_ap_locker = None


    def _set_up_mocks(self, retries):
        """Sets up mocks.

        @param retries: an integer.
        """
        self.mock_ap_locker = ap_batch_locker.ApLocker(
                MockApConfigurator(), retries)
        self.mock_batch_locker.aps_to_lock = [self.mock_ap_locker]


    def testLockApInAfe_WithLockableAp(self):
        """Tests AP can be locked and removed from ap_list."""
        self._set_up_mocks(self.retries)
        self.mock_batch_locker.manager.lock([MOCK_AP]).AndReturn(True)
        self.mox.ReplayAll()
        actual = self.mock_batch_locker.lock_ap_in_afe(self.mock_ap_locker)
        self.assertEquals(True, actual)
        self.assertEquals(False, self.mock_ap_locker.to_be_locked)


    def testLockApInAfe_WithUnlockableApAndRetriesRemaining(self):
        """Tests retries counter (of an unlockable AP) is properly deducted."""
        self._set_up_mocks(self.retries)
        expected_retries = self.retries - 1
        self.mock_batch_locker.manager.lock([MOCK_AP]).AndReturn(False)
        self.mox.ReplayAll()
        actual_ret = self.mock_batch_locker.lock_ap_in_afe(self.mock_ap_locker)
        self.assertEquals(False, actual_ret)
        self.assertEquals(expected_retries, self.mock_ap_locker.retries)


    def testLockApInAfe_WithUnlockableApAndNoRetriesRemaining(self):
        """Tests removal of an unlockable AP w/ no retries remaining."""
        self.retries = 1
        expected_retries = 0
        self._set_up_mocks(self.retries)
        self.mock_batch_locker.manager.lock([MOCK_AP]).AndReturn(False)
        self.mox.ReplayAll()
        actual_ret = self.mock_batch_locker.lock_ap_in_afe(self.mock_ap_locker)
        self.assertEquals(False, actual_ret)
        self.assertEquals(expected_retries, self.mock_ap_locker.retries)



class ApBatchLockerGetApBatchTest(mox.MoxTestBase):
    """Unit tests for ap_batch_locker.ApBatchLocker.get_ap_batch()."""


    class MockApBatchLocker(ap_batch_locker.ApBatchLocker):
        """Mock of ap_batch_locker.ApBatchLocker().

        @attribute aps_to_lock: a list of ApLocker objects.
        """

        def __init__(self):
            """Initialize."""
            self.aps_to_lock = []


    def setUp(self):
        """Initialize."""
        super(ApBatchLockerGetApBatchTest, self).setUp()
        self.mox.StubOutWithMock(ap_batch_locker.ApBatchLocker,
                                 'lock_ap_in_afe')
        self.mock_batch_locker = self.MockApBatchLocker()
        self.batch_size = 2
        self.retries = 2


    def testGetApBatch_WithEmptyApList(self):
        """Tests an empty list is returned by default."""
        actual = self.mock_batch_locker.get_ap_batch(self.batch_size)
        self.assertEquals([], actual)


    def testGetApBatch_WithListOfOneApAndBatchSizeOne(self):
        """Tests batch_size of 1 returns inside while loop."""
        self.batch_size = 1
        mock_ap = MockApConfigurator()
        mock_ap_locker = ap_batch_locker.ApLocker(mock_ap, self.retries)
        self.mock_batch_locker.aps_to_lock = [mock_ap_locker]
        self.mock_batch_locker.lock_ap_in_afe(mock_ap_locker).AndReturn(True)
        self.mox.ReplayAll()
        actual = self.mock_batch_locker.get_ap_batch(self.batch_size)
        self.assertEquals([mock_ap], actual)