# Copyright 2016 Mirantis, Inc.
# All Rights Reserved.
#
# 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.common.apiclient import exceptions
from ironicclient.tests.unit import utils
import ironicclient.v1.portgroup
import ironicclient.v1.portgroup_shell as pg_shell
PORTGROUP = {'uuid': '11111111-2222-3333-4444-555555555555',
'name': 'Portgroup-name',
'node_uuid': '66666666-7777-8888-9999-000000000000',
'address': 'AA:BB:CC:DD:EE:FF',
'extra': {}}
PORTGROUP2 = {'uuid': '55555555-4444-3333-2222-111111111111',
'name': 'Portgroup2-name',
'node_uuid': '66666666-7777-8888-9999-000000000000',
'address': 'AA:AA:AA:BB:BB:BB',
'extra': {}}
NODE1 = {'uuid': '66666666-7777-8888-9999-000000000000',
'chassis_uuid': 'aaaaaaaa-1111-bbbb-2222-cccccccccccc',
'maintenance': False,
'provision_state': 'available',
'driver': 'fake',
'driver_info': {'user': 'foo', 'password': 'bar'},
'properties': {'num_cpu': 4},
'name': 'fake-node-1',
'resource_class': 'foo',
'extra': {}}
PORT = {'uuid': '11111111-2222-3333-4444-555555555555',
'portgroup_uuid': '11111111-2222-3333-4444-555555555555',
'node_uuid': '66666666-7777-8888-9999-000000000000',
'address': 'AA:AA:AA:AA:AA:AA',
'extra': {}}
CREATE_PORTGROUP = copy.deepcopy(PORTGROUP)
del CREATE_PORTGROUP['uuid']
UPDATED_PORTGROUP = copy.deepcopy(PORTGROUP)
NEW_ADDR = 'AA:AA:AA:AA:AA:AA'
UPDATED_PORTGROUP['address'] = NEW_ADDR
fake_responses = {
'/v1/portgroups':
{
'GET': (
{},
{"portgroups": [PORTGROUP, PORTGROUP2]},
),
'POST': (
{},
CREATE_PORTGROUP,
),
},
'/v1/portgroups/detail':
{
'GET': (
{},
{"portgroups": [PORTGROUP, PORTGROUP2]},
),
},
'/v1/portgroups/%s' % PORTGROUP['uuid']:
{
'GET': (
{},
PORTGROUP,
),
'DELETE': (
{},
None,
),
'PATCH': (
{},
UPDATED_PORTGROUP,
),
},
'/v1/portgroups/detail?address=%s' % PORTGROUP['address']:
{
'GET': (
{},
{"portgroups": [PORTGROUP]},
),
},
'/v1/portgroups/?address=%s' % PORTGROUP['address']:
{
'GET': (
{},
{"portgroups": [PORTGROUP]},
),
},
'/v1/portgroups/%s/ports' % PORTGROUP['name']:
{
'GET': (
{},
{"ports": [PORT]},
),
},
'/v1/portgroups/%s/ports' % PORTGROUP['uuid']:
{
'GET': (
{},
{"ports": [PORT]},
),
},
}
fake_responses_pagination = {
'/v1/portgroups':
{
'GET': (
{},
{"portgroups": [PORTGROUP],
"next": "http://127.0.0.1:6385/v1/portgroups/?limit=1"}
),
},
'/v1/portgroups/?limit=1':
{
'GET': (
{},
{"portgroups": [PORTGROUP2]}
),
},
'/v1/portgroups/?marker=%s' % PORTGROUP['uuid']:
{
'GET': (
{},
{"portgroups": [PORTGROUP2]}
),
},
}
fake_responses_sorting = {
'/v1/portgroups/?sort_key=updated_at':
{
'GET': (
{},
{"portgroups": [PORTGROUP2, PORTGROUP]}
),
},
'/v1/portgroups/?sort_dir=desc':
{
'GET': (
{},
{"portgroups": [PORTGROUP2, PORTGROUP]}
),
},
}
[docs]class PortgroupManagerTest(testtools.TestCase):
[docs] def setUp(self):
super(PortgroupManagerTest, self).setUp()
self.api = utils.FakeAPI(fake_responses)
self.mgr = ironicclient.v1.portgroup.PortgroupManager(self.api)
[docs] def test_portgroups_list(self):
portgroups = self.mgr.list()
expect = [
('GET', '/v1/portgroups', {}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(2, len(portgroups))
expected_resp = ({}, {"portgroups": [PORTGROUP, PORTGROUP2]},)
self.assertEqual(expected_resp,
self.api.responses['/v1/portgroups']['GET'])
[docs] def test_portgroups_list_by_address(self):
portgroups = self.mgr.list(address=PORTGROUP['address'])
expect = [
('GET', '/v1/portgroups/?address=%s' % PORTGROUP['address'], {},
None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(1, len(portgroups))
expected_resp = ({}, {"portgroups": [PORTGROUP, PORTGROUP2]},)
self.assertEqual(expected_resp,
self.api.responses['/v1/portgroups']['GET'])
[docs] def test_portgroups_list_by_address_detail(self):
portgroups = self.mgr.list(address=PORTGROUP['address'], detail=True)
expect = [
('GET', '/v1/portgroups/detail?address=%s' % PORTGROUP['address'],
{}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(1, len(portgroups))
self.assertIn(
PORTGROUP,
self.api.responses['/v1/portgroups']['GET'][1]['portgroups'])
[docs] def test_portgroups_list_detail(self):
portgroups = self.mgr.list(detail=True)
expect = [
('GET', '/v1/portgroups/detail', {}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(2, len(portgroups))
expected_resp = ({}, {"portgroups": [PORTGROUP, PORTGROUP2]},)
self.assertEqual(expected_resp,
self.api.responses['/v1/portgroups']['GET'])
[docs] def test_portgroups_show(self):
portgroup = self.mgr.get(PORTGROUP['uuid'])
expect = [
('GET', '/v1/portgroups/%s' % PORTGROUP['uuid'], {}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(PORTGROUP['uuid'], portgroup.uuid)
self.assertEqual(PORTGROUP['name'], portgroup.name)
self.assertEqual(PORTGROUP['node_uuid'], portgroup.node_uuid)
self.assertEqual(PORTGROUP['address'], portgroup.address)
expected_resp = ({}, PORTGROUP,)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/%s'
% PORTGROUP['uuid']]['GET'])
[docs] def test_portgroups_show_by_address(self):
portgroup = self.mgr.get_by_address(PORTGROUP['address'])
expect = [
('GET', '/v1/portgroups/detail?address=%s' % PORTGROUP['address'],
{}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(PORTGROUP['uuid'], portgroup.uuid)
self.assertEqual(PORTGROUP['name'], portgroup.name)
self.assertEqual(PORTGROUP['node_uuid'], portgroup.node_uuid)
self.assertEqual(PORTGROUP['address'], portgroup.address)
expected_resp = ({}, {"portgroups": [PORTGROUP]},)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/detail?address=%s'
% PORTGROUP['address']]['GET'])
[docs] def test_create(self):
portgroup = self.mgr.create(**CREATE_PORTGROUP)
expect = [
('POST', '/v1/portgroups', {}, CREATE_PORTGROUP),
]
self.assertEqual(expect, self.api.calls)
self.assertTrue(portgroup)
self.assertIn(
PORTGROUP,
self.api.responses['/v1/portgroups']['GET'][1]['portgroups'])
[docs] def test_delete(self):
portgroup = self.mgr.delete(portgroup_id=PORTGROUP['uuid'])
expect = [
('DELETE', '/v1/portgroups/%s' % PORTGROUP['uuid'], {}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertIsNone(portgroup)
expected_resp = ({}, PORTGROUP,)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/%s'
% PORTGROUP['uuid']]['GET'])
[docs] def test_do_portgroup_delete_multiple_with_exception(self):
client_mock = mock.MagicMock()
client_mock.portgroup.delete.side_effect = (
[exceptions.ClientException, None])
args = mock.MagicMock()
args.portgroup = ['pg_uuid1', 'pg_uuid2']
self.assertRaises(exceptions.ClientException,
pg_shell.do_portgroup_delete,
client_mock, args)
client_mock.portgroup.delete.assert_has_calls(
[mock.call('pg_uuid1'), mock.call('pg_uuid2')])
expected_resp = ({}, PORTGROUP,)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/%s'
% PORTGROUP['uuid']]['GET'])
[docs] def test_update(self):
patch = {'op': 'replace',
'value': NEW_ADDR,
'path': '/address'}
portgroup = self.mgr.update(portgroup_id=PORTGROUP['uuid'],
patch=patch)
expect = [
('PATCH', '/v1/portgroups/%s' % PORTGROUP['uuid'], {}, patch),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(NEW_ADDR, portgroup.address)
expected_resp = ({}, PORTGROUP,)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/%s'
% PORTGROUP['uuid']]['GET'])
[docs] def test_portgroup_port_list_with_uuid(self):
ports = self.mgr.list_ports(PORTGROUP['uuid'])
expect = [
('GET', '/v1/portgroups/%s/ports' % PORTGROUP['uuid'], {}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(1, len(ports))
self.assertEqual(PORT['uuid'], ports[0].uuid)
self.assertEqual(PORT['address'], ports[0].address)
expected_resp = ({}, {"ports": [PORT]},)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/%s/ports'
% PORTGROUP['uuid']]['GET'])
[docs] def test_portgroup_port_list_with_name(self):
ports = self.mgr.list_ports(PORTGROUP['name'])
expect = [
('GET', '/v1/portgroups/%s/ports' % PORTGROUP['name'], {}, None),
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(1, len(ports))
self.assertEqual(PORT['uuid'], ports[0].uuid)
self.assertEqual(PORT['address'], ports[0].address)
expected_resp = ({}, {"ports": [PORT]},)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/%s/ports'
% PORTGROUP['name']]['GET'])
[docs]class PortgroupManagerSortingTest(testtools.TestCase):
[docs] def setUp(self):
super(PortgroupManagerSortingTest, self).setUp()
self.api = utils.FakeAPI(fake_responses_sorting)
self.mgr = ironicclient.v1.portgroup.PortgroupManager(self.api)
[docs] def test_portgroups_list_sort_key(self):
portgroups = self.mgr.list(sort_key='updated_at')
expect = [
('GET', '/v1/portgroups/?sort_key=updated_at', {}, None)
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(2, len(portgroups))
expected_resp = ({}, {"portgroups": [PORTGROUP2, PORTGROUP]},)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/?sort_key=updated_at']['GET'])
[docs] def test_portgroups_list_sort_dir(self):
portgroups = self.mgr.list(sort_dir='desc')
expect = [
('GET', '/v1/portgroups/?sort_dir=desc', {}, None)
]
self.assertEqual(expect, self.api.calls)
self.assertEqual(2, len(portgroups))
expected_resp = ({}, {"portgroups": [PORTGROUP2, PORTGROUP]},)
self.assertEqual(
expected_resp,
self.api.responses['/v1/portgroups/?sort_dir=desc']['GET'])