Merge "Add support for QoS type and specs (2.94)"
This commit is contained in:
Zuul
@@ -221,3 +221,10 @@ resource locks
|
||||
|
||||
.. autoprogram-cliff:: openstack.share.v2
|
||||
:command: share lock *
|
||||
|
||||
=========
|
||||
qos types
|
||||
=========
|
||||
|
||||
.. autoprogram-cliff:: openstack.share.v2
|
||||
:command: share qos type *
|
||||
|
||||
@@ -27,7 +27,7 @@ from manilaclient import utils
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
MAX_VERSION = '2.93'
|
||||
MAX_VERSION = '2.94'
|
||||
MIN_VERSION = '2.0'
|
||||
DEPRECATED_VERSION = '1.0'
|
||||
_VERSIONED_METHOD_MAP = {}
|
||||
|
||||
@@ -113,6 +113,14 @@ BACKUP_SORT_KEY_VALUES = (
|
||||
'project_id',
|
||||
)
|
||||
|
||||
QOS_TYPE_SORT_KEY_VALUES = (
|
||||
'id',
|
||||
'name',
|
||||
'created_at',
|
||||
'updated_at',
|
||||
)
|
||||
|
||||
|
||||
TASK_STATE_MIGRATION_SUCCESS = 'migration_success'
|
||||
TASK_STATE_MIGRATION_ERROR = 'migration_error'
|
||||
TASK_STATE_MIGRATION_CANCELLED = 'migration_cancelled'
|
||||
@@ -172,3 +180,4 @@ REPLICA_GRADUATION_VERSION = '2.56'
|
||||
REPLICA_PRE_GRADUATION_VERSION = '2.55'
|
||||
SHARE_TRANSFER_VERSION = '2.77'
|
||||
RESOURCE_LOCK_VERSION = '2.81'
|
||||
QOS_TYPE_VERSION = '2.94'
|
||||
|
||||
390
manilaclient/osc/v2/qos_types.py
Normal file
390
manilaclient/osc/v2/qos_types.py
Normal file
@@ -0,0 +1,390 @@
|
||||
# Copyright (c) 2025 Cloudification GmbH.
|
||||
#
|
||||
# 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 logging
|
||||
|
||||
from osc_lib.cli import parseractions
|
||||
from osc_lib.command import command
|
||||
from osc_lib import exceptions
|
||||
from osc_lib import utils as oscutils
|
||||
|
||||
from manilaclient.common._i18n import _
|
||||
from manilaclient.common.apiclient import utils as apiutils
|
||||
from manilaclient.common import constants
|
||||
from manilaclient.osc import utils
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
ATTRIBUTES = [
|
||||
'id',
|
||||
'name',
|
||||
'description',
|
||||
'specs',
|
||||
'created_at',
|
||||
'updated_at',
|
||||
]
|
||||
|
||||
|
||||
def format_qos_type(qos_type, formatter='table'):
|
||||
specs = qos_type.specs
|
||||
if formatter == 'table':
|
||||
qos_type._info.update({'specs': utils.format_properties(specs)})
|
||||
else:
|
||||
qos_type._info.update({'specs': specs})
|
||||
return qos_type
|
||||
|
||||
|
||||
class CreateQosType(command.ShowOne):
|
||||
"""Create new qos type."""
|
||||
|
||||
_description = _("Create new qos type")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super().get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'name',
|
||||
metavar="<name>",
|
||||
help=_('QoS type name. This must be unique.'),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--description",
|
||||
metavar="<description>",
|
||||
default=None,
|
||||
help=_("QoS type description."),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--spec",
|
||||
type=str,
|
||||
metavar='<key=value>',
|
||||
action='append',
|
||||
default=None,
|
||||
help=_(
|
||||
"Spec key and value of QoS type that will be"
|
||||
" used for QoS type creation. OPTIONAL: Default=None."
|
||||
" Example: --spec qos_type='fixed' --spec peak_iops=300."
|
||||
),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
share_client = self.app.client_manager.share
|
||||
|
||||
kwargs = {'name': parsed_args.name}
|
||||
if parsed_args.description:
|
||||
kwargs['description'] = parsed_args.description
|
||||
|
||||
if parsed_args.spec:
|
||||
specs = utils.extract_properties(parsed_args.spec)
|
||||
kwargs['specs'] = specs
|
||||
|
||||
qos_type = share_client.qos_types.create(**kwargs)
|
||||
formatted_type = format_qos_type(qos_type, parsed_args.formatter)
|
||||
|
||||
return (
|
||||
ATTRIBUTES,
|
||||
oscutils.get_dict_properties(formatted_type._info, ATTRIBUTES),
|
||||
)
|
||||
|
||||
|
||||
class DeleteQosType(command.Command):
|
||||
"""Delete a qos type."""
|
||||
|
||||
_description = _("Delete a qos type")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super().get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'qos_types',
|
||||
metavar="<qos_types>",
|
||||
nargs="+",
|
||||
help=_("Name or ID of the qos type(s) to delete"),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
share_client = self.app.client_manager.share
|
||||
result = 0
|
||||
|
||||
for qos_type in parsed_args.qos_types:
|
||||
try:
|
||||
qos_type_obj = apiutils.find_resource(
|
||||
share_client.qos_types, qos_type
|
||||
)
|
||||
|
||||
share_client.qos_types.delete(qos_type_obj)
|
||||
except Exception as e:
|
||||
result += 1
|
||||
LOG.error(
|
||||
_(
|
||||
"Failed to delete qos type with "
|
||||
"name or ID '%(qos_type)s': %(e)s"
|
||||
),
|
||||
{'qos_type': qos_type, 'e': e},
|
||||
)
|
||||
|
||||
if result > 0:
|
||||
total = len(parsed_args.qos_types)
|
||||
msg = _("%(result)s of %(total)s qos types failed to delete.") % {
|
||||
'result': result,
|
||||
'total': total,
|
||||
}
|
||||
raise exceptions.CommandError(msg)
|
||||
|
||||
|
||||
class SetQosType(command.Command):
|
||||
"""Set qos type description or specs."""
|
||||
|
||||
_description = _("Set qos type description or specs")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super().get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'qos_type',
|
||||
metavar="<qos_type>",
|
||||
help=_("Name or ID of the qos type to modify"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--description",
|
||||
metavar="<description>",
|
||||
default=None,
|
||||
help=_("New description of qos type."),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--spec",
|
||||
type=str,
|
||||
metavar='<key=value>',
|
||||
action='append',
|
||||
default=None,
|
||||
help=_(
|
||||
"Spec key and value of qos type that will be "
|
||||
"used for QoS type. OPTIONAL: Default=None. For "
|
||||
"example --spec qos_type='fixed' --spec peak_iops=300"
|
||||
),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
share_client = self.app.client_manager.share
|
||||
|
||||
qos_type = apiutils.find_resource(
|
||||
share_client.qos_types, parsed_args.qos_type
|
||||
)
|
||||
|
||||
kwargs = {}
|
||||
if parsed_args.description:
|
||||
kwargs['description'] = parsed_args.description
|
||||
if kwargs:
|
||||
try:
|
||||
qos_type.update(**kwargs)
|
||||
except Exception as e:
|
||||
raise exceptions.CommandError(
|
||||
_("Failed to set qos type description: %s") % e
|
||||
)
|
||||
|
||||
# These are dict of key=value to be added as qos type specs.
|
||||
if parsed_args.spec:
|
||||
specs = utils.extract_properties(parsed_args.spec)
|
||||
try:
|
||||
qos_type.set_keys(specs)
|
||||
except Exception as e:
|
||||
raise exceptions.CommandError(
|
||||
_("Failed to set qos type spec: %s") % e
|
||||
)
|
||||
|
||||
|
||||
class UnsetQosType(command.Command):
|
||||
"""Unset qos type specs."""
|
||||
|
||||
_description = _("Unset qos type specs")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super().get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'qos_type',
|
||||
metavar="<qos_type>",
|
||||
help=_("Name or ID of the qos type to modify"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--description",
|
||||
action='store_true',
|
||||
help=_("Unset qos type description."),
|
||||
)
|
||||
parser.add_argument(
|
||||
'--spec',
|
||||
metavar='<key>',
|
||||
action='append',
|
||||
help=_('Remove specified spec from this qos type'),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
share_client = self.app.client_manager.share
|
||||
|
||||
qos_type = apiutils.find_resource(
|
||||
share_client.qos_types, parsed_args.qos_type
|
||||
)
|
||||
|
||||
kwargs = {}
|
||||
if parsed_args.description:
|
||||
kwargs['description'] = None
|
||||
if kwargs:
|
||||
try:
|
||||
qos_type.update(**kwargs)
|
||||
except Exception as e:
|
||||
raise exceptions.CommandError(
|
||||
_("Failed to unset qos type description: %s") % e
|
||||
)
|
||||
|
||||
# These are list of keys to be deleted from qos type specs.
|
||||
if parsed_args.spec:
|
||||
try:
|
||||
qos_type.unset_keys(parsed_args.spec)
|
||||
except Exception as e:
|
||||
raise exceptions.CommandError(
|
||||
_("Failed to remove qos type spec: %s") % e
|
||||
)
|
||||
|
||||
|
||||
class ListQosType(command.Lister):
|
||||
"""List Qos Types."""
|
||||
|
||||
_description = _("List qos types")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super().get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'--name',
|
||||
metavar="<name>",
|
||||
default=None,
|
||||
help=_('Filter results by name. Default=None.'),
|
||||
)
|
||||
parser.add_argument(
|
||||
'--description',
|
||||
metavar="<description>",
|
||||
default=None,
|
||||
help=_("Filter results by description. Default=None."),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--name~",
|
||||
metavar="<name~>",
|
||||
default=None,
|
||||
help=_("Filter results matching a qos name pattern. "),
|
||||
)
|
||||
parser.add_argument(
|
||||
'--description~',
|
||||
metavar="<description~>",
|
||||
default=None,
|
||||
help=_("Filter results matching a qos description pattern."),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
metavar="<num-qos-types>",
|
||||
type=int,
|
||||
default=None,
|
||||
action=parseractions.NonNegativeAction,
|
||||
help=_("Limit the number of qos types returned. Default=None."),
|
||||
)
|
||||
parser.add_argument(
|
||||
'--offset',
|
||||
metavar="<offset>",
|
||||
default=None,
|
||||
help=_('Start position of qos type records listing.'),
|
||||
)
|
||||
parser.add_argument(
|
||||
'--sort-key',
|
||||
'--sort_key',
|
||||
metavar='<sort_key>',
|
||||
type=str,
|
||||
default=None,
|
||||
help=_(
|
||||
'Key to be sorted with, available keys are '
|
||||
'%(keys)s. Default=None.'
|
||||
)
|
||||
% {'keys': constants.QOS_TYPE_SORT_KEY_VALUES},
|
||||
)
|
||||
parser.add_argument(
|
||||
'--sort-dir',
|
||||
'--sort_dir',
|
||||
metavar='<sort_dir>',
|
||||
type=str,
|
||||
default=None,
|
||||
help=_(
|
||||
'Sort direction, available values are '
|
||||
'%(dirs)s. OPTIONAL: Default=None.'
|
||||
)
|
||||
% {'dirs': constants.SORT_DIR_VALUES},
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
share_client = self.app.client_manager.share
|
||||
|
||||
search_opts = {
|
||||
'name': parsed_args.name,
|
||||
'description': parsed_args.description,
|
||||
'limit': parsed_args.limit,
|
||||
'offset': parsed_args.offset,
|
||||
}
|
||||
search_opts['name~'] = getattr(parsed_args, 'name~')
|
||||
search_opts['description~'] = getattr(parsed_args, 'description~')
|
||||
|
||||
qos_types = share_client.qos_types.list(
|
||||
search_opts=search_opts,
|
||||
sort_key=parsed_args.sort_key,
|
||||
sort_dir=parsed_args.sort_dir,
|
||||
)
|
||||
|
||||
formatted_types = []
|
||||
for qos_type in qos_types:
|
||||
formatted_types.append(
|
||||
format_qos_type(qos_type, parsed_args.formatter)
|
||||
)
|
||||
|
||||
values = (
|
||||
oscutils.get_dict_properties(s._info, ATTRIBUTES)
|
||||
for s in formatted_types
|
||||
)
|
||||
|
||||
columns = utils.format_column_headers(ATTRIBUTES)
|
||||
|
||||
return (columns, values)
|
||||
|
||||
|
||||
class ShowQosType(command.ShowOne):
|
||||
"""Show a qos type."""
|
||||
|
||||
_description = _("Display qos type details")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super().get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'qos_type',
|
||||
metavar="<qos_type>",
|
||||
help=_("Qos type to display (name or ID)"),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
share_client = self.app.client_manager.share
|
||||
|
||||
qos_type = apiutils.find_resource(
|
||||
share_client.qos_types, parsed_args.qos_type
|
||||
)
|
||||
|
||||
formatted_type = format_qos_type(qos_type, parsed_args.formatter)
|
||||
|
||||
return (
|
||||
ATTRIBUTES,
|
||||
oscutils.get_dict_properties(formatted_type._info, ATTRIBUTES),
|
||||
)
|
||||
@@ -563,3 +563,36 @@ class OSCClientTestBase(base.ClientTestBase):
|
||||
)
|
||||
|
||||
return backup_object
|
||||
|
||||
def create_qos_type(
|
||||
self,
|
||||
name=None,
|
||||
description=None,
|
||||
specs=None,
|
||||
add_cleanup=True,
|
||||
client=None,
|
||||
formatter=None,
|
||||
):
|
||||
name = name or data_utils.rand_name('autotest_qos_type_name')
|
||||
specs = specs or {}
|
||||
|
||||
cmd = f'create {name} '
|
||||
if description:
|
||||
cmd += f' --description {description}'
|
||||
if specs:
|
||||
q_specs = ''
|
||||
for key, value in specs.items():
|
||||
q_specs += f' --spec {key}={value}'
|
||||
cmd += q_specs
|
||||
|
||||
if formatter == 'json':
|
||||
cmd = f'share qos type {cmd} -f {formatter} '
|
||||
qos_type = json.loads(self.openstack(cmd, client=client))
|
||||
else:
|
||||
qos_type = self.dict_result('share qos type', cmd, client=client)
|
||||
|
||||
if add_cleanup:
|
||||
self.addCleanup(
|
||||
self.openstack, f'share qos type delete {qos_type["id"]}'
|
||||
)
|
||||
return qos_type
|
||||
|
||||
146
manilaclient/tests/functional/osc/test_qos_types.py
Normal file
146
manilaclient/tests/functional/osc/test_qos_types.py
Normal file
@@ -0,0 +1,146 @@
|
||||
# Copyright (c) 2025 Cloudification GmbH.
|
||||
#
|
||||
# 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 json
|
||||
|
||||
from manilaclient.tests.functional.osc import base
|
||||
|
||||
|
||||
class QosTypesCLITest(base.OSCClientTestBase):
|
||||
def test_qos_type_create(self):
|
||||
name = 'test_qos_type'
|
||||
description = 'Description'
|
||||
qos_type = self.create_qos_type(name=name, description=description)
|
||||
|
||||
self.assertEqual(name, qos_type["name"])
|
||||
self.assertEqual(description, qos_type["description"])
|
||||
|
||||
qos_types_list = self.listing_result('share qos type', 'list')
|
||||
self.assertIn(qos_type["id"], [item['ID'] for item in qos_types_list])
|
||||
|
||||
def test_qos_type_create_specs(self):
|
||||
name = 'test_qos_type2'
|
||||
qos_type = self.create_qos_type(
|
||||
name=name,
|
||||
specs={"foo": "bar", "manila": "gazpacho"},
|
||||
formatter='json',
|
||||
)
|
||||
|
||||
specs = qos_type["specs"]
|
||||
self.assertEqual("bar", specs["foo"])
|
||||
self.assertEqual("gazpacho", specs["manila"])
|
||||
|
||||
def test_qos_type_create_specs_using_command(self):
|
||||
self.openstack(
|
||||
'share qos type create test_qos_type2_cli '
|
||||
'--description test_cli --spec foo1=bar1 --spec foo2=bar2'
|
||||
)
|
||||
self.addCleanup(
|
||||
self.openstack, 'share qos type delete test_qos_type2_cli'
|
||||
)
|
||||
|
||||
qos_type = json.loads(
|
||||
self.openstack('share qos type show test_qos_type2_cli -f json')
|
||||
)
|
||||
|
||||
self.assertEqual('test_cli', qos_type["description"])
|
||||
self.assertEqual('bar1', qos_type["specs"]["foo1"])
|
||||
self.assertEqual('bar2', qos_type["specs"]["foo2"])
|
||||
|
||||
def test_qos_type_delete(self):
|
||||
qos_type_1 = self.create_qos_type(
|
||||
name='test_qos_type3', add_cleanup=False
|
||||
)
|
||||
qos_type_2 = self.create_qos_type(
|
||||
name='test_qos_type4', add_cleanup=False
|
||||
)
|
||||
|
||||
self.openstack(
|
||||
f'share qos type delete {qos_type_1["id"]} {qos_type_2["id"]}'
|
||||
)
|
||||
|
||||
self.check_object_deleted('share qos type', qos_type_1["id"])
|
||||
self.check_object_deleted('share qos type', qos_type_2["id"])
|
||||
|
||||
def test_qos_type_set(self):
|
||||
qos_type = self.create_qos_type(name='test_qos_type5')
|
||||
|
||||
self.openstack(
|
||||
f'share qos type set {qos_type["id"]} --description Description'
|
||||
' --spec foo=bar2'
|
||||
)
|
||||
|
||||
qos_type = json.loads(
|
||||
self.openstack(f'share qos type show {qos_type["id"]} -f json')
|
||||
)
|
||||
|
||||
self.assertEqual('Description', qos_type["description"])
|
||||
self.assertEqual('bar2', qos_type["specs"]["foo"])
|
||||
|
||||
def test_qos_type_unset(self):
|
||||
qos_type = self.create_qos_type(
|
||||
name='test_qos_type6', specs={'foo': 'bar', 'foo1': 'bar1'}
|
||||
)
|
||||
|
||||
self.openstack(
|
||||
f'share qos type unset {qos_type["id"]} --spec foo --spec foo1'
|
||||
)
|
||||
|
||||
qos_type = json.loads(
|
||||
self.openstack(f'share qos type show {qos_type["id"]} -f json')
|
||||
)
|
||||
|
||||
self.assertNotIn('foo', qos_type["specs"])
|
||||
self.assertNotIn('foo1', qos_type["specs"])
|
||||
|
||||
def test_qos_type_list(self):
|
||||
qos_type_1 = self.create_qos_type(name='test_qos_type7')
|
||||
qos_type_2 = self.create_qos_type(
|
||||
name='test_qos_type8', specs={'foo': 'bar'}
|
||||
)
|
||||
|
||||
types_list = self.listing_result(
|
||||
'share qos type', 'list', client=self.admin_client
|
||||
)
|
||||
|
||||
self.assertTableStruct(
|
||||
types_list,
|
||||
[
|
||||
'ID',
|
||||
'Name',
|
||||
'Description',
|
||||
'Specs',
|
||||
],
|
||||
)
|
||||
id_list = [item['ID'] for item in types_list]
|
||||
self.assertIn(qos_type_1['id'], id_list)
|
||||
self.assertIn(qos_type_2['id'], id_list)
|
||||
|
||||
types_list = self.listing_result('share qos type', 'list')
|
||||
|
||||
id_list = [item['ID'] for item in types_list]
|
||||
self.assertIn(qos_type_1['id'], id_list)
|
||||
self.assertIn(qos_type_2['id'], id_list)
|
||||
|
||||
def test_qos_type_show(self):
|
||||
qos_type = self.create_qos_type(
|
||||
name='test_qos_type10', specs={'foo': 'bar'}
|
||||
)
|
||||
|
||||
result = json.loads(
|
||||
self.openstack(f'share qos type show {qos_type["id"]} -f json')
|
||||
)
|
||||
|
||||
self.assertEqual(qos_type["name"], result["name"])
|
||||
self.assertEqual('bar', result["specs"]["foo"])
|
||||
@@ -64,6 +64,7 @@ class FakeShareClient:
|
||||
self.share_group_type_access = mock.Mock()
|
||||
self.share_servers = mock.Mock()
|
||||
self.resource_locks = mock.Mock()
|
||||
self.qos_types = mock.Mock()
|
||||
|
||||
|
||||
class TestShare(osc_utils.TestCommand):
|
||||
@@ -1626,3 +1627,77 @@ class FakeShareBackup:
|
||||
for n in range(0, count):
|
||||
share_backups.append(FakeShareBackup.create_one_backup(attrs))
|
||||
return share_backups
|
||||
|
||||
|
||||
class FakeQosType:
|
||||
"""Fake one or more qos types"""
|
||||
|
||||
@staticmethod
|
||||
def create_one_qostype(attrs=None, methods=None):
|
||||
"""Create a fake qos type
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A FakeResource object, with project_id, resource and so on
|
||||
"""
|
||||
|
||||
attrs = attrs or {}
|
||||
methods = methods or {}
|
||||
|
||||
qos_type_info = {
|
||||
"id": 'qos-type-id-' + uuid.uuid4().hex,
|
||||
"name": 'qos-type-name-' + uuid.uuid4().hex,
|
||||
"description": 'qos-type-description-' + uuid.uuid4().hex,
|
||||
"specs": {
|
||||
"expected_iops": "2000",
|
||||
"peak_iops": "5000",
|
||||
},
|
||||
"created_at": 'time-' + uuid.uuid4().hex,
|
||||
"updated_at": 'time-' + uuid.uuid4().hex,
|
||||
}
|
||||
|
||||
qos_type_info.update(attrs)
|
||||
qos_type = osc_fakes.FakeResource(
|
||||
info=copy.deepcopy(qos_type_info), methods=methods, loaded=True
|
||||
)
|
||||
return qos_type
|
||||
|
||||
@staticmethod
|
||||
def create_qos_types(attrs=None, count=2):
|
||||
"""Create multiple fake qos types.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param Integer count:
|
||||
The number of qos types to be faked
|
||||
:return:
|
||||
A list of FakeResource objects
|
||||
"""
|
||||
|
||||
qos_types = []
|
||||
for n in range(0, count):
|
||||
qos_types.append(FakeQosType.create_one_qostype(attrs))
|
||||
|
||||
return qos_types
|
||||
|
||||
@staticmethod
|
||||
def get_qos_types(qos_types=None, count=2):
|
||||
"""Get an iterable MagicMock object with a list of faked types.
|
||||
|
||||
If types list is provided, then initialize the Mock object with the
|
||||
list. Otherwise create one.
|
||||
|
||||
:param List types:
|
||||
A list of FakeResource objects faking types
|
||||
:param Integer count:
|
||||
The number of types to be faked
|
||||
:return
|
||||
An iterable Mock object with side_effect set to a list of faked
|
||||
types
|
||||
"""
|
||||
|
||||
if qos_types is None:
|
||||
qos_types = FakeQosType.create_qos_types(count)
|
||||
|
||||
return mock.Mock(side_effect=qos_types)
|
||||
|
||||
440
manilaclient/tests/unit/osc/v2/test_qos_types.py
Normal file
440
manilaclient/tests/unit/osc/v2/test_qos_types.py
Normal file
@@ -0,0 +1,440 @@
|
||||
# Copyright (c) 2025 Cloudification GmbH.
|
||||
#
|
||||
# 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.
|
||||
#
|
||||
|
||||
from unittest import mock
|
||||
|
||||
from osc_lib import exceptions
|
||||
from osc_lib import utils as oscutils
|
||||
|
||||
from manilaclient import api_versions
|
||||
from manilaclient.common.apiclient.exceptions import BadRequest
|
||||
from manilaclient.common.apiclient.exceptions import NotFound
|
||||
from manilaclient.osc import utils
|
||||
from manilaclient.osc.v2 import qos_types as osc_qos_types
|
||||
from manilaclient.tests.unit.osc import osc_utils
|
||||
from manilaclient.tests.unit.osc.v2 import fakes as manila_fakes
|
||||
|
||||
COLUMNS = [
|
||||
'id',
|
||||
'name',
|
||||
'description',
|
||||
'specs',
|
||||
'created_at',
|
||||
'updated_at',
|
||||
]
|
||||
|
||||
|
||||
class TestQosType(manila_fakes.TestShare):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.qos_types_mock = self.app.client_manager.share.qos_types
|
||||
self.qos_types_mock.reset_mock()
|
||||
self.app.client_manager.share.api_version = api_versions.APIVersion(
|
||||
api_versions.MAX_VERSION
|
||||
)
|
||||
|
||||
|
||||
class TestQosTypeCreate(TestQosType):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.new_qos_type = manila_fakes.FakeQosType.create_one_qostype()
|
||||
self.qos_types_mock.create.return_value = self.new_qos_type
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = osc_qos_types.CreateQosType(self.app, None)
|
||||
|
||||
self.data = [
|
||||
self.new_qos_type.id,
|
||||
self.new_qos_type.name,
|
||||
self.new_qos_type.description,
|
||||
('expected_iops : 2000\npeak_iops : 5000'),
|
||||
self.new_qos_type.created_at,
|
||||
self.new_qos_type.updated_at,
|
||||
]
|
||||
|
||||
self.raw_data = [
|
||||
self.new_qos_type.id,
|
||||
self.new_qos_type.name,
|
||||
self.new_qos_type.description,
|
||||
{
|
||||
'expected_iops': '2000',
|
||||
'peak_iops': '5000',
|
||||
},
|
||||
self.new_qos_type.created_at,
|
||||
self.new_qos_type.updated_at,
|
||||
]
|
||||
|
||||
def test_qos_type_create_required_args(self):
|
||||
"""Verifies required arguments."""
|
||||
|
||||
arglist = [self.new_qos_type.name]
|
||||
verifylist = [
|
||||
('name', self.new_qos_type.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.qos_types_mock.create.assert_called_with(
|
||||
name=self.new_qos_type.name,
|
||||
)
|
||||
|
||||
self.assertCountEqual(COLUMNS, columns)
|
||||
self.assertCountEqual(self.data, data)
|
||||
|
||||
def test_qos_type_create_json_format(self):
|
||||
"""Verifies --format json."""
|
||||
|
||||
arglist = [self.new_qos_type.name, '-f', 'json']
|
||||
verifylist = [
|
||||
('name', self.new_qos_type.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.qos_types_mock.create.assert_called_with(
|
||||
name=self.new_qos_type.name,
|
||||
)
|
||||
|
||||
self.assertCountEqual(COLUMNS, columns)
|
||||
self.assertCountEqual(self.raw_data, data)
|
||||
|
||||
def test_qos_type_create_missing_required_arg(self):
|
||||
"""Verifies missing required arguments."""
|
||||
|
||||
arglist = []
|
||||
verifylist = []
|
||||
|
||||
self.assertRaises(
|
||||
osc_utils.ParserException,
|
||||
self.check_parser,
|
||||
self.cmd,
|
||||
arglist,
|
||||
verifylist,
|
||||
)
|
||||
|
||||
def test_qos_type_create_specs(self):
|
||||
arglist = [
|
||||
self.new_qos_type.name,
|
||||
'--spec',
|
||||
'peak_iops=100',
|
||||
'--spec',
|
||||
'expected_iops=20',
|
||||
]
|
||||
verifylist = [
|
||||
('name', self.new_qos_type.name),
|
||||
('spec', ['peak_iops=100', 'expected_iops=20']),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.qos_types_mock.create.assert_called_with(
|
||||
name=self.new_qos_type.name,
|
||||
specs={'peak_iops': '100', 'expected_iops': '20'},
|
||||
)
|
||||
|
||||
self.assertCountEqual(COLUMNS, columns)
|
||||
self.assertCountEqual(self.data, data)
|
||||
|
||||
|
||||
class TestQosTypeDelete(TestQosType):
|
||||
qos_types = manila_fakes.FakeQosType.create_qos_types(count=2)
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.qos_types_mock.get = manila_fakes.FakeQosType.get_qos_types(
|
||||
self.qos_types
|
||||
)
|
||||
self.qos_types_mock.delete.return_value = None
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = osc_qos_types.DeleteQosType(self.app, None)
|
||||
|
||||
def test_qos_type_delete_one(self):
|
||||
arglist = [self.qos_types[0].id]
|
||||
|
||||
verifylist = [('qos_types', [self.qos_types[0].id])]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.qos_types_mock.delete.assert_called_with(self.qos_types[0])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_qos_type_delete_multiple(self):
|
||||
arglist = []
|
||||
for t in self.qos_types:
|
||||
arglist.append(t.id)
|
||||
verifylist = [
|
||||
('qos_types', arglist),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
|
||||
calls = []
|
||||
for t in self.qos_types:
|
||||
calls.append(mock.call(t))
|
||||
self.qos_types_mock.delete.assert_has_calls(calls)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_delete_qos_type_with_exception(self):
|
||||
arglist = [
|
||||
'non_existing_type',
|
||||
]
|
||||
verifylist = [
|
||||
('qos_types', arglist),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
self.qos_types_mock.delete.side_effect = exceptions.CommandError()
|
||||
self.assertRaises(
|
||||
exceptions.CommandError, self.cmd.take_action, parsed_args
|
||||
)
|
||||
|
||||
|
||||
class TestQosTypeSet(TestQosType):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.qos_type = manila_fakes.FakeQosType.create_one_qostype(
|
||||
methods={'set_keys': None, 'update': None}
|
||||
)
|
||||
self.qos_types_mock.get.return_value = self.qos_type
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = osc_qos_types.SetQosType(self.app, None)
|
||||
|
||||
def test_qos_type_set_specs(self):
|
||||
arglist = [
|
||||
self.qos_type.id,
|
||||
'--spec',
|
||||
'peak_iops=100',
|
||||
]
|
||||
verifylist = [
|
||||
('qos_type', self.qos_type.id),
|
||||
('spec', ['peak_iops=100']),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.qos_type.set_keys.assert_called_with({'peak_iops': '100'})
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_qos_type_set_description(self):
|
||||
arglist = [self.qos_type.id, '--description', 'new description']
|
||||
verifylist = [
|
||||
('qos_type', self.qos_type.id),
|
||||
('description', 'new description'),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.qos_type.update.assert_called_with(description='new description')
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_qos_type_set_specs_exception(self):
|
||||
arglist = [
|
||||
self.qos_type.id,
|
||||
'--spec',
|
||||
'peak_iops=100',
|
||||
]
|
||||
verifylist = [
|
||||
('qos_type', self.qos_type.id),
|
||||
('spec', ['peak_iops=100']),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
self.qos_type.set_keys.side_effect = BadRequest()
|
||||
self.assertRaises(
|
||||
exceptions.CommandError, self.cmd.take_action, parsed_args
|
||||
)
|
||||
|
||||
|
||||
class TestQosTypeUnset(TestQosType):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.qos_type = manila_fakes.FakeQosType.create_one_qostype(
|
||||
methods={'unset_keys': None, 'update': None}
|
||||
)
|
||||
|
||||
self.qos_types_mock.get.return_value = self.qos_type
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = osc_qos_types.UnsetQosType(self.app, None)
|
||||
|
||||
def test_qos_type_unset_description(self):
|
||||
arglist = [self.qos_type.id, '--description']
|
||||
verifylist = [
|
||||
('qos_type', self.qos_type.id),
|
||||
('description', True),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.qos_type.update.assert_called_with(description=None)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_qos_type_unset_specs(self):
|
||||
arglist = [self.qos_type.id, '--spec', 'peak_iops']
|
||||
verifylist = [
|
||||
('qos_type', self.qos_type.id),
|
||||
('spec', ['peak_iops']),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.qos_type.unset_keys.assert_called_with(['peak_iops'])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_qos_type_unset_exception(self):
|
||||
arglist = [self.qos_type.id, '--spec', 'peak_iops']
|
||||
verifylist = [
|
||||
('qos_type', self.qos_type.id),
|
||||
('spec', ['peak_iops']),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
self.qos_type.unset_keys.side_effect = NotFound()
|
||||
self.assertRaises(
|
||||
exceptions.CommandError, self.cmd.take_action, parsed_args
|
||||
)
|
||||
|
||||
|
||||
class TestQosTypeList(TestQosType):
|
||||
qos_types = manila_fakes.FakeQosType.create_qos_types()
|
||||
columns = utils.format_column_headers(COLUMNS)
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.qos_types_mock.list.return_value = self.qos_types
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = osc_qos_types.ListQosType(self.app, None)
|
||||
|
||||
self.values = (
|
||||
oscutils.get_dict_properties(s._info, COLUMNS)
|
||||
for s in self.qos_types
|
||||
)
|
||||
|
||||
def test_qos_type_list(self):
|
||||
arglist = []
|
||||
verifylist = []
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
self.qos_types_mock.list.assert_called_once_with(
|
||||
search_opts={
|
||||
'offset': None,
|
||||
'limit': None,
|
||||
'name': None,
|
||||
'description': None,
|
||||
'name~': None,
|
||||
'description~': None,
|
||||
},
|
||||
sort_key=None,
|
||||
sort_dir=None,
|
||||
)
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertEqual(list(self.values), list(data))
|
||||
|
||||
def test_qos_type_list_by_name(self):
|
||||
arglist = ['--name', 'fake_name']
|
||||
verifylist = [('name', 'fake_name')]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
self.qos_types_mock.list.assert_called_with(
|
||||
search_opts={
|
||||
'offset': None,
|
||||
'limit': None,
|
||||
'name': 'fake_name',
|
||||
'description': None,
|
||||
'name~': None,
|
||||
'description~': None,
|
||||
},
|
||||
sort_key=None,
|
||||
sort_dir=None,
|
||||
)
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertEqual(list(self.values), list(data))
|
||||
|
||||
|
||||
class TestQosTypeShow(TestQosType):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.qos_type = manila_fakes.FakeQosType.create_one_qostype()
|
||||
|
||||
self.qos_types_mock.get.return_value = self.qos_type
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = osc_qos_types.ShowQosType(self.app, None)
|
||||
|
||||
self.data = [
|
||||
self.qos_type.id,
|
||||
self.qos_type.name,
|
||||
self.qos_type.description,
|
||||
('expected_iops : 2000\npeak_iops : 5000'),
|
||||
self.qos_type.created_at,
|
||||
self.qos_type.updated_at,
|
||||
]
|
||||
|
||||
self.raw_data = [
|
||||
self.qos_type.id,
|
||||
self.qos_type.name,
|
||||
self.qos_type.description,
|
||||
{
|
||||
'expected_iops': '2000',
|
||||
'peak_iops': '5000',
|
||||
},
|
||||
self.qos_type.created_at,
|
||||
self.qos_type.updated_at,
|
||||
]
|
||||
|
||||
def test_qos_type_show(self):
|
||||
arglist = [self.qos_type.id]
|
||||
verifylist = [("qos_type", self.qos_type.id)]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
self.qos_types_mock.get.assert_called_with(self.qos_type.id)
|
||||
|
||||
self.assertCountEqual(COLUMNS, columns)
|
||||
self.assertCountEqual(self.data, data)
|
||||
|
||||
def test_qos_type_show_json_format(self):
|
||||
arglist = [
|
||||
self.qos_type.id,
|
||||
'-f',
|
||||
'json',
|
||||
]
|
||||
verifylist = [("qos_type", self.qos_type.id)]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
self.qos_types_mock.get.assert_called_with(self.qos_type.id)
|
||||
|
||||
self.assertCountEqual(COLUMNS, columns)
|
||||
self.assertCountEqual(self.raw_data, data)
|
||||
@@ -1236,6 +1236,62 @@ class FakeHTTPClient(fakes.FakeHTTPClient):
|
||||
def delete_types_1(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def get_qos_types(self, **kw):
|
||||
response_body = {
|
||||
'qos_types': [
|
||||
{
|
||||
'id': 1,
|
||||
'name': 'test-type-1',
|
||||
'specs': {'test1': 'test1'},
|
||||
},
|
||||
{
|
||||
'id': 2,
|
||||
'name': 'test-type-2',
|
||||
'specs': {'test1': 'test1'},
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
return 200, {}, response_body
|
||||
|
||||
def get_qos_types_1(self, **kw):
|
||||
return (
|
||||
200,
|
||||
{},
|
||||
{
|
||||
'qos_type': {
|
||||
'id': 1,
|
||||
'name': 'test-qos-type-1',
|
||||
'specs': {'test1': 'test1'},
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
def post_qos_types(self, body, **kw):
|
||||
qos_type = body['qos_type']
|
||||
return (
|
||||
202,
|
||||
{},
|
||||
{
|
||||
'qos_type': {
|
||||
'id': 3,
|
||||
'name': 'test-qos-type-3',
|
||||
'description': 'test description',
|
||||
'specs': qos_type['specs'],
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
def post_qos_types_1_specs(self, body, **kw):
|
||||
assert list(body) == ['specs']
|
||||
return (200, {}, {'specs': {'k': 'v'}})
|
||||
|
||||
def delete_qos_types_1_specs_k(self, **kw):
|
||||
return (204, {}, None)
|
||||
|
||||
def delete_qos_types_1(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def get_types_3_os_share_type_access(self, **kw):
|
||||
return (
|
||||
200,
|
||||
|
||||
180
manilaclient/tests/unit/v2/test_qos_types.py
Normal file
180
manilaclient/tests/unit/v2/test_qos_types.py
Normal file
@@ -0,0 +1,180 @@
|
||||
# Copyright (c) 2025 Cloudification GmbH.
|
||||
#
|
||||
# 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
|
||||
|
||||
import ddt
|
||||
|
||||
from manilaclient import api_versions
|
||||
from manilaclient.common import constants
|
||||
from manilaclient import config
|
||||
from manilaclient.tests.unit import utils
|
||||
from manilaclient.tests.unit.v2 import fakes
|
||||
from manilaclient.v2 import qos_types
|
||||
|
||||
cs = fakes.FakeClient(api_versions.APIVersion(constants.QOS_TYPE_VERSION))
|
||||
|
||||
CONF = config.CONF
|
||||
|
||||
LATEST_MICROVERSION = CONF.max_api_microversion
|
||||
|
||||
|
||||
@ddt.ddt
|
||||
class QosTypesTest(utils.TestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
microversion = api_versions.APIVersion(constants.QOS_TYPE_VERSION)
|
||||
mock_microversion = mock.Mock(api_version=microversion)
|
||||
self.manager = qos_types.QosTypeManager(api=mock_microversion)
|
||||
|
||||
@ddt.data(
|
||||
{'expected_iops': '100'},
|
||||
{'peak_iops': '500', 'expected_iops': '100'},
|
||||
)
|
||||
def test_init(self, specs):
|
||||
info = {'specs': specs}
|
||||
|
||||
qos_type = qos_types.QosType(qos_types.QosTypeManager, info)
|
||||
|
||||
self.assertTrue(hasattr(qos_type, '_specs'))
|
||||
self.assertIsInstance(qos_type._specs, dict)
|
||||
|
||||
def test_list_types(self):
|
||||
tl = cs.qos_types.list()
|
||||
cs.assert_called('GET', '/qos-types')
|
||||
for t in tl:
|
||||
t.api_version = api_versions.APIVersion(constants.QOS_TYPE_VERSION)
|
||||
self.assertIsInstance(t, qos_types.QosType)
|
||||
self.assertTrue(callable(getattr(t, 'get_keys', '')))
|
||||
self.assertEqual({'test1': 'test1'}, t.get_keys())
|
||||
|
||||
@ddt.data(
|
||||
{'qos_type': 'fixed', 'max_iops': '100'},
|
||||
{'qos_type': 'adaptive', 'peak_iops': '200'},
|
||||
{'qos_type': 'fixed', 'max_iops': '100', 'expected_iops': '100'},
|
||||
)
|
||||
def test_create(self, specs):
|
||||
specs = copy.copy(specs)
|
||||
|
||||
self.mock_object(
|
||||
self.manager, '_create', mock.Mock(return_value="fake")
|
||||
)
|
||||
|
||||
result = self.manager.create(
|
||||
'test-qos-type-1',
|
||||
specs=specs,
|
||||
)
|
||||
|
||||
if specs is None:
|
||||
specs = {}
|
||||
|
||||
expected_specs = dict(specs)
|
||||
|
||||
expected_body = {
|
||||
"qos_type": {
|
||||
"name": 'test-qos-type-1',
|
||||
"specs": expected_specs,
|
||||
}
|
||||
}
|
||||
|
||||
self.manager._create.assert_called_once_with(
|
||||
"/qos-types", expected_body, "qos_type"
|
||||
)
|
||||
self.assertEqual("fake", result)
|
||||
|
||||
def test_set_key(self):
|
||||
t = cs.qos_types.get(1)
|
||||
t.api_version = api_versions.APIVersion(constants.QOS_TYPE_VERSION)
|
||||
t.set_keys({'k': 'v'})
|
||||
cs.assert_called('POST', '/qos-types/1/specs', {'specs': {'k': 'v'}})
|
||||
|
||||
def test_unset_keys(self):
|
||||
t = cs.qos_types.get(1)
|
||||
t.api_version = api_versions.APIVersion(constants.QOS_TYPE_VERSION)
|
||||
t.unset_keys(['k'])
|
||||
cs.assert_called('DELETE', '/qos-types/1/specs/k')
|
||||
|
||||
def test_update(self):
|
||||
self.mock_object(
|
||||
self.manager, '_update', mock.Mock(return_value="fake")
|
||||
)
|
||||
qos_type = 1234
|
||||
body = dict(description="updated test description")
|
||||
expected_body = {
|
||||
"qos_type": dict(description="updated test description")
|
||||
}
|
||||
result = self.manager.update(qos_type, **body)
|
||||
self.manager._update.assert_called_once_with(
|
||||
f"/qos-types/{qos_type}", expected_body, "qos_type"
|
||||
)
|
||||
self.assertEqual("fake", result)
|
||||
|
||||
def test_delete(self):
|
||||
cs.qos_types.delete(1)
|
||||
cs.assert_called('DELETE', '/qos-types/1')
|
||||
|
||||
def test_get_keys_from_resource_data(self):
|
||||
version = api_versions.APIVersion(constants.QOS_TYPE_VERSION)
|
||||
manager = mock.Mock()
|
||||
manager.api = mock.Mock()
|
||||
manager.api.client = mock.Mock(return_value=None)
|
||||
manager.api.client.get = mock.Mock(return_value=(200, {}))
|
||||
|
||||
valid_specs = {'test': 'test'}
|
||||
qos_type = qos_types.QosType(
|
||||
manager,
|
||||
{'specs': valid_specs, 'name': 'test'},
|
||||
loaded=True,
|
||||
)
|
||||
qos_type.api_version = version
|
||||
|
||||
actual_result = qos_type.get_keys()
|
||||
|
||||
self.assertEqual(actual_result, valid_specs)
|
||||
self.assertEqual(manager.api.client.get.call_count, 0)
|
||||
|
||||
@ddt.data(
|
||||
{'prefer_resource_data': True, 'resource_specs': {}},
|
||||
{
|
||||
'prefer_resource_data': False,
|
||||
'resource_specs': {'fake': 'fake'},
|
||||
},
|
||||
{'prefer_resource_data': False, 'resource_specs': {}},
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_get_keys_from_api(self, prefer_resource_data, resource_specs):
|
||||
version = api_versions.APIVersion(constants.QOS_TYPE_VERSION)
|
||||
manager = mock.Mock()
|
||||
manager.api = mock.Mock()
|
||||
manager.api.client = mock.Mock(return_value=None)
|
||||
|
||||
valid_specs = {'test': 'test'}
|
||||
manager.api.client.get = mock.Mock(
|
||||
return_value=(200, {'specs': valid_specs})
|
||||
)
|
||||
|
||||
info = {
|
||||
'name': 'test',
|
||||
'uuid': 'fake',
|
||||
'specs': resource_specs,
|
||||
}
|
||||
qos_type = qos_types.QosType(manager, info, loaded=True)
|
||||
qos_type.api_version = version
|
||||
|
||||
actual_result = qos_type.get_keys(prefer_resource_data)
|
||||
|
||||
self.assertEqual(actual_result, valid_specs)
|
||||
self.assertEqual(manager.api.client.get.call_count, 1)
|
||||
@@ -22,6 +22,7 @@ from manilaclient import exceptions
|
||||
from manilaclient.v2 import availability_zones
|
||||
from manilaclient.v2 import limits
|
||||
from manilaclient.v2 import messages
|
||||
from manilaclient.v2 import qos_types
|
||||
from manilaclient.v2 import quota_classes
|
||||
from manilaclient.v2 import quotas
|
||||
from manilaclient.v2 import resource_locks
|
||||
@@ -231,6 +232,7 @@ class Client:
|
||||
self.limits = limits.LimitsManager(self)
|
||||
self.transfers = share_transfers.ShareTransferManager(self)
|
||||
self.messages = messages.MessageManager(self)
|
||||
self.qos_types = qos_types.QosTypeManager(self)
|
||||
self.services = services.ServiceManager(self)
|
||||
self.security_services = security_services.SecurityServiceManager(self)
|
||||
self.share_networks = share_networks.ShareNetworkManager(self)
|
||||
|
||||
185
manilaclient/v2/qos_types.py
Normal file
185
manilaclient/v2/qos_types.py
Normal file
@@ -0,0 +1,185 @@
|
||||
# Copyright (c) 2025 Cloudification GmbH.
|
||||
#
|
||||
# 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.
|
||||
|
||||
|
||||
"""
|
||||
Qos Type interface.
|
||||
"""
|
||||
|
||||
from manilaclient import api_versions
|
||||
from manilaclient import base
|
||||
from manilaclient.common import constants
|
||||
|
||||
|
||||
class QosType(base.Resource):
|
||||
"""A Qos Type represents Quality of service of Manila resource."""
|
||||
|
||||
def __init__(self, manager, info, loaded=False):
|
||||
super().__init__(manager, info, loaded)
|
||||
self._specs = info.get('specs', {})
|
||||
|
||||
def __repr__(self):
|
||||
return f"<QosType: {self.name}>"
|
||||
|
||||
def get_keys(self, prefer_resource_data=True):
|
||||
"""Get specs from a qos type.
|
||||
|
||||
:param prefer_resource_data: By default specs are retrieved from
|
||||
resource data, but user can force this method to make API call.
|
||||
:return: dict with specs
|
||||
"""
|
||||
specs = getattr(self, 'specs', None)
|
||||
|
||||
if prefer_resource_data and specs:
|
||||
return specs
|
||||
|
||||
qos_type_id = base.getid(self)
|
||||
_resp, body = self.manager.api.client.get(
|
||||
f"/qos-types/{qos_type_id}/specs"
|
||||
)
|
||||
|
||||
self.specs = body["specs"]
|
||||
|
||||
return body["specs"]
|
||||
|
||||
def set_keys(self, metadata):
|
||||
"""Set specs on a qos type.
|
||||
|
||||
:param metadata: A dict of key/value pairs to be set
|
||||
"""
|
||||
body = {'specs': metadata}
|
||||
qos_type_id = base.getid(self)
|
||||
return self.manager._create(
|
||||
f"/qos-types/{qos_type_id}/specs",
|
||||
body,
|
||||
"specs",
|
||||
return_raw=True,
|
||||
)
|
||||
|
||||
def unset_keys(self, keys):
|
||||
"""Unset specs on a qos type.
|
||||
|
||||
:param keys: A list of keys to be unset
|
||||
"""
|
||||
qos_type_id = base.getid(self)
|
||||
for k in keys:
|
||||
self.manager._delete(f"/qos-types/{qos_type_id}/specs/{k}")
|
||||
|
||||
def update(self, **kwargs):
|
||||
"""Update this qos type."""
|
||||
return self.manager.update(self, **kwargs)
|
||||
|
||||
def delete(self):
|
||||
"""Delete this qos type."""
|
||||
return self.manager.delete(self)
|
||||
|
||||
|
||||
class QosTypeManager(base.ManagerWithFind):
|
||||
"""Manage :class:`QosType` resources."""
|
||||
|
||||
resource_class = QosType
|
||||
|
||||
@api_versions.wraps(constants.QOS_TYPE_VERSION)
|
||||
def list(self, search_opts=None, sort_key=None, sort_dir=None):
|
||||
"""Get a list of all qos types.
|
||||
|
||||
:param search_opts: Search options to filter out qos types.
|
||||
:param sort_key: Key to be sorted.
|
||||
:param sort_dir: Sort direction, should be 'desc' or 'asc'.
|
||||
:rtype: list of :class:`QosType`.
|
||||
"""
|
||||
search_opts = search_opts or {}
|
||||
|
||||
if sort_key is not None:
|
||||
if sort_key in constants.QOS_TYPE_SORT_KEY_VALUES:
|
||||
search_opts['sort_key'] = sort_key
|
||||
else:
|
||||
raise ValueError(
|
||||
'sort_key must be one of the following: {}.'.format(
|
||||
', '.join(constants.QOS_TYPE_SORT_KEY_VALUES)
|
||||
)
|
||||
)
|
||||
|
||||
if sort_dir is not None:
|
||||
if sort_dir in constants.SORT_DIR_VALUES:
|
||||
search_opts['sort_dir'] = sort_dir
|
||||
else:
|
||||
raise ValueError(
|
||||
'sort_dir must be one of the following: {}.'.format(
|
||||
', '.join(constants.SORT_DIR_VALUES)
|
||||
)
|
||||
)
|
||||
|
||||
query_string = self._build_query_string(search_opts)
|
||||
return self._list(f"/qos-types{query_string}", "qos_types")
|
||||
|
||||
@api_versions.wraps(constants.QOS_TYPE_VERSION)
|
||||
def get(self, qos_type):
|
||||
"""Get a specific qos type.
|
||||
|
||||
:param qos_type: The ID of the :class:`QosType` to get.
|
||||
:rtype: :class:`QosType`
|
||||
"""
|
||||
qos_type_id = base.getid(qos_type)
|
||||
return self._get(f"/qos-types/{qos_type_id}", "qos_type")
|
||||
|
||||
@api_versions.wraps(constants.QOS_TYPE_VERSION)
|
||||
def delete(self, qos_type):
|
||||
"""Delete a specific qos_type.
|
||||
|
||||
:param qos_type: The name or ID of the :class:`QosType` to get.
|
||||
"""
|
||||
qos_type_id = base.getid(qos_type)
|
||||
self._delete(f"/qos-types/{qos_type_id}")
|
||||
|
||||
@api_versions.wraps(constants.QOS_TYPE_VERSION)
|
||||
def create(self, name, description=None, specs=None):
|
||||
"""Create a qos type.
|
||||
|
||||
:param name: Descriptive name of the qos type
|
||||
:param specs: Specs of the qos type
|
||||
:rtype: :class:`QosType`
|
||||
"""
|
||||
if specs is None:
|
||||
specs = {}
|
||||
|
||||
body = {
|
||||
"qos_type": {
|
||||
"name": name,
|
||||
"specs": specs,
|
||||
}
|
||||
}
|
||||
|
||||
if description:
|
||||
body["qos_type"]["description"] = description
|
||||
return self._create("/qos-types", body, "qos_type")
|
||||
|
||||
@api_versions.wraps(constants.QOS_TYPE_VERSION)
|
||||
def update(self, qos_type, **kwargs):
|
||||
"""Update the description for a qos type.
|
||||
|
||||
:param qos_type: the ID of the :class: `QosType` to update.
|
||||
:param description: Description of the qos type.
|
||||
:rtype: :class:`QosType`
|
||||
"""
|
||||
|
||||
if not kwargs:
|
||||
return
|
||||
|
||||
body = {
|
||||
'qos_type': kwargs,
|
||||
}
|
||||
qos_type_id = base.getid(qos_type)
|
||||
return self._update(f"/qos-types/{qos_type_id}", body, "qos_type")
|
||||
@@ -184,6 +184,12 @@ share_lock_show = "manilaclient.osc.v2.resource_locks:ShowResourceLock"
|
||||
share_lock_set = " manilaclient.osc.v2.resource_locks:SetResourceLock"
|
||||
share_lock_unset = "manilaclient.osc.v2.resource_locks:UnsetResourceLock"
|
||||
share_lock_delete = "manilaclient.osc.v2.resource_locks:DeleteResourceLock"
|
||||
share_qos_type_create = "manilaclient.osc.v2.qos_types:CreateQosType"
|
||||
share_qos_type_list = "manilaclient.osc.v2.qos_types:ListQosType"
|
||||
share_qos_type_show = "manilaclient.osc.v2.qos_types:ShowQosType"
|
||||
share_qos_type_set = "manilaclient.osc.v2.qos_types:SetQosType"
|
||||
share_qos_type_unset = "manilaclient.osc.v2.qos_types:UnsetQosType"
|
||||
share_qos_type_delete = "manilaclient.osc.v2.qos_types:DeleteQosType"
|
||||
|
||||
[tool.setuptools]
|
||||
packages = [
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
---
|
||||
features:
|
||||
- |
|
||||
Added support for QoS types and specs via the Python SDK and the
|
||||
``openstack share qos type`` CLI commands (create, delete, list, show,
|
||||
set, unset). Available from microversion 2.94 and above.
|
||||
Reference in New Issue
Block a user