openstack
/
python-manilaclient
Code Issues Proposed changes

[OSC] Init OpenStack Client implementation 

Add an OSC plugin for manila, the subcommand
chosen is "share". Add initial commands parser
for create, delete, list and show shares.

Also initializes basic unit testing structure for
the added commands.

Co-Authored-By: Soledad Kuczala <sol.kuczala@gmail.com>
Co-Authored-By: Victoria Martinez de la Cruz <victoria@redhat.com>

Partially-implements: bp openstack-client-support
Change-Id: I90a36f4332796675ea808e0c2a0f5ce32709624a
This commit is contained in:
Goutham Pacha Ravi 2019-03-09 03:52:23 -08:00 committed by Victoria Martinez de la Cruz
parent c07a11194f
commit ac5ca461e8
18 changed files with 2130 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lower-constraints.txt
View File

@ -44,7 +44,7 @@ openstacksdk==0.11.2
os-client-config==1.28.0
os-service-types==1.2.0
os-testr==1.0.0
osc-lib==1.8.0
osc-lib==1.10.0
oslo.concurrency==3.25.0
oslo.config==5.2.0
oslo.context==2.19.2

18
manilaclient/common/cliutils.py
View File

@ -269,3 +269,21 @@ def exit(msg=''):
if msg:
print(msg, file=sys.stderr)
sys.exit(1)
def transform_export_locations_to_string_view(export_locations):
export_locations_string_view = ''
replica_export_location_ignored_keys = (
'replica_state', 'availability_zone', 'share_replica_id')
for el in export_locations:
if hasattr(el, '_info'):
export_locations_dict = el._info
else:
export_locations_dict = el
for k, v in export_locations_dict.items():
# NOTE(gouthamr): We don't want to show replica related info
# twice in the output, so ignore those.
if k not in replica_export_location_ignored_keys:
export_locations_string_view += '\n%(k)s = %(v)s' % {
'k': k, 'v': v}
return export_locations_string_view

3
manilaclient/common/constants.py
View File

@ -77,6 +77,9 @@ TASK_STATE_DATA_COPYING_COMPLETED = 'data_copying_completed'
EXPERIMENTAL_HTTP_HEADER = 'X-OpenStack-Manila-API-Experimental'
V1_SERVICE_TYPE = 'share'
V2_SERVICE_TYPE = 'sharev2'
# Service type authority recommends using 'shared-file-system' as
# the service type. See: https://opendev.org/openstack/service-types-authority
SFS_SERVICE_TYPE = 'shared-file-system'
SERVICE_TYPES = {'1': V1_SERVICE_TYPE, '2': V2_SERVICE_TYPE}

0
manilaclient/osc/__init__.py Normal file
View File

105
manilaclient/osc/plugin.py Normal file
View File

@ -0,0 +1,105 @@
# Copyright 2019 Red Hat, 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.
"""OpenStackClient plugin for the Shared File System Service."""
import logging
from osc_lib import utils
from manilaclient import api_versions
from manilaclient.common import constants
from manilaclient import exceptions
LOG = logging.getLogger(__name__)
API_NAME = 'share'
API_VERSION_OPTION = 'os_share_api_version'
CLIENT_CLASS = 'manilaclient.v2.client.Client'
LATEST_VERSION = api_versions.MAX_VERSION
LATEST_MINOR_VERSION = api_versions.MAX_VERSION.split('.')[-1]
API_VERSIONS = {
'2.%d' % i: CLIENT_CLASS
for i in range(0, int(LATEST_MINOR_VERSION) + 1)
}
def _get_manila_url_from_service_catalog(instance):
service_type = constants.SFS_SERVICE_TYPE
url = instance.get_endpoint_for_service_type(
constants.SFS_SERVICE_TYPE, region_name=instance._region_name,
interface=instance.interface)
# Fallback if cloud is using an older service type name
if not url:
url = instance.get_endpoint_for_service_type(
constants.V2_SERVICE_TYPE, region_name=instance._region_name,
interface=instance.interface)
service_type = constants.V2_SERVICE_TYPE
if url is None:
raise exceptions.EndpointNotFound(
message="Could not find manila / shared-file-system endpoint in "
"the service catalog.")
return service_type, url
def make_client(instance):
"""Returns a shared file system service client."""
requested_api_version = instance._api_version[API_NAME]
shared_file_system_client = utils.get_client_class(
API_NAME, requested_api_version, API_VERSIONS)
# Cast the API version into an object for further processing
requested_api_version = api_versions.APIVersion(
version_str=requested_api_version)
LOG.debug('Instantiating Shared File System (share) client: %s',
shared_file_system_client)
LOG.debug('Shared File System API version: %s',
requested_api_version)
service_type, manila_endpoint_url = _get_manila_url_from_service_catalog(
instance)
instance.setup_auth()
debugging_enabled = instance._cli_options.debug
client = shared_file_system_client(session=instance.session,
service_catalog_url=manila_endpoint_url,
endpoint_type=instance.interface,
region_name=instance.region_name,
service_type=service_type,
auth=instance.auth,
http_log_debug=debugging_enabled,
api_version=requested_api_version)
return client
def build_option_parser(parser):
"""Hook to add global options."""
default_api_version = utils.env('OS_SHARE_API_VERSION') or LATEST_VERSION
parser.add_argument(
'--os-share-api-version',
metavar='<shared-file-system-api-version>',
default=default_api_version,
choices=sorted(
API_VERSIONS,
key=lambda k: [int(x) for x in k.split('.')]),
help='Shared File System API version, default=' + default_api_version +
'version supported by both the client and the server). '
'(Env: OS_SHARE_API_VERSION)',
)
return parser

35
manilaclient/osc/utils.py Normal file
View File

@ -0,0 +1,35 @@
# Copyright 2019 Red Hat, 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.
from manilaclient import exceptions
def extract_key_value_options(pairs):
result_dict = {}
duplicate_options = []
pairs = pairs or {}
for attr, value in pairs.items():
if attr not in result_dict:
result_dict[attr] = value
else:
duplicate_options.append(attr)
if pairs and len(duplicate_options) > 0:
duplicate_str = ', '.join(duplicate_options)
msg = "Following options were duplicated: %s" % duplicate_str
raise exceptions.CommandError(msg)
return result_dict

0
manilaclient/osc/v2/__init__.py Normal file
View File

545
manilaclient/osc/v2/share.py Normal file
View File

@ -0,0 +1,545 @@
# Copyright 2019 Red Hat, 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 logging
from openstackclient.identity import common as identity_common
from osc_lib.cli import format_columns
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 cliutils
from manilaclient.osc import utils
LOG = logging.getLogger(__name__)
SHARE_ATTRIBUTES = [
'id',
'name',
'size',
'share_proto',
'status',
'is_public',
'share_type_name',
'availability_zone',
'description',
'share_network_id',
'share_server_id',
'share_type',
'share_group_id',
'host',
'user_id',
'project_id',
'access_rules_status',
'snapshot_id',
'snapshot_support',
'create_share_from_snapshot_support',
'mount_snapshot_support',
'revert_to_snapshot_support',
'task_state',
'source_share_group_snapshot_member_id',
'replication_type',
'has_replicas',
'created_at',
'metadata'
]
SHARE_ATTRIBUTES_HEADERS = [
'ID',
'Name',
'Size',
'Share Protocol',
'Status',
'Is Public',
'Share Type Name',
'Availability Zone',
'Description',
'Share Network ID',
'Share Server ID',
'Share Type',
'Share Group ID',
'Host',
'User ID',
'Project ID',
'Access Rules Status',
'Source Snapshot ID',
'Supports Creating Snapshots',
'Supports Cloning Snapshots',
'Supports Mounting snapshots',
'Supports Reverting to Snapshot',
'Migration Task Status',
'Source Share Group Snapshot Member ID',
'Replication Type',
'Has Replicas',
'Created At',
'Properties',
]
class CreateShare(command.ShowOne):
"""Create a new share."""
_description = _("Create new share")
log = logging.getLogger(__name__ + ".CreateShare")
def get_parser(self, prog_name):
parser = super(CreateShare, self).get_parser(prog_name)
parser.add_argument(
'share_proto',
metavar="<share_protocol>",
help=_('Share protocol (NFS, CIFS, CephFS, GlusterFS or HDFS)')
)
parser.add_argument(
'size',
metavar="<size>",
type=int,
help=_('Share size in GiB.')
)
parser.add_argument(
'--name',
metavar="<name>",
default=None,
help=_('Optional share name. (Default=None)')
)
parser.add_argument(
'--snapshot-id',
metavar="<snapshot-id>",
default=None,
help=_("Optional snapshot ID to create the share from."
" (Default=None)")
)
# NOTE(vkmc) --property replaces --metadata in osc
parser.add_argument(
"--property",
metavar="<key=value>",
default={},
action=parseractions.KeyValueAction,
help=_("Set a property to this share "
"(repeat option to set multiple properties)"),
)
parser.add_argument(
'--share-network',
metavar='<network-info>',
default=None,
help=_('Optional network info ID or name.'),
)
parser.add_argument(
'--description',
metavar='<description>',
default=None,
help=_('Optional share description. (Default=None)')
)
parser.add_argument(
'--public',
default=False,
help=_('Level of visibility for share. '
'Defines whether other tenants are able to see it or not.')
)
parser.add_argument(
'--share-type',
metavar='<share-type>',
default=None,
help=_('Optional share type. Use of optional shares type '
'is deprecated. (Default=Default)')
)
parser.add_argument(
'--availability-zone',
metavar='<availability-zone>',
default=None,
help=_('Availability zone in which share should be created.')
)
parser.add_argument(
'--share-group',
metavar='<share-group>',
default=None,
help=_('Optional share group name or ID in which to create '
'the share. (Experimental, Default=None).')
)
return parser
def take_action(self, parsed_args):
# TODO(s0ru): the table shows 'Field', 'Value'
share_client = self.app.client_manager.share
share_type = None
if parsed_args.share_type:
share_type = apiutils.find_resource(share_client.share_types,
parsed_args.share_type).id
share_network = None
if parsed_args.share_network:
share_network = apiutils.find_resource(
share_client.share_networks,
parsed_args.share_network).id
share_group = None
if parsed_args.share_group:
share_group = apiutils.find_resource(share_client.share_groups,
parsed_args.share_group).id
size = parsed_args.size
snapshot_id = None
if parsed_args.snapshot_id:
snapshot = apiutils.find_resource(share_client.share_snapshots,
parsed_args.snapshot_id)
size = max(size or 0, snapshot.size)
body = {
'share_proto': parsed_args.share_proto,
'size': parsed_args.size,
'snapshot_id': snapshot_id,
'name': parsed_args.name,
'description': parsed_args.description,
'metadata': parsed_args.property,
'share_network': share_network,
'share_type': share_type,
'is_public': parsed_args.public,
'availability_zone': parsed_args.availability_zone,
'share_group_id': share_group
}
share = share_client.shares.create(**body)
printable_share = share._info
printable_share.pop('links', None)
printable_share.pop('shares_type', None)
return self.dict2columns(printable_share)
class DeleteShare(command.Command):
"""Delete a share."""
_description = _("Delete a share")
def get_parser(self, prog_name):
parser = super(DeleteShare, self).get_parser(prog_name)
parser.add_argument(
"shares",
metavar="<share>",
nargs="+",
help=_("Share(s) to delete (name or ID)")
)
parser.add_argument(
"--share-group",
metavar="<share-group>",
default=None,
help=_("Optional share group (name or ID)"
"which contains the share")
)
parser.add_argument(
"--force",
action='store_true',
default=False,
help=_("Attempt forced removal of share(s), regardless of state "
"(defaults to False)")
)
return parser
def take_action(self, parsed_args):
share_client = self.app.client_manager.share
result = 0
for share in parsed_args.shares:
try:
share_obj = apiutils.find_resource(
share_client.shares, share
)
share_group_id = (share_group_id if parsed_args.share_group
else None)
if parsed_args.force:
share_client.shares.force_delete(share_obj)
else:
share_client.shares.delete(share_obj,
share_group_id)
except Exception as exc:
result += 1
LOG.error(_("Failed to delete share with "
"name or ID '%(share)s': %(e)s"),
{'share': share, 'e': exc})
if result > 0:
total = len(parsed_args.shares)
msg = (_("%(result)s of %(total)s shares failed "
"to delete.") % {'result': result, 'total': total})
raise exceptions.CommandError(msg)
class ListShare(command.Lister):
"""List Shared file systems (shares)."""
_description = _("List shares")
def get_parser(self, prog_name):
parser = super(ListShare, self).get_parser(prog_name)
parser.add_argument(
'--name',
metavar="<share-name>",
help=_('Filter shares by share name')
)
parser.add_argument(
'--status',
metavar="<share-status>",
help=_('Filter shares by status')
)
parser.add_argument(
'--snapshot',
metavar='<share-network-id>',
help=_('Filter shares by snapshot name or id.'),
)
parser.add_argument(
'--public',
action='store_true',
default=False,
help=_('Include public shares'),
)
parser.add_argument(
'--share-network',
metavar='<share-network-name-or-id>',
help=_('Filter shares exported on a given share network'),
)
parser.add_argument(
'--share-type',
metavar='<share-type-name-or-id>',
help=_('Filter shares of a given share type'),
)
parser.add_argument(
'--share-group',
metavar='<share-group-name-or-id>',
help=_('Filter shares belonging to a given share group'),
)
parser.add_argument(
'--host',
metavar='<share-host>',
help=_('Filter shares belonging to a given host (admin only)'),
)
parser.add_argument(
'--share-server',
metavar='<share-server-id>',
help=_('Filter shares exported via a given share server '
'(admin only)'),
)
parser.add_argument(
'--project',
metavar='<project>',
help=_('Filter shares by project (name or ID) (admin only)')
)
identity_common.add_project_domain_option_to_parser(parser)
parser.add_argument(
'--user',
metavar='<user>',
help=_('Filter results by user (name or ID) (admin only)')
)
identity_common.add_user_domain_option_to_parser(parser)
parser.add_argument(
'--all-projects',
action='store_true',
default=False,
help=_('Include all projects (admin only)'),
)
parser.add_argument(
'--property',
metavar='<key=value>',
action=parseractions.KeyValueAction,
help=_('Filter shares having a given metadata key=value property '
'(repeat option to filter by multiple properties)'),
)
parser.add_argument(
'--extra-spec',
metavar='<key=value>',
action=parseractions.KeyValueAction,
help=_('Filter shares with extra specs (key=value) of the share '
'type that they belong to. '
'(repeat option to filter by multiple extra specs)'),
)
parser.add_argument(
'--long',
action='store_true',
default=False,
help=_('List additional fields in output'),
)
parser.add_argument(
'--sort',
metavar="<key>[:<direction>]",
default='name:asc',
help=_("Sort output by selected keys and directions(asc or desc) "
"(default: name:asc), multiple keys and directions can be "
"specified separated by comma"),
)
parser.add_argument(
'--limit',
metavar="<num-shares>",
type=int,
action=parseractions.NonNegativeAction,
help=_('Maximum number of shares to display'),
)
parser.add_argument(
'--marker',
metavar="<share>",
help=_('The last share ID of the previous page'),
)
return parser
def take_action(self, parsed_args):
share_client = self.app.client_manager.share
identity_client = self.app.client_manager.identity
# TODO(gouthamr): Add support for ~name, ~description
# export_location filtering
if parsed_args.long:
columns = SHARE_ATTRIBUTES
column_headers = SHARE_ATTRIBUTES_HEADERS
else:
columns = [
'id',
'name',
'size',
'share_proto',
'status',
'is_public',
'share_type_name',
'host',
'availability_zone'
]
column_headers = [
'ID',
'Name',
'Size',
'Share Proto',
'Status',
'Is Public',
'Share Type Name',
'Host',
'Availability Zone'
]
project_id = None
if parsed_args.project:
project_id = identity_common.find_project(
identity_client,
parsed_args.project,
parsed_args.project_domain).id
user_id = None
if parsed_args.user:
user_id = identity_common.find_user(identity_client,
parsed_args.user,
parsed_args.user_domain).id
# set value of 'all_projects' when using project option
all_projects = bool(parsed_args.project) or parsed_args.all_projects
share_type_id = None
if parsed_args.share_type:
share_type_id = apiutils.find_resource(share_client.share_types,
parsed_args.share_type).id
snapshot_id = None
if parsed_args.snapshot:
snapshot_id = apiutils.find_resource(share_client.share_snapshots,
parsed_args.snapshot).id
share_network_id = None
if parsed_args.share_network:
share_network_id = apiutils.find_resource(
share_client.share_networks,
parsed_args.share_network).id
share_group_id = None
if parsed_args.share_group:
share_group_id = apiutils.find_resource(share_client.share_groups,
parsed_args.share_group).id
share_server_id = None
if parsed_args.share_server:
share_server_id = apiutils.find_resource(
share_client.share_servers,
parsed_args.share_server).id
search_opts = {
'all_projects': all_projects,
'is_public': parsed_args.public,
'metadata': utils.extract_key_value_options(
parsed_args.property),
'extra_specs': utils.extract_key_value_options(
parsed_args.extra_spec),
'limit': parsed_args.limit,
'name': parsed_args.name,
'status': parsed_args.status,
'host': parsed_args.host,
'share_server_id': share_server_id,
'share_network_id': share_network_id,
'share_type_id': share_type_id,
'snapshot_id': snapshot_id,
'share_group_id': share_group_id,
'project_id': project_id,
'user_id': user_id,
'offset': parsed_args.marker,
'limit': parsed_args.limit,
}
# NOTE(vkmc) We implemented sorting and filtering in manilaclient
# but we will use the one provided by osc
data = share_client.shares.list(search_opts=search_opts)
data = oscutils.sort_items(data, parsed_args.sort, str)
return (column_headers, (oscutils.get_item_properties
(s, columns, formatters={'Metadata': oscutils.format_dict},)
for s in data))
class ShowShare(command.ShowOne):
"""Show a share."""
_description = _("Display share details")
def get_parser(self, prog_name):
parser = super(ShowShare, self).get_parser(prog_name)
parser.add_argument(
'share',
metavar="<share>",
help=_('Share to display (name or ID)')
)
return parser
def take_action(self, parsed_args):
share_client = self.app.client_manager.share
share_obj = apiutils.find_resource(share_client.shares,
parsed_args.share)
export_locations = share_client.share_export_locations.list(share_obj)
export_locations = (
cliutils.transform_export_locations_to_string_view(
export_locations))
data = share_obj._info
data['export_locations'] = export_locations
# Special mapping for columns to make the output easier to read:
# 'metadata' --> 'properties'
data.update(
{
'properties':
format_columns.DictColumn(data.pop('metadata', {})),
},
)
# Remove key links from being displayed
data.pop("links", None)
data.pop("shares_type", None)
return self.dict2columns(data)

0
manilaclient/tests/osc/__init__.py Normal file
View File

0
manilaclient/tests/osc/unit/__init__.py Normal file
View File

249
manilaclient/tests/osc/unit/osc_fakes.py Normal file
View File

@ -0,0 +1,249 @@
# Copyright 2013 Nebula 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 oslo_serialization import jsonutils
import sys
from keystoneauth1 import fixture
import requests
AUTH_TOKEN = "foobar"
AUTH_URL = "http://0.0.0.0"
USERNAME = "itchy"
PASSWORD = "scratchy"
PROJECT_NAME = "poochie"
REGION_NAME = "richie"
INTERFACE = "catchy"
VERSION = "3"
TEST_RESPONSE_DICT = fixture.V2Token(token_id=AUTH_TOKEN,
user_name=USERNAME)
_s = TEST_RESPONSE_DICT.add_service('identity', name='keystone')
_s.add_endpoint(AUTH_URL + ':5000/v2.0')
_s = TEST_RESPONSE_DICT.add_service('network', name='neutron')
_s.add_endpoint(AUTH_URL + ':9696')
_s = TEST_RESPONSE_DICT.add_service('compute', name='nova')
_s.add_endpoint(AUTH_URL + ':8774/v2.1')
_s = TEST_RESPONSE_DICT.add_service('image', name='glance')
_s.add_endpoint(AUTH_URL + ':9292')
_s = TEST_RESPONSE_DICT.add_service('object', name='swift')
_s.add_endpoint(AUTH_URL + ':8080/v1')
TEST_RESPONSE_DICT_V3 = fixture.V3Token(user_name=USERNAME)
TEST_RESPONSE_DICT_V3.set_project_scope()
TEST_VERSIONS = fixture.DiscoveryList(href=AUTH_URL)
def to_unicode_dict(catalog_dict):
"""Converts dict to unicode dict"""
if isinstance(catalog_dict, dict):
return {to_unicode_dict(key): to_unicode_dict(value)
for key, value in catalog_dict.items()}
elif isinstance(catalog_dict, list):
return [to_unicode_dict(element) for element in catalog_dict]
elif isinstance(catalog_dict, str):
return catalog_dict + u""
else:
return catalog_dict
class FakeStdout(object):
def __init__(self):
self.content = []
def write(self, text):
self.content.append(text)
def make_string(self):
result = ''
for line in self.content:
result = result + line
return result
class FakeLog(object):
def __init__(self):
self.messages = {}
def debug(self, msg):
self.messages['debug'] = msg
def info(self, msg):
self.messages['info'] = msg
def warning(self, msg):
self.messages['warning'] = msg
def error(self, msg):
self.messages['error'] = msg
def critical(self, msg):
self.messages['critical'] = msg
class FakeApp(object):
def __init__(self, _stdout, _log):
self.stdout = _stdout
self.client_manager = None
self.stdin = sys.stdin
self.stdout = _stdout or sys.stdout
self.stderr = sys.stderr
self.log = _log
class FakeOptions(object):
def __init__(self, **kwargs):
self.os_beta_command = False
class FakeClient(object):
def __init__(self, **kwargs):
self.endpoint = kwargs['endpoint']
self.token = kwargs['token']
class FakeClientManager(object):
_api_version = {
'image': '2',
}
def __init__(self):
self.compute = None
self.identity = None
self.image = None
self.object_store = None
self.volume = None
self.network = None
self.session = None
self.auth_ref = None
self.auth_plugin_name = None
self.network_endpoint_enabled = True
def get_configuration(self):
return {
'auth': {
'username': USERNAME,
'password': PASSWORD,
'token': AUTH_TOKEN,
},
'region': REGION_NAME,
'identity_api_version': VERSION,
}
def is_network_endpoint_enabled(self):
return self.network_endpoint_enabled
class FakeModule(object):
def __init__(self, name, version):
self.name = name
self.__version__ = version
# Workaround for openstacksdk case
self.version = mock.Mock()
self.version.__version__ = version
class FakeResource(object):
def __init__(self, manager=None, info=None, loaded=False, methods=None):
"""Set attributes and methods for a resource.
:param manager:
The resource manager
:param Dictionary info:
A dictionary with all attributes
:param bool loaded:
True if the resource is loaded in memory
:param Dictionary methods:
A dictionary with all methods
"""
info = info or {}
methods = methods or {}
self.__name__ = type(self).__name__
self.manager = manager
self._info = info
self._add_details(info)
self._add_methods(methods)
self._loaded = loaded
def _add_details(self, info):
for (k, v) in info.items():
setattr(self, k, v)
def _add_methods(self, methods):
"""Fake methods with MagicMock objects.
For each <@key, @value> pairs in methods, add an callable MagicMock
object named @key as an attribute, and set the mock's return_value to
@value. When users access the attribute with (), @value will be
returned, which looks like a function call.
"""
for (name, ret) in methods.items():
method = mock.Mock(return_value=ret)
setattr(self, name, method)
def __repr__(self):
reprkeys = sorted(k for k in self.__dict__.keys() if k[0] != '_' and
k != 'manager')
info = ", ".join("%s=%s" % (k, getattr(self, k)) for k in reprkeys)
return "<%s %s>" % (self.__class__.__name__, info)
def keys(self):
return self._info.keys()
def to_dict(self):
return self._info
@property
def info(self):
return self._info
def __getitem__(self, item):
return self._info.get(item)
def get(self, item, default=None):
return self._info.get(item, default)
class FakeResponse(requests.Response):
def __init__(self, headers=None, status_code=200,
data=None, encoding=None):
super(FakeResponse, self).__init__()
headers = headers or {}
self.status_code = status_code
self.headers.update(headers)
self._content = jsonutils.dump_as_bytes(data)
class FakeModel(dict):
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)

75
manilaclient/tests/osc/unit/osc_utils.py Normal file
View File

@ -0,0 +1,75 @@
# Copyright 2012-2013 OpenStack Foundation
# Copyright 2013 Nebula 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 fixtures
import os
import testtools
from openstackclient.tests.unit import fakes
class ParserException(Exception):
pass
class TestCase(testtools.TestCase):
def setUp(self):
testtools.TestCase.setUp(self)
if (os.environ.get("OS_STDOUT_CAPTURE") == "True" or
os.environ.get("OS_STDOUT_CAPTURE") == "1"):
stdout = self.useFixture(fixtures.StringStream("stdout")).stream
self.useFixture(fixtures.MonkeyPatch("sys.stdout", stdout))
if (os.environ.get("OS_STDERR_CAPTURE") == "True" or
os.environ.get("OS_STDERR_CAPTURE") == "1"):
stderr = self.useFixture(fixtures.StringStream("stderr")).stream
self.useFixture(fixtures.MonkeyPatch("sys.stderr", stderr))
def assertNotCalled(self, m, msg=None):
"""Assert a function was not called"""
if m.called:
if not msg:
msg = 'method %s should not have been called' % m
self.fail(msg)
class TestCommand(TestCase):
"""Test cliff command classes"""
def setUp(self):
super(TestCommand, self).setUp()
# Build up a fake app
self.fake_stdout = fakes.FakeStdout()
self.fake_log = fakes.FakeLog()
self.app = fakes.FakeApp(self.fake_stdout, self.fake_log)
self.app.client_manager = fakes.FakeClientManager()
self.app.options = fakes.FakeOptions()
def check_parser(self, cmd, args, verify_args):
cmd_parser = cmd.get_parser('check_parser')
try:
parsed_args = cmd_parser.parse_args(args)
except SystemExit:
raise ParserException("Argument parse failed")
for av in verify_args:
attr, value = av
if attr:
self.assertIn(attr, parsed_args)
self.assertEqual(value, getattr(parsed_args, attr))
return parsed_args

0
manilaclient/tests/osc/unit/v2/__init__.py Normal file
View File

229
manilaclient/tests/osc/unit/v2/fakes.py Normal file
View File

@ -0,0 +1,229 @@
# 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 random
import uuid
from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes
from manilaclient.tests.osc.unit import osc_fakes
from manilaclient.tests.osc.unit import osc_utils
class FakeShareClient(object):
def __init__(self, **kwargs):
super(FakeShareClient, self).__init__()
self.auth_token = kwargs['token']
self.management_url = kwargs['endpoint']
self.shares = mock.Mock()
self.shares.resource_class = osc_fakes.FakeResource(None, {})
class ManilaParseException(Exception):
"""The base exception class for all exceptions this library raises."""
def __init__(self, message=None, details=None):
self.message = message or "Argument parse exception"
self.details = details or None
def __str__(self):
return self.message
class TestShare(osc_utils.TestCommand):
def setUp(self):
super(TestShare, self).setUp()
self.app.client_manager.share = FakeShareClient(
endpoint=osc_fakes.AUTH_URL,
token=osc_fakes.AUTH_TOKEN
)
self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
endpoint=osc_fakes.AUTH_URL,
token=osc_fakes.AUTH_TOKEN
)
class FakeShare(object):
"""Fake one or more shares."""
@staticmethod
def create_one_share(attrs=None):
"""Create a fake share.
:param Dictionary attrs:
A dictionary with all attributes
:return:
A FakeResource object, with flavor_id, image_id, and so on
"""
attrs = attrs or {}
# set default attributes.
share_info = {
"status": None,
"share_server_id": None,
"project_id": 'project-id-' + uuid.uuid4().hex,
"name": 'share-name-' + uuid.uuid4().hex,
"share_type": 'share-type-' + uuid.uuid4().hex,
"share_type_name": "default",
"availability_zone": None,
"created_at": 'time-' + uuid.uuid4().hex,
"share_network_id": None,
"share_group_id": None,
"share_proto": "NFS",
"host": None,
"access_rules_status": "active",
"has_replicas": False,
"replication_type": None,
"task_state": None,
"snapshot_support": True,
"snapshot_id": None,
"is_public": True,
"metadata": {},
"id": 'share-id-' + uuid.uuid4().hex,
"size": random.randint(1, 20),
"description": 'share-description-' + uuid.uuid4().hex,
"user_id": 'share-user-id-' + uuid.uuid4().hex,
"create_share_from_snapshot_support": False,
"mount_snapshot_support": False,
"revert_to_snapshot_support": False,
"source_share_group_snapshot_member_id": None,
}
# Overwrite default attributes.
share_info.update(attrs)
share = osc_fakes.FakeResource(info=copy.deepcopy(share_info),
loaded=True)
return share
@staticmethod
def create_shares(attrs=None, count=2):
"""Create multiple fake shares.
:param Dictionary attrs:
A dictionary with all share attributes
:param Integer count:
The number of shares to be faked
:return:
A list of FakeResource objects
"""
shares = []
for n in range(0, count):
shares.append(FakeShare.create_one_share(attrs))
return shares
@staticmethod
def get_shares(shares=None, count=2):
"""Get an iterable MagicMock object with a list of faked shares.
If a shares list is provided, then initialize the Mock object with the
list. Otherwise create one.
:param List shares:
A list of FakeResource objects faking shares
:param Integer count:
The number of shares to be faked
:return
An iterable Mock object with side_effect set to a list of faked
shares
"""
if shares is None:
shares = FakeShare.create_shares(count)
return mock.Mock(side_effect=shares)
@staticmethod
def get_share_columns(share=None):
"""Get the shares columns from a faked shares object.
:param shares:
A FakeResource objects faking shares
:return
A tuple which may include the following keys:
('id', 'name', 'description', 'status', 'size', 'share_type',
'metadata', 'snapshot', 'availability_zone')
"""
if share is not None:
return tuple(k for k in sorted(share.keys()))
return tuple([])
@staticmethod
def get_share_data(share=None):
"""Get the shares data from a faked shares object.
:param shares:
A FakeResource objects faking shares
:return
A tuple which may include the following values:
('ce26708d', 'fake name', 'fake description', 'available',
20, 'fake share type', "Manila='zorilla', Zorilla='manila',
Zorilla='zorilla'", 1, 'nova')
"""
data_list = []
if share is not None:
for x in sorted(share.keys()):
if x == 'tags':
# The 'tags' should be format_list
data_list.append(
format_columns.ListColumn(share.info.get(x)))
else:
data_list.append(share.info.get(x))
return tuple(data_list)
class FakeShareType(object):
"""Fake one or more share types"""
@staticmethod
def create_one_sharetype(attrs=None):
"""Create a fake share type
:param Dictionary attrs:
A dictionary with all attributes
:return:
A FakeResource object, with project_id, resource and so on
"""
attrs = attrs or {}
share_type_info = {
"required_extra_specs": {
"driver_handles_share_servers": True
},
"share_type_access:is_public": True,
"extra_specs": {
"replication_type": "readable",
"driver_handles_share_servers": True,
"mount_snapshot_support": False,
"revert_to_snapshot_support": False,
"create_share_from_snapshot_support": True,
"snapshot_support": True
},
"id": 'share-type-id-' + uuid.uuid4().hex,
"name": 'share-type-name-' + uuid.uuid4().hex,
"is_default": False,
"description": 'share-type-description-' + uuid.uuid4().hex
}
share_type_info.update(attrs)
share_type = osc_fakes.FakeResource(info=copy.deepcopy(
share_type_info),
loaded=True)
return share_type

856
manilaclient/tests/osc/unit/v2/test_share.py Normal file
View File

@ -0,0 +1,856 @@
# Copyright 2019 Red Hat 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 argparse
import mock
from mock import call
from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes
from manilaclient.osc.v2 import share as osc_shares
from manilaclient.tests.osc.unit import osc_utils
from manilaclient.tests.osc.unit.v2 import fakes as manila_fakes
class TestShare(manila_fakes.TestShare):
def setUp(self):
super(TestShare, self).setUp()
self.shares_mock = self.app.client_manager.share.shares
self.shares_mock.reset_mock()
self.projects_mock = self.app.client_manager.identity.projects
self.projects_mock.reset_mock()
self.users_mock = self.app.client_manager.identity.users
self.users_mock.reset_mock()
def setup_shares_mock(self, count):
shares = manila_fakes.FakeShare.create_shares(count=count)
self.shares_mock.get = manila_fakes.FakeShare.get_shares(
shares,
0)
return shares
class TestShareCreate(TestShare):
def setUp(self):
super(TestShareCreate, self).setUp()
self.new_share = manila_fakes.FakeShare.create_one_share()
self.shares_mock.create.return_value = self.new_share
# Get the command object to test
self.cmd = osc_shares.CreateShare(self.app, None)
self.datalist = tuple(self.new_share._info.values())
self.columns = tuple(self.new_share._info.keys())
def test_share_create_required_args(self):
"""Verifies required arguments."""
arglist = [
self.new_share.share_proto,
str(self.new_share.size),
]
verifylist = [
('share_proto', self.new_share.share_proto),
('size', self.new_share.size)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.shares_mock.create.assert_called_with(
availability_zone=None,
description=None,
is_public=False,
metadata={},
name=None,
share_group_id=None,
share_network=None,
share_proto=self.new_share.share_proto,
share_type=None,
size=self.new_share.size,
snapshot_id=None
)
self.assertCountEqual(self.columns, columns)
self.assertCountEqual(self.datalist, data)
def test_share_create_missing_required_arg(self):
"""Verifies missing required arguments."""
arglist = [
self.new_share.share_proto,
]
verifylist = [
('share_proto', self.new_share.share_proto)
]
self.assertRaises(osc_utils.ParserException,
self.check_parser, self.cmd, arglist, verifylist)
def test_share_create_metadata(self):
arglist = [
self.new_share.share_proto,
str(self.new_share.size),
'--property', 'Manila=zorilla',
'--property', 'Zorilla=manila'
]
verifylist = [
('share_proto', self.new_share.share_proto),
('size', self.new_share.size),
('property', {'Manila': 'zorilla', 'Zorilla': 'manila'}),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.shares_mock.create.assert_called_with(
availability_zone=None,
description=None,
is_public=False,
metadata={'Manila': 'zorilla', 'Zorilla': 'manila'},
name=None,
share_group_id=None,
share_network=None,
share_proto=self.new_share.share_proto,
share_type=None,
size=self.new_share.size,
snapshot_id=None
)
self.assertCountEqual(self.columns, columns)
self.assertCountEqual(self.datalist, data)
# TODO(vkmc) Add test with snapshot when
# we implement snapshot support in OSC
# def test_share_create_with_snapshot(self):
class TestShareDelete(TestShare):
def setUp(self):
super(TestShareDelete, self).setUp()
self.shares_mock.delete = mock.Mock()
self.shares_mock.delete.return_value = None
# Get the command object to test
self.cmd = osc_shares.DeleteShare(self.app, None)
def test_share_delete_one(self):
shares = self.setup_shares_mock(count=1)
arglist = [
shares[0].name
]
verifylist = [
("force", False),
("share_group", None),
('shares', [shares[0].name])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.shares_mock.delete.assert_called_with(shares[0].name, None)
self.assertIsNone(result)
def test_share_delete_many(self):
shares = self.setup_shares_mock(count=3)
arglist = [v.id for v in shares]
verifylist = [
("force", False),
("share_group", None),
('shares', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = [call(s.name, None) for s in shares]
self.shares_mock.delete.assert_has_calls(calls)
self.assertIsNone(result)
def test_share_delete_with_force(self):
shares = self.setup_shares_mock(count=1)
arglist = [
'--force',
shares[0].name,
]
verifylist = [
('force', True),
("share_group", None),
('shares', [shares[0].name]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.shares_mock.force_delete.assert_called_once_with(shares[0].name)
self.assertIsNone(result)
def test_share_delete_wrong_name(self):
shares = self.setup_shares_mock(count=1)
arglist = [
shares[0].name + '-wrong-name'
]
verifylist = [
("force", False),
("share_group", None),
('shares', [shares[0].name + '-wrong-name'])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.assertIsNone(result)
def test_share_delete_no_name(self):
# self.setup_shares_mock(count=1)
arglist = []
verifylist = [
("force", False),
("share_group", None),
('shares', '')
]
self.assertRaises(osc_utils.ParserException,
self.check_parser, self.cmd, arglist, verifylist)
class TestShareList(TestShare):
project = identity_fakes.FakeProject.create_one_project()
user = identity_fakes.FakeUser.create_one_user()
columns = [
'ID',
'Name',
'Size',
'Share Proto',
'Status',
'Is Public',
'Share Type Name',
'Host',
'Availability Zone'
]
def setUp(self):
super(TestShareList, self).setUp()
self.new_share = manila_fakes.FakeShare.create_one_share()
self.shares_mock.list.return_value = [self.new_share]
self.users_mock.get.return_value = self.user
self.projects_mock.get.return_value = self.project
# Get the command object to test
self.cmd = osc_shares.ListShare(self.app, None)
def _get_data(self):
data = ((
self.new_share.id,
self.new_share.name,
self.new_share.size,
self.new_share.share_proto,
self.new_share.status,
self.new_share.is_public,
self.new_share.share_type_name,
self.new_share.host,
self.new_share.availability_zone,
),)
return data
def _get_search_opts(self):
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
return search_opts
def test_share_list_no_options(self):
arglist = []
verifylist = [
('long', False),
('all_projects', False),
('name', None),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_project(self):
arglist = [
'--project', self.project.name,
]
verifylist = [
('project', self.project.name),
('long', False),
('all_projects', False),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['project_id'] = self.project.id
search_opts['all_projects'] = True
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_project_domain(self):
arglist = [
'--project', self.project.name,
'--project-domain', self.project.domain_id,
]
verifylist = [
('project', self.project.name),
('project_domain', self.project.domain_id),
('long', False),
('all_projects', False),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['project_id'] = self.project.id
search_opts['all_projects'] = True
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_user(self):
arglist = [
'--user', self.user.name,
]
verifylist = [
('user', self.user.name),
('long', False),
('all_projects', False),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['user_id'] = self.user.id
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_user_domain(self):
arglist = [
'--user', self.user.name,
'--user-domain', self.user.domain_id,
]
verifylist = [
('user', self.user.name),
('user_domain', self.user.domain_id),
('long', False),
('all_projects', False),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['user_id'] = self.user.id
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_name(self):
arglist = [
'--name', self.new_share.name,
]
verifylist = [
('long', False),
('all_projects', False),
('name', self.new_share.name),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['name'] = self.new_share.name
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_status(self):
arglist = [
'--status', self.new_share.status,
]
verifylist = [
('long', False),
('all_projects', False),
('name', None),
('status', self.new_share.status),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['status'] = self.new_share.status
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_all_projects(self):
arglist = [
'--all-projects',
]
verifylist = [
('long', False),
('all_projects', True),
('name', None),
('status', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
search_opts['all_projects'] = True
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
self.assertEqual(self.columns, cmd_columns)
data = self._get_data()
self.assertEqual(data, tuple(cmd_data))
def test_share_list_long(self):
arglist = [
'--long',
]
verifylist = [
('long', True),
('all_projects', False),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': None,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': None,
}
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts,
)
collist = [
'ID',
'Name',
'Size',
'Share Protocol',
'Status',
'Is Public',
'Share Type Name',
'Availability Zone',
'Description',
'Share Network ID',
'Share Server ID',
'Share Type',
'Share Group ID',
'Host',
'User ID',
'Project ID',
'Access Rules Status',
'Source Snapshot ID',
'Supports Creating Snapshots',
'Supports Cloning Snapshots',
'Supports Mounting snapshots',
'Supports Reverting to Snapshot',
'Migration Task Status',
'Source Share Group Snapshot Member ID',
'Replication Type',
'Has Replicas',
'Created At',
'Properties',
]
self.assertEqual(collist, cmd_columns)
data = ((
self.new_share.id,
self.new_share.name,
self.new_share.size,
self.new_share.share_proto,
self.new_share.status,
self.new_share.is_public,
self.new_share.share_type_name,
self.new_share.availability_zone,
self.new_share.description,
self.new_share.share_network_id,
self.new_share.share_server_id,
self.new_share.share_type,
self.new_share.share_group_id,
self.new_share.host,
self.new_share.user_id,
self.new_share.project_id,
self.new_share.access_rules_status,
self.new_share.snapshot_id,
self.new_share.snapshot_support,
self.new_share.create_share_from_snapshot_support,
self.new_share.mount_snapshot_support,
self.new_share.revert_to_snapshot_support,
self.new_share.task_state,
self.new_share.source_share_group_snapshot_member_id,
self.new_share.replication_type,
self.new_share.has_replicas,
self.new_share.created_at,
self.new_share.metadata
),)
self.assertEqual(data, tuple(cmd_data))
def test_share_list_with_marker_and_limit(self):
arglist = [
"--marker", self.new_share.id,
"--limit", "2",
]
verifylist = [
('long', False),
('all_projects', False),
('name', None),
('status', None),
('marker', self.new_share.id),
('limit', 2),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cmd_columns, cmd_data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, cmd_columns)
search_opts = {
'all_projects': False,
'is_public': False,
'metadata': {},
'extra_specs': {},
'limit': 2,
'name': None,
'status': None,
'host': None,
'share_server_id': None,
'share_network_id': None,
'share_type_id': None,
'snapshot_id': None,
'share_group_id': None,
'project_id': None,
'user_id': None,
'offset': self.new_share.id
}
data = self._get_data()
self.shares_mock.list.assert_called_once_with(
search_opts=search_opts
)
self.assertEqual(data, tuple(cmd_data))
def test_share_list_negative_limit(self):
arglist = [
"--limit", "-2",
]
verifylist = [
("limit", -2),
]
self.assertRaises(argparse.ArgumentTypeError, self.check_parser,
self.cmd, arglist, verifylist)
class TestShareShow(TestShare):
def setUp(self):
super(TestShareShow, self).setUp()
self._share = manila_fakes.FakeShare.create_one_share()
self.shares_mock.get.return_value = self._share
# Get the command object to test
self.cmd = osc_shares.ShowShare(self.app, None)
def test_share_show(self):
arglist = [
self._share.id
]
verifylist = [
("share", self._share.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.shares_mock.get.assert_called_with(self._share.id)
self.assertEqual(
manila_fakes.FakeShare.get_share_columns(self._share),
columns)
self.assertEqual(
manila_fakes.FakeShare.get_share_data(self._share),
data)

26
manilaclient/v2/shell.py
View File

@ -67,24 +67,6 @@ def _find_share(cs, share):
return apiclient_utils.find_resource(cs.shares, share)
def _transform_export_locations_to_string_view(export_locations):
export_locations_string_view = ''
replica_export_location_ignored_keys = (
'replica_state', 'availability_zone', 'share_replica_id')
for el in export_locations:
if hasattr(el, '_info'):
export_locations_dict = el._info
else:
export_locations_dict = el
for k, v in export_locations_dict.items():
# NOTE(gouthamr): We don't want to show replica related info
# twice in the output, so ignore those.
if k not in replica_export_location_ignored_keys:
export_locations_string_view += '\n%(k)s = %(v)s' % {
'k': k, 'v': v}
return export_locations_string_view
@api_versions.wraps("1.0", "2.8")
def _print_share(cs, share):
info = share._info.copy()
@ -142,7 +124,7 @@ def _print_share(cs, share):
# +-------------------+--------------------------------------------+
if info.get('export_locations'):
info['export_locations'] = (
_transform_export_locations_to_string_view(
cliutils.transform_export_locations_to_string_view(
info['export_locations']))
# No need to print both volume_type and share_type to CLI
@ -191,7 +173,7 @@ def _print_share_instance(cs, instance):
info.pop('links', None)
if info.get('export_locations'):
info['export_locations'] = (
_transform_export_locations_to_string_view(
cliutils.transform_export_locations_to_string_view(
info['export_locations']))
cliutils.print_dict(info)
@ -214,7 +196,7 @@ def _print_share_replica(cs, replica):
info.pop('links', None)
if info.get('export_locations'):
info['export_locations'] = (
_transform_export_locations_to_string_view(
cliutils.transform_export_locations_to_string_view(
info['export_locations']))
cliutils.print_dict(info)
@ -267,7 +249,7 @@ def _print_share_snapshot(cs, snapshot):
if info.get('export_locations'):
info['export_locations'] = (
_transform_export_locations_to_string_view(
cliutils.transform_export_locations_to_string_view(
info['export_locations']))
cliutils.print_dict(info)

1
requirements.txt
View File

@ -15,5 +15,6 @@ requests>=2.14.2 # Apache-2.0
simplejson>=3.5.1 # MIT
Babel!=2.4.0,>=2.3.4 # BSD
six>=1.10.0 # MIT
osc-lib>=1.10.0 # Apache-2.0
python-keystoneclient>=3.8.0 # Apache-2.0
debtcollector>=1.2.0 # Apache-2.0

9
setup.cfg
View File

@ -36,6 +36,15 @@ console_scripts =
oslo.config.opts =
manilaclient.config = manilaclient.config:list_opts
openstack.cli.extension =
share = manilaclient.osc.plugin
openstack.share.v2 =
share_list = manilaclient.osc.v2.share:ListShare
share_create = manilaclient.osc.v2.share:CreateShare
share_delete = manilaclient.osc.v2.share:DeleteShare
share_show = manilaclient.osc.v2.share:ShowShare
[wheel]
universal = 1