python-ironicclient/ironicclient/tests/unit/v1/test_allocation.py

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import copy

import mock
import testtools

from ironicclient import exc
from ironicclient.tests.unit import utils
import ironicclient.v1.allocation

ALLOCATION = {'uuid': '11111111-2222-3333-4444-555555555555',
              'name': 'Allocation-name',
              'owner': None,
              'state': 'active',
              'node_uuid': '66666666-7777-8888-9999-000000000000',
              'last_error': None,
              'resource_class': 'baremetal',
              'traits': [],
              'candidate_nodes': [],
              'extra': {}}

ALLOCATION2 = {'uuid': '55555555-4444-3333-2222-111111111111',
               'name': 'Allocation2-name',
               'owner': 'fake-owner',
               'state': 'allocating',
               'node_uuid': None,
               'last_error': None,
               'resource_class': 'baremetal',
               'traits': [],
               'candidate_nodes': [],
               'extra': {}}

CREATE_ALLOCATION = copy.deepcopy(ALLOCATION)
for field in ('state', 'node_uuid', 'last_error'):
    del CREATE_ALLOCATION[field]

fake_responses = {
    '/v1/allocations':
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION, ALLOCATION2]},
        ),
        'POST': (
            {},
            CREATE_ALLOCATION,
        ),
    },
    '/v1/allocations/%s' % ALLOCATION['uuid']:
    {
        'GET': (
            {},
            ALLOCATION,
        ),
        'DELETE': (
            {},
            None,
        ),
    },
    '/v1/allocations/?node=%s' % ALLOCATION['node_uuid']:
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION]},
        ),
    },
    '/v1/allocations/?owner=%s' % ALLOCATION2['owner']:
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION2]},
        ),
    },
}

fake_responses_pagination = {
    '/v1/allocations':
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION],
             "next": "http://127.0.0.1:6385/v1/allocations/?limit=1"}
        ),
    },
    '/v1/allocations/?limit=1':
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION2]}
        ),
    },
    '/v1/allocations/?marker=%s' % ALLOCATION['uuid']:
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION2]}
        ),
    },
}

fake_responses_sorting = {
    '/v1/allocations/?sort_key=updated_at':
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION2, ALLOCATION]}
        ),
    },
    '/v1/allocations/?sort_dir=desc':
    {
        'GET': (
            {},
            {"allocations": [ALLOCATION2, ALLOCATION]}
        ),
    },
}


class AllocationManagerTest(testtools.TestCase):

    def setUp(self):
        super(AllocationManagerTest, self).setUp()
        self.api = utils.FakeAPI(fake_responses)
        self.mgr = ironicclient.v1.allocation.AllocationManager(self.api)

    def test_allocations_list(self):
        allocations = self.mgr.list()
        expect = [
            ('GET', '/v1/allocations', {}, None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(2, len(allocations))

        expected_resp = ({}, {"allocations": [ALLOCATION, ALLOCATION2]},)
        self.assertEqual(expected_resp,
                         self.api.responses['/v1/allocations']['GET'])

    def test_allocations_list_by_node(self):
        allocations = self.mgr.list(node=ALLOCATION['node_uuid'])
        expect = [
            ('GET', '/v1/allocations/?node=%s' % ALLOCATION['node_uuid'], {},
             None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(1, len(allocations))

        expected_resp = ({}, {"allocations": [ALLOCATION, ALLOCATION2]},)
        self.assertEqual(expected_resp,
                         self.api.responses['/v1/allocations']['GET'])

    def test_allocations_list_by_owner(self):
        allocations = self.mgr.list(owner=ALLOCATION2['owner'])
        expect = [
            ('GET', '/v1/allocations/?owner=%s' % ALLOCATION2['owner'], {},
             None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(1, len(allocations))

        expected_resp = ({}, {"allocations": [ALLOCATION, ALLOCATION2]},)
        self.assertEqual(expected_resp,
                         self.api.responses['/v1/allocations']['GET'])

    def test_allocations_show(self):
        allocation = self.mgr.get(ALLOCATION['uuid'])
        expect = [
            ('GET', '/v1/allocations/%s' % ALLOCATION['uuid'], {}, None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(ALLOCATION['uuid'], allocation.uuid)
        self.assertEqual(ALLOCATION['name'], allocation.name)
        self.assertEqual(ALLOCATION['owner'], allocation.owner)
        self.assertEqual(ALLOCATION['node_uuid'], allocation.node_uuid)
        self.assertEqual(ALLOCATION['state'], allocation.state)
        self.assertEqual(ALLOCATION['resource_class'],
                         allocation.resource_class)

        expected_resp = ({}, ALLOCATION,)
        self.assertEqual(
            expected_resp,
            self.api.responses['/v1/allocations/%s'
                               % ALLOCATION['uuid']]['GET'])

    def test_create(self):
        allocation = self.mgr.create(**CREATE_ALLOCATION)
        expect = [
            ('POST', '/v1/allocations', {}, CREATE_ALLOCATION),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertTrue(allocation)

        self.assertIn(
            ALLOCATION,
            self.api.responses['/v1/allocations']['GET'][1]['allocations'])

    def test_delete(self):
        allocation = self.mgr.delete(allocation_id=ALLOCATION['uuid'])
        expect = [
            ('DELETE', '/v1/allocations/%s' % ALLOCATION['uuid'], {}, None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertIsNone(allocation)

        expected_resp = ({}, ALLOCATION,)
        self.assertEqual(
            expected_resp,
            self.api.responses['/v1/allocations/%s'
                               % ALLOCATION['uuid']]['GET'])


class AllocationManagerPaginationTest(testtools.TestCase):

    def setUp(self):
        super(AllocationManagerPaginationTest, self).setUp()
        self.api = utils.FakeAPI(fake_responses_pagination)
        self.mgr = ironicclient.v1.allocation.AllocationManager(self.api)

    def test_allocations_list_limit(self):
        allocations = self.mgr.list(limit=1)
        expect = [
            ('GET', '/v1/allocations/?limit=1', {}, None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(1, len(allocations))

        expected_resp = (
            {}, {"next": "http://127.0.0.1:6385/v1/allocations/?limit=1",
                 "allocations": [ALLOCATION]},)
        self.assertEqual(expected_resp,
                         self.api.responses['/v1/allocations']['GET'])

    def test_allocations_list_marker(self):
        allocations = self.mgr.list(marker=ALLOCATION['uuid'])
        expect = [
            ('GET', '/v1/allocations/?marker=%s' % ALLOCATION['uuid'],
             {}, None),
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(1, len(allocations))

        expected_resp = (
            {}, {"next": "http://127.0.0.1:6385/v1/allocations/?limit=1",
                 "allocations": [ALLOCATION]},)
        self.assertEqual(expected_resp,
                         self.api.responses['/v1/allocations']['GET'])

    def test_allocations_list_pagination_no_limit(self):
        allocations = self.mgr.list(limit=0)
        expect = [
            ('GET', '/v1/allocations', {}, None),
            ('GET', '/v1/allocations/?limit=1', {}, None)
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(2, len(allocations))

        expected_resp = (
            {}, {"next": "http://127.0.0.1:6385/v1/allocations/?limit=1",
                 "allocations": [ALLOCATION]},)
        self.assertEqual(expected_resp,
                         self.api.responses['/v1/allocations']['GET'])


class AllocationManagerSortingTest(testtools.TestCase):

    def setUp(self):
        super(AllocationManagerSortingTest, self).setUp()
        self.api = utils.FakeAPI(fake_responses_sorting)
        self.mgr = ironicclient.v1.allocation.AllocationManager(self.api)

    def test_allocations_list_sort_key(self):
        allocations = self.mgr.list(sort_key='updated_at')
        expect = [
            ('GET', '/v1/allocations/?sort_key=updated_at', {}, None)
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(2, len(allocations))

        expected_resp = ({}, {"allocations": [ALLOCATION2, ALLOCATION]},)
        self.assertEqual(
            expected_resp,
            self.api.responses['/v1/allocations/?sort_key=updated_at']['GET'])

    def test_allocations_list_sort_dir(self):
        allocations = self.mgr.list(sort_dir='desc')
        expect = [
            ('GET', '/v1/allocations/?sort_dir=desc', {}, None)
        ]
        self.assertEqual(expect, self.api.calls)
        self.assertEqual(2, len(allocations))

        expected_resp = ({}, {"allocations": [ALLOCATION2, ALLOCATION]},)
        self.assertEqual(
            expected_resp,
            self.api.responses['/v1/allocations/?sort_dir=desc']['GET'])


@mock.patch('time.sleep', autospec=True)
@mock.patch('ironicclient.v1.allocation.AllocationManager.get', autospec=True)
class AllocationWaitTest(testtools.TestCase):

    def setUp(self):
        super(AllocationWaitTest, self).setUp()
        self.mgr = ironicclient.v1.allocation.AllocationManager(mock.Mock())

    def _fake_allocation(self, state, error=None):
        return mock.Mock(state=state, last_error=error)

    def test_success(self, mock_get, mock_sleep):
        allocations = [
            self._fake_allocation('allocating'),
            self._fake_allocation('allocating'),
            self._fake_allocation('active'),
        ]
        mock_get.side_effect = allocations

        result = self.mgr.wait('alloc1')
        self.assertIs(result, allocations[2])
        self.assertEqual(3, mock_get.call_count)
        self.assertEqual(2, mock_sleep.call_count)
        mock_get.assert_called_with(self.mgr, 'alloc1')

    def test_error(self, mock_get, mock_sleep):
        allocations = [
            self._fake_allocation('allocating'),
            self._fake_allocation('error'),
        ]
        mock_get.side_effect = allocations

        self.assertRaises(exc.StateTransitionFailed,
                          self.mgr.wait, 'alloc1')

        self.assertEqual(2, mock_get.call_count)
        self.assertEqual(1, mock_sleep.call_count)
        mock_get.assert_called_with(self.mgr, 'alloc1')

    def test_timeout(self, mock_get, mock_sleep):
        mock_get.return_value = self._fake_allocation('allocating')

        self.assertRaises(exc.StateTransitionTimeout,
                          self.mgr.wait, 'alloc1', timeout=0.001)

        mock_get.assert_called_with(self.mgr, 'alloc1')