4cabf6f7a2
This removes the need for a number of base test case subclasses. We use 'volume_client' rather than 'client' to avoid conflicts with clients for other services. Change-Id: I28d78f32142bb3a3fde0ba1780c504adc31cc77c Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
472 lines
14 KiB
Python
472 lines
14 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 copy
|
|
from unittest import mock
|
|
from unittest.mock import call
|
|
|
|
from osc_lib.cli import format_columns
|
|
from osc_lib import exceptions
|
|
from osc_lib import utils
|
|
|
|
from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes
|
|
from openstackclient.volume.v1 import qos_specs
|
|
|
|
|
|
class TestQos(volume_fakes.TestVolumev1):
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
self.qos_mock = self.volume_client.qos_specs
|
|
self.qos_mock.reset_mock()
|
|
|
|
self.types_mock = self.volume_client.volume_types
|
|
self.types_mock.reset_mock()
|
|
|
|
|
|
class TestQosAssociate(TestQos):
|
|
volume_type = volume_fakes.create_one_volume_type()
|
|
qos_spec = volume_fakes.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super().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().setUp()
|
|
self.new_qos_spec = volume_fakes.create_one_qos()
|
|
self.datalist = (
|
|
self.new_qos_spec.consumer,
|
|
self.new_qos_spec.id,
|
|
self.new_qos_spec.name,
|
|
format_columns.DictColumn(self.new_qos_spec.specs),
|
|
)
|
|
self.qos_mock.create.return_value = self.new_qos_spec
|
|
# 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.assertCountEqual(self.datalist, 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.assertCountEqual(self.datalist, 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.new_qos_spec.specs.update(
|
|
{'consumer': self.new_qos_spec.consumer}
|
|
)
|
|
self.qos_mock.create.assert_called_with(
|
|
self.new_qos_spec.name, self.new_qos_spec.specs
|
|
)
|
|
|
|
self.assertEqual(self.columns, columns)
|
|
self.assertCountEqual(self.datalist, data)
|
|
|
|
|
|
class TestQosDelete(TestQos):
|
|
qos_specs = volume_fakes.create_qoses(count=2)
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
self.qos_mock.get = volume_fakes.get_qoses(self.qos_specs)
|
|
# Get the command object to test
|
|
self.cmd = qos_specs.DeleteQos(self.app, None)
|
|
|
|
def test_qos_delete_with_id(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_name(self):
|
|
arglist = [self.qos_specs[0].name]
|
|
verifylist = [('qos_specs', [self.qos_specs[0].name])]
|
|
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.create_one_volume_type()
|
|
qos_spec = volume_fakes.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super().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.create_qoses(count=2)
|
|
qos_association = volume_fakes.create_one_qos_association()
|
|
|
|
columns = (
|
|
'ID',
|
|
'Name',
|
|
'Consumer',
|
|
'Associations',
|
|
'Properties',
|
|
)
|
|
data = []
|
|
for q in qos_specs:
|
|
data.append(
|
|
(
|
|
q.id,
|
|
q.name,
|
|
q.consumer,
|
|
format_columns.ListColumn([qos_association.name]),
|
|
format_columns.DictColumn(q.specs),
|
|
)
|
|
)
|
|
|
|
def setUp(self):
|
|
super().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.assertCountEqual(self.data, list(data))
|
|
|
|
def test_qos_list_no_association(self):
|
|
self.qos_mock.reset_mock()
|
|
self.qos_mock.get_associations.side_effect = [
|
|
[self.qos_association],
|
|
exceptions.NotFound("NotFound"),
|
|
]
|
|
|
|
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)
|
|
|
|
ex_data = copy.deepcopy(self.data)
|
|
ex_data[1] = (
|
|
self.qos_specs[1].id,
|
|
self.qos_specs[1].name,
|
|
self.qos_specs[1].consumer,
|
|
format_columns.ListColumn(None),
|
|
format_columns.DictColumn(self.qos_specs[1].specs),
|
|
)
|
|
self.assertCountEqual(ex_data, list(data))
|
|
|
|
|
|
class TestQosSet(TestQos):
|
|
qos_spec = volume_fakes.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super().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 = [
|
|
'--no-property',
|
|
'--property',
|
|
'a=b',
|
|
'--property',
|
|
'c=d',
|
|
self.qos_spec.id,
|
|
]
|
|
new_property = {"a": "b", "c": "d"}
|
|
verifylist = [
|
|
('no_property', True),
|
|
('property', new_property),
|
|
('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,
|
|
list(self.qos_spec.specs.keys()),
|
|
)
|
|
self.qos_mock.set_keys.assert_called_with(
|
|
self.qos_spec.id,
|
|
{"a": "b", "c": "d"},
|
|
)
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestQosShow(TestQos):
|
|
qos_spec = volume_fakes.create_one_qos()
|
|
qos_association = volume_fakes.create_one_qos_association()
|
|
|
|
def setUp(self):
|
|
super().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)
|
|
|
|
collist = ('associations', 'consumer', 'id', 'name', 'properties')
|
|
self.assertEqual(collist, columns)
|
|
datalist = (
|
|
format_columns.ListColumn([self.qos_association.name]),
|
|
self.qos_spec.consumer,
|
|
self.qos_spec.id,
|
|
self.qos_spec.name,
|
|
format_columns.DictColumn(self.qos_spec.specs),
|
|
)
|
|
self.assertCountEqual(datalist, tuple(data))
|
|
|
|
|
|
class TestQosUnset(TestQos):
|
|
qos_spec = volume_fakes.create_one_qos()
|
|
|
|
def setUp(self):
|
|
super().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)
|
|
|
|
def test_qos_unset_nothing(self):
|
|
arglist = [
|
|
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.assertIsNone(result)
|