f353253122
This patch fixed a 'volume qos create' display mistake in argument of 'specs'[1]. For command such as: $ openstack volume qos create hello [1]https://bugs.launchpad.net/python-openstackclient/+bug/1656767 Closes-bug:#1656767 Change-Id: Ia9fce833d318d9b52b97c12cfb89e2d3c5465fbe
456 lines
13 KiB
Python
456 lines
13 KiB
Python
# Copyright 2015 iWeb Technologies Inc.
|
|
#
|
|
# 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 mock
|
|
from mock import call
|
|
|
|
from osc_lib import exceptions
|
|
from osc_lib import utils
|
|
|
|
from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
|
|
from openstackclient.volume.v2 import qos_specs
|
|
|
|
|
|
class TestQos(volume_fakes.TestVolume):
|
|
|
|
def setUp(self):
|
|
super(TestQos, self).setUp()
|
|
|
|
self.qos_mock = self.app.client_manager.volume.qos_specs
|
|
self.qos_mock.reset_mock()
|
|
|
|
self.types_mock = self.app.client_manager.volume.volume_types
|
|
self.types_mock.reset_mock()
|
|
|
|
|
|
class TestQosAssociate(TestQos):
|
|
|
|
volume_type = volume_fakes.FakeType.create_one_type()
|
|
qos_spec = volume_fakes.FakeQos.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super(TestQosAssociate, self).setUp()
|
|
|
|
self.qos_mock.get.return_value = self.qos_spec
|
|
self.types_mock.get.return_value = self.volume_type
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.AssociateQos(self.app, None)
|
|
|
|
def test_qos_associate(self):
|
|
arglist = [
|
|
self.qos_spec.id,
|
|
self.volume_type.id
|
|
]
|
|
verifylist = [
|
|
('qos_spec', self.qos_spec.id),
|
|
('volume_type', self.volume_type.id)
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.associate.assert_called_with(
|
|
self.qos_spec.id,
|
|
self.volume_type.id
|
|
)
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestQosCreate(TestQos):
|
|
|
|
columns = (
|
|
'consumer',
|
|
'id',
|
|
'name',
|
|
'properties'
|
|
)
|
|
|
|
def setUp(self):
|
|
super(TestQosCreate, self).setUp()
|
|
|
|
self.new_qos_spec = volume_fakes.FakeQos.create_one_qos()
|
|
self.qos_mock.create.return_value = self.new_qos_spec
|
|
|
|
self.data = (
|
|
self.new_qos_spec.consumer,
|
|
self.new_qos_spec.id,
|
|
self.new_qos_spec.name,
|
|
utils.format_dict(self.new_qos_spec.specs)
|
|
)
|
|
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.CreateQos(self.app, None)
|
|
|
|
def test_qos_create_without_properties(self):
|
|
arglist = [
|
|
self.new_qos_spec.name,
|
|
]
|
|
verifylist = [
|
|
('name', self.new_qos_spec.name),
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.create.assert_called_with(
|
|
self.new_qos_spec.name,
|
|
{'consumer': 'both'}
|
|
)
|
|
|
|
self.assertEqual(self.columns, columns)
|
|
self.assertEqual(self.data, data)
|
|
|
|
def test_qos_create_with_consumer(self):
|
|
arglist = [
|
|
'--consumer', self.new_qos_spec.consumer,
|
|
self.new_qos_spec.name,
|
|
]
|
|
verifylist = [
|
|
('consumer', self.new_qos_spec.consumer),
|
|
('name', self.new_qos_spec.name),
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.create.assert_called_with(
|
|
self.new_qos_spec.name,
|
|
{'consumer': self.new_qos_spec.consumer}
|
|
)
|
|
|
|
self.assertEqual(self.columns, columns)
|
|
self.assertEqual(self.data, data)
|
|
|
|
def test_qos_create_with_properties(self):
|
|
arglist = [
|
|
'--consumer', self.new_qos_spec.consumer,
|
|
'--property', 'foo=bar',
|
|
'--property', 'iops=9001',
|
|
self.new_qos_spec.name,
|
|
]
|
|
verifylist = [
|
|
('consumer', self.new_qos_spec.consumer),
|
|
('property', self.new_qos_spec.specs),
|
|
('name', self.new_qos_spec.name),
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.create.assert_called_with(
|
|
self.new_qos_spec.name,
|
|
{'consumer': self.new_qos_spec.consumer,
|
|
'foo': 'bar',
|
|
'iops': '9001'}
|
|
)
|
|
|
|
self.assertEqual(self.columns, columns)
|
|
self.assertEqual(self.data, data)
|
|
|
|
|
|
class TestQosDelete(TestQos):
|
|
|
|
qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
|
|
|
|
def setUp(self):
|
|
super(TestQosDelete, self).setUp()
|
|
|
|
self.qos_mock.get = (
|
|
volume_fakes.FakeQos.get_qoses(self.qos_specs))
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.DeleteQos(self.app, None)
|
|
|
|
def test_qos_delete(self):
|
|
arglist = [
|
|
self.qos_specs[0].id
|
|
]
|
|
verifylist = [
|
|
('qos_specs', [self.qos_specs[0].id])
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.delete.assert_called_with(
|
|
self.qos_specs[0].id, False)
|
|
self.assertIsNone(result)
|
|
|
|
def test_qos_delete_with_force(self):
|
|
arglist = [
|
|
'--force',
|
|
self.qos_specs[0].id
|
|
]
|
|
verifylist = [
|
|
('force', True),
|
|
('qos_specs', [self.qos_specs[0].id])
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.delete.assert_called_with(
|
|
self.qos_specs[0].id, True)
|
|
self.assertIsNone(result)
|
|
|
|
def test_delete_multiple_qoses(self):
|
|
arglist = []
|
|
for q in self.qos_specs:
|
|
arglist.append(q.id)
|
|
verifylist = [
|
|
('qos_specs', arglist),
|
|
]
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
calls = []
|
|
for q in self.qos_specs:
|
|
calls.append(call(q.id, False))
|
|
self.qos_mock.delete.assert_has_calls(calls)
|
|
self.assertIsNone(result)
|
|
|
|
def test_delete_multiple_qoses_with_exception(self):
|
|
arglist = [
|
|
self.qos_specs[0].id,
|
|
'unexist_qos',
|
|
]
|
|
verifylist = [
|
|
('qos_specs', arglist),
|
|
]
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
find_mock_result = [self.qos_specs[0], exceptions.CommandError]
|
|
with mock.patch.object(utils, 'find_resource',
|
|
side_effect=find_mock_result) as find_mock:
|
|
try:
|
|
self.cmd.take_action(parsed_args)
|
|
self.fail('CommandError should be raised.')
|
|
except exceptions.CommandError as e:
|
|
self.assertEqual(
|
|
'1 of 2 QoS specifications failed to delete.', str(e))
|
|
|
|
find_mock.assert_any_call(self.qos_mock, self.qos_specs[0].id)
|
|
find_mock.assert_any_call(self.qos_mock, 'unexist_qos')
|
|
|
|
self.assertEqual(2, find_mock.call_count)
|
|
self.qos_mock.delete.assert_called_once_with(
|
|
self.qos_specs[0].id, False
|
|
)
|
|
|
|
|
|
class TestQosDisassociate(TestQos):
|
|
|
|
volume_type = volume_fakes.FakeType.create_one_type()
|
|
qos_spec = volume_fakes.FakeQos.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super(TestQosDisassociate, self).setUp()
|
|
|
|
self.qos_mock.get.return_value = self.qos_spec
|
|
self.types_mock.get.return_value = self.volume_type
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.DisassociateQos(self.app, None)
|
|
|
|
def test_qos_disassociate_with_volume_type(self):
|
|
arglist = [
|
|
'--volume-type', self.volume_type.id,
|
|
self.qos_spec.id,
|
|
]
|
|
verifylist = [
|
|
('volume_type', self.volume_type.id),
|
|
('qos_spec', self.qos_spec.id),
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.disassociate.assert_called_with(
|
|
self.qos_spec.id,
|
|
self.volume_type.id
|
|
)
|
|
self.assertIsNone(result)
|
|
|
|
def test_qos_disassociate_with_all_volume_types(self):
|
|
arglist = [
|
|
'--all',
|
|
self.qos_spec.id,
|
|
]
|
|
verifylist = [
|
|
('qos_spec', self.qos_spec.id)
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.disassociate_all.assert_called_with(self.qos_spec.id)
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestQosList(TestQos):
|
|
|
|
qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
|
|
qos_association = volume_fakes.FakeQos.create_one_qos_association()
|
|
|
|
columns = (
|
|
'ID',
|
|
'Name',
|
|
'Consumer',
|
|
'Associations',
|
|
'Properties',
|
|
)
|
|
data = []
|
|
for q in qos_specs:
|
|
data.append((
|
|
q.id,
|
|
q.name,
|
|
q.consumer,
|
|
qos_association.name,
|
|
utils.format_dict(q.specs),
|
|
))
|
|
|
|
def setUp(self):
|
|
super(TestQosList, self).setUp()
|
|
|
|
self.qos_mock.list.return_value = self.qos_specs
|
|
self.qos_mock.get_associations.return_value = [self.qos_association]
|
|
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.ListQos(self.app, None)
|
|
|
|
def test_qos_list(self):
|
|
arglist = []
|
|
verifylist = []
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
self.qos_mock.list.assert_called_with()
|
|
|
|
self.assertEqual(self.columns, columns)
|
|
self.assertEqual(self.data, list(data))
|
|
|
|
|
|
class TestQosSet(TestQos):
|
|
|
|
qos_spec = volume_fakes.FakeQos.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super(TestQosSet, self).setUp()
|
|
|
|
self.qos_mock.get.return_value = self.qos_spec
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.SetQos(self.app, None)
|
|
|
|
def test_qos_set_with_properties_with_id(self):
|
|
arglist = [
|
|
'--property', 'foo=bar',
|
|
'--property', 'iops=9001',
|
|
self.qos_spec.id,
|
|
]
|
|
verifylist = [
|
|
('property', self.qos_spec.specs),
|
|
('qos_spec', self.qos_spec.id),
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.set_keys.assert_called_with(
|
|
self.qos_spec.id,
|
|
self.qos_spec.specs
|
|
)
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestQosShow(TestQos):
|
|
|
|
qos_spec = volume_fakes.FakeQos.create_one_qos()
|
|
qos_association = volume_fakes.FakeQos.create_one_qos_association()
|
|
|
|
columns = (
|
|
'associations',
|
|
'consumer',
|
|
'id',
|
|
'name',
|
|
'properties'
|
|
)
|
|
data = (
|
|
qos_association.name,
|
|
qos_spec.consumer,
|
|
qos_spec.id,
|
|
qos_spec.name,
|
|
utils.format_dict(qos_spec.specs),
|
|
)
|
|
|
|
def setUp(self):
|
|
super(TestQosShow, self).setUp()
|
|
|
|
self.qos_mock.get.return_value = self.qos_spec
|
|
self.qos_mock.get_associations.return_value = [self.qos_association]
|
|
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.ShowQos(self.app, None)
|
|
|
|
def test_qos_show(self):
|
|
arglist = [
|
|
self.qos_spec.id
|
|
]
|
|
verifylist = [
|
|
('qos_spec', self.qos_spec.id)
|
|
]
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
self.qos_mock.get.assert_called_with(
|
|
self.qos_spec.id
|
|
)
|
|
|
|
self.assertEqual(self.columns, columns)
|
|
self.assertEqual(self.data, tuple(data))
|
|
|
|
|
|
class TestQosUnset(TestQos):
|
|
|
|
qos_spec = volume_fakes.FakeQos.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super(TestQosUnset, self).setUp()
|
|
|
|
self.qos_mock.get.return_value = self.qos_spec
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.UnsetQos(self.app, None)
|
|
|
|
def test_qos_unset_with_properties(self):
|
|
arglist = [
|
|
'--property', 'iops',
|
|
'--property', 'foo',
|
|
self.qos_spec.id,
|
|
]
|
|
verifylist = [
|
|
('property', ['iops', 'foo']),
|
|
('qos_spec', self.qos_spec.id),
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
result = self.cmd.take_action(parsed_args)
|
|
|
|
self.qos_mock.unset_keys.assert_called_with(
|
|
self.qos_spec.id,
|
|
['iops', 'foo']
|
|
)
|
|
self.assertIsNone(result)
|