openstack
/
python-manilaclient
Code Issues Proposed changes
Python bindings for Manila
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
869 Commits
8 Branches
10 MiB
 
 
Tree: 346c608535
Branches Tags
${ item.name }
Create branch ${ searchTerm }
from '346c608535'
${ noResults }
python-manilaclient/manilaclient/tests/unit/v2/test_shell.py

3568 lines
140 KiB
Raw Blame History

# -*- coding: utf-8 -*-
# Copyright 2013 OpenStack Foundation
# Copyright 2014 Mirantis, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from unittest import mock
import ddt
import fixtures
from oslo_utils import strutils
import six
from manilaclient import api_versions
from manilaclient import client
from manilaclient.common.apiclient import utils as apiclient_utils
from manilaclient.common import cliutils
from manilaclient.common import constants
from manilaclient import exceptions
from manilaclient import shell
from manilaclient.tests.unit import utils as test_utils
from manilaclient.tests.unit.v2 import fakes
from manilaclient import utils
from manilaclient.v2 import messages
from manilaclient.v2 import security_services
from manilaclient.v2 import share_instances
from manilaclient.v2 import share_network_subnets
from manilaclient.v2 import share_networks
from manilaclient.v2 import share_servers
from manilaclient.v2 import share_snapshots
from manilaclient.v2 import share_types
from manilaclient.v2 import shares
from manilaclient.v2 import shell as shell_v2
@ddt.ddt
class ShellTest(test_utils.TestCase):
FAKE_ENV = {
'MANILA_USERNAME': 'username',
'MANILA_PASSWORD': 'password',
'MANILA_PROJECT_ID': 'project_id',
'MANILA_URL': 'http://no.where',
}
# Patch os.environ to avoid required auth info.
def setUp(self):
"""Run before each test."""
super(ShellTest, self).setUp()
for var in self.FAKE_ENV:
self.useFixture(fixtures.EnvironmentVariable(var,
self.FAKE_ENV[var]))
self.shell = shell.OpenStackManilaShell()
# HACK(bcwaldon): replace this when we start using stubs
self.old_get_client_class = client.get_client_class
client.get_client_class = lambda *_: fakes.FakeClient
# Following shows available separators for optional params
# and its values
self.separators = [' ', '=']
self.create_share_body = {
"share": {
"share_type": None,
"name": None,
"snapshot_id": None,
"description": None,
"metadata": {},
"share_proto": "nfs",
"share_network_id": None,
"size": 1,
"is_public": False,
"availability_zone": None,
}
}
def tearDown(self):
# For some method like test_image_meta_bad_action we are
# testing a SystemExit to be thrown and object self.shell has
# no time to get instantatiated which is OK in this case, so
# we make sure the method is there before launching it.
if hasattr(self.shell, 'cs') and hasattr(self.shell.cs,
'clear_callstack'):
self.shell.cs.clear_callstack()
# HACK(bcwaldon): replace this when we start using stubs
client.get_client_class = self.old_get_client_class
super(ShellTest, self).tearDown()
def run_command(self, cmd, version=None):
if version:
args = ['--os-share-api-version', version] + cmd.split()
else:
args = cmd.split()
self.shell.main(args)
def assert_called(self, method, url, body=None, **kwargs):
return self.shell.cs.assert_called(method, url, body, **kwargs)
def assert_called_anytime(self, method, url, body=None,
clear_callstack=True):
return self.shell.cs.assert_called_anytime(
method, url, body, clear_callstack=clear_callstack)
def test_availability_zone_list(self):
self.run_command('availability-zone-list')
self.assert_called('GET', '/availability-zones')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_availability_zone_list_select_column(self):
self.run_command('availability-zone-list --columns id,name')
self.assert_called('GET', '/availability-zones')
cliutils.print_list.assert_called_once_with(
mock.ANY, fields=['Id', 'Name'])
def test_service_list(self):
self.run_command('service-list')
self.assert_called('GET', '/services')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_service_list_select_column(self):
self.run_command('service-list --columns id,host')
self.assert_called('GET', '/services')
cliutils.print_list.assert_called_once_with(
mock.ANY, fields=['Id', 'Host'])
def test_service_enable(self):
self.run_command('service-enable foo_host@bar_backend manila-share')
self.assert_called(
'PUT',
'/services/enable',
{'host': 'foo_host@bar_backend', 'binary': 'manila-share'})
def test_service_disable(self):
self.run_command('service-disable foo_host@bar_backend manila-share')
self.assert_called(
'PUT',
'/services/disable',
{'host': 'foo_host@bar_backend', 'binary': 'manila-share'})
def test_list(self):
self.run_command('list')
# NOTE(jdg): we default to detail currently
self.assert_called('GET', '/shares/detail')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_list_select_column(self):
self.run_command('list --column id,name')
self.assert_called('GET', '/shares/detail')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['Id', 'Name'], sortby_index=None)
def test_list_sort_by_name(self):
self.run_command('list --sort_key name')
self.assert_called('GET', '/shares/detail?sort_key=name')
def test_list_filter_status(self):
for separator in self.separators:
self.run_command('list --status' + separator + 'available')
self.assert_called('GET', '/shares/detail?status=available')
def test_list_filter_name(self):
for separator in self.separators:
self.run_command('list --name' + separator + '1234')
self.assert_called('GET', '/shares/detail?name=1234')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_list_all_tenants_only_key(self):
self.run_command('list --all-tenants')
self.assert_called('GET', '/shares/detail?all_tenants=1')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['ID', 'Name', 'Size', 'Share Proto', 'Status', 'Is Public',
'Share Type Name', 'Host', 'Availability Zone', 'Project ID'],
sortby_index=None)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_list_select_column_and_all_tenants(self):
self.run_command('list --columns ID,Name --all-tenants')
self.assert_called('GET', '/shares/detail?all_tenants=1')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['Id', 'Name'], sortby_index=None)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_list_select_column_and_public(self):
self.run_command('list --columns ID,Name --public')
self.assert_called('GET', '/shares/detail?is_public=True')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['Id', 'Name'], sortby_index=None)
def test_list_all_tenants_key_and_value_1(self):
for separator in self.separators:
self.run_command('list --all-tenants' + separator + '1')
self.assert_called('GET', '/shares/detail?all_tenants=1')
def test_list_all_tenants_key_and_value_0(self):
for separator in self.separators:
self.run_command('list --all-tenants' + separator + '0')
self.assert_called('GET', '/shares/detail')
def test_list_filter_by_share_server_and_its_aliases(self):
aliases = [
'--share-server-id', '--share-server_id',
'--share_server-id', '--share_server_id',
]
for alias in aliases:
for separator in self.separators:
self.run_command('list ' + alias + separator + '1234')
self.assert_called(
'GET', '/shares/detail?share_server_id=1234')
def test_list_filter_by_metadata(self):
self.run_command('list --metadata key=value')
self.assert_called(
'GET', '/shares/detail?metadata=%7B%27key%27%3A+%27value%27%7D')
def test_list_filter_by_extra_specs_and_its_aliases(self):
aliases = ['--extra-specs', '--extra_specs', ]
for alias in aliases:
self.run_command('list ' + alias + ' key=value')
self.assert_called(
'GET',
'/shares/detail?extra_specs=%7B%27key%27%3A+%27value%27%7D',
)
def test_list_filter_by_share_type_and_its_aliases(self):
fake_st = type('Empty', (object,), {'id': 'fake_st'})
aliases = [
'--share-type', '--share_type', '--share-type-id',
'--share-type_id', '--share_type-id', '--share_type_id',
]
for alias in aliases:
for separator in self.separators:
with mock.patch.object(
apiclient_utils,
'find_resource',
mock.Mock(return_value=fake_st)):
self.run_command('list ' + alias + separator + fake_st.id)
self.assert_called(
'GET', '/shares/detail?share_type_id=' + fake_st.id)
def test_list_filter_by_inexact_name(self):
for separator in self.separators:
self.run_command('list --name~' + separator +
'fake_name')
self.assert_called(
'GET',
'/shares/detail?name~=fake_name')
def test_list_filter_by_inexact_description(self):
for separator in self.separators:
self.run_command('list --description~' + separator +
'fake_description')
self.assert_called(
'GET',
'/shares/detail?description~=fake_description')
def test_list_filter_by_inexact_unicode_name(self):
for separator in self.separators:
self.run_command('list --name~' + separator +
u'Ñ„Ñ„Ñ„')
self.assert_called(
'GET',
'/shares/detail?name~=%D1%84%D1%84%D1%84')
def test_list_filter_by_inexact_unicode_description(self):
for separator in self.separators:
self.run_command('list --description~' + separator +
u'Ñ„Ñ„Ñ„')
self.assert_called(
'GET',
'/shares/detail?description~=%D1%84%D1%84%D1%84')
def test_list_filter_by_share_type_not_found(self):
for separator in self.separators:
self.assertRaises(
exceptions.CommandError,
self.run_command,
'list --share-type' + separator + 'not_found_expected',
)
self.assert_called('GET', '/types?all_tenants=1&is_public=all')
def test_list_with_limit(self):
for separator in self.separators:
self.run_command('list --limit' + separator + '50')
self.assert_called('GET', '/shares/detail?limit=50')
def test_list_with_offset(self):
for separator in self.separators:
self.run_command('list --offset' + separator + '50')
self.assert_called('GET', '/shares/detail?offset=50')
def test_list_with_sort_dir_verify_keys(self):
# Verify allowed aliases and keys
aliases = ['--sort_dir', '--sort-dir']
for alias in aliases:
for key in constants.SORT_DIR_VALUES:
for separator in self.separators:
self.run_command('list ' + alias + separator + key)
self.assert_called('GET', '/shares/detail?sort_dir=' + key)
def test_list_with_fake_sort_dir(self):
self.assertRaises(
ValueError,
self.run_command,
'list --sort-dir fake_sort_dir',
)
def test_list_with_sort_key_verify_keys(self):
# Verify allowed aliases and keys
aliases = ['--sort_key', '--sort-key']
for alias in aliases:
for key in constants.SHARE_SORT_KEY_VALUES:
for separator in self.separators:
self.run_command('list ' + alias + separator + key)
key = 'share_network_id' if key == 'share_network' else key
key = 'snapshot_id' if key == 'snapshot' else key
key = 'share_type_id' if key == 'share_type' else key
self.assert_called('GET', '/shares/detail?sort_key=' + key)
def test_list_with_fake_sort_key(self):
self.assertRaises(
ValueError,
self.run_command,
'list --sort-key fake_sort_key',
)
def test_list_filter_by_snapshot(self):
fake_s = type('Empty', (object,), {'id': 'fake_snapshot_id'})
for separator in self.separators:
with mock.patch.object(
apiclient_utils,
'find_resource',
mock.Mock(return_value=fake_s)):
self.run_command('list --snapshot' + separator + fake_s.id)
self.assert_called(
'GET', '/shares/detail?snapshot_id=' + fake_s.id)
def test_list_filter_by_snapshot_not_found(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'list --snapshot not_found_expected',
)
self.assert_called('GET', '/snapshots/detail?all_tenants=1')
def test_list_filter_by_host(self):
for separator in self.separators:
self.run_command('list --host' + separator + 'fake_host')
self.assert_called('GET', '/shares/detail?host=fake_host')
@ddt.data(('id', 'b4991315-eb7d-43ec-979e-5715d4399827'),
('path', 'fake_path'))
@ddt.unpack
def test_share_list_filter_by_export_location(self, filter_type, value):
for separator in self.separators:
self.run_command('list --export_location' + separator + value)
self.assert_called(
'GET',
'/shares/detail?export_location_' + filter_type + '=' + value)
@ddt.data('list', 'share-instance-list')
def test_share_or_instance_list_filter_by_export_location_version_invalid(
self, cmd):
self.assertRaises(
exceptions.CommandError,
self.run_command,
cmd + ' --export_location=fake',
'2.34'
)
def test_list_filter_by_share_network(self):
aliases = ['--share-network', '--share_network', ]
fake_sn = type('Empty', (object,), {'id': 'fake_share_network_id'})
for alias in aliases:
for separator in self.separators:
with mock.patch.object(
apiclient_utils,
'find_resource',
mock.Mock(return_value=fake_sn)):
self.run_command('list ' + alias + separator + fake_sn.id)
self.assert_called(
'GET', '/shares/detail?share_network_id=' + fake_sn.id)
def test_list_filter_by_share_network_not_found(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'list --share-network not_found_expected',
)
self.assert_called('GET', '/share-networks/detail?all_tenants=1')
@ddt.data('True', 'False')
def test_list_filter_with_count(self, value):
except_url = '/shares/detail?with_count=' + value
if value == 'False':
except_url = '/shares/detail'
for separator in self.separators:
self.run_command('list --count' + separator + value)
self.assert_called('GET', except_url)
@ddt.data('True', 'False')
def test_list_filter_with_count_invalid_version(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'list --count ' + value,
version='2.41'
)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_instance_list(self):
self.run_command('share-instance-list')
self.assert_called('GET', '/share_instances')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['ID', 'Share ID', 'Host', 'Status', 'Availability Zone',
'Share Network ID', 'Share Server ID', 'Share Type ID'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_instance_list_select_column(self):
self.run_command('share-instance-list --column id,host,status')
self.assert_called('GET', '/share_instances')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['Id', 'Host', 'Status'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
@ddt.data(('id', 'b4991315-eb7d-43ec-979e-5715d4399827'),
('path', 'fake_path'))
@ddt.unpack
def test_share_instance_list_filter_by_export_location(self, filter_type,
value):
for separator in self.separators:
self.run_command('share-instance-list --export_location' +
separator + value)
self.assert_called(
'GET',
('/share_instances?export_location_' +
filter_type + '=' + value))
@mock.patch.object(apiclient_utils, 'find_resource',
mock.Mock(return_value='fake'))
def test_share_instance_list_with_share(self):
self.run_command('share-instance-list --share-id=fake')
self.assert_called('GET', '/shares/fake/instances')
def test_share_instance_list_invalid_share(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'share-instance-list --share-id=not-found-id',
)
def test_share_instance_show(self):
self.run_command('share-instance-show 1234')
self.assert_called_anytime('GET', '/share_instances/1234')
def test_share_instance_export_location_list(self):
self.run_command('share-instance-export-location-list 1234')
self.assert_called_anytime(
'GET', '/share_instances/1234/export_locations')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_instance_export_location_list_with_columns(self):
self.run_command(
'share-instance-export-location-list 1234 --columns uuid,path')
self.assert_called_anytime(
'GET', '/share_instances/1234/export_locations')
cliutils.print_list.assert_called_once_with(mock.ANY, ['Uuid', 'Path'])
def test_share_instance_export_location_show(self):
self.run_command(
'share-instance-export-location-show 1234 fake_el_uuid')
self.assert_called_anytime(
'GET', '/share_instances/1234/export_locations/fake_el_uuid')
def test_share_instance_reset_state(self):
self.run_command('share-instance-reset-state 1234')
expected = {'reset_status': {'status': 'available'}}
self.assert_called('POST', '/share_instances/1234/action',
body=expected)
def test_share_instance_force_delete(self):
manager_mock = mock.Mock()
share_instance = share_instances.ShareInstance(
manager_mock, {'id': 'fake'}, True)
with mock.patch.object(shell_v2, '_find_share_instance',
mock.Mock(return_value=share_instance)):
self.run_command('share-instance-force-delete 1234')
manager_mock.force_delete.assert_called_once_with(share_instance)
def test_type_show_details(self):
self.run_command('type-show 1234')
self.assert_called_anytime('GET', '/types/1234')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
@ddt.data(*itertools.product(
('type-list --columns id,is_default', 'type-list --columns id,name',
'type-list --columns is_default', 'type-list'),
{'2.45', '2.46', api_versions.MAX_VERSION}))
@ddt.unpack
def test_type_list(self, command, version):
self.run_command(command, version=version)
columns_requested = ['ID', 'Name', 'visibility',
'is_default', 'required_extra_specs',
'optional_extra_specs', 'Description']
if 'columns' in command:
columns_requested = command.split('--columns ')[1].split(',')
is_default_in_api = (api_versions.APIVersion(version) >=
api_versions.APIVersion('2.46'))
if not is_default_in_api and 'is_default' in columns_requested:
self.assert_called('GET', '/types/default')
self.assert_called_anytime('GET', '/types')
else:
self.assert_called('GET', '/types')
cliutils.print_list.assert_called_with(
mock.ANY, columns_requested, mock.ANY)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_type_list_select_column(self):
self.run_command('type-list --columns id,name')
self.assert_called('GET', '/types')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['id', 'name'],
mock.ANY)
def test_type_list_all(self):
self.run_command('type-list --all')
self.assert_called_anytime('GET', '/types?is_public=all')
@ddt.data(True, False)
def test_type_create_with_access(self, public):
expected = {
'share_type': {
'name': 'test-type-3',
'extra_specs': {
'driver_handles_share_servers': False,
},
'share_type_access:is_public': public
}
}
self.run_command(
'type-create test-type-3 false --is-public %s' %
six.text_type(public))
self.assert_called('POST', '/types', body=expected)
def test_type_access_list(self):
self.run_command('type-access-list 3')
self.assert_called('GET', '/types/3/share_type_access')
def test_type_access_add_project(self):
expected = {'addProjectAccess': {'project': '101'}}
self.run_command('type-access-add 3 101')
self.assert_called('POST', '/types/3/action', body=expected)
def test_type_access_remove_project(self):
expected = {'removeProjectAccess': {'project': '101'}}
self.run_command('type-access-remove 3 101')
self.assert_called('POST', '/types/3/action', body=expected)
def test_list_filter_by_project_id(self):
aliases = ['--project-id', '--project_id']
for alias in aliases:
for separator in self.separators:
self.run_command('list ' + alias + separator + 'fake_id')
self.assert_called('GET', '/shares/detail?project_id=fake_id')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_list_with_public_shares(self):
listed_fields = [
'ID',
'Name',
'Size',
'Share Proto',
'Status',
'Is Public',
'Share Type Name',
'Host',
'Availability Zone',
'Project ID'
]
self.run_command('list --public')
self.assert_called('GET', '/shares/detail?is_public=True')
cliutils.print_list.assert_called_with(mock.ANY, listed_fields,
sortby_index=None)
def test_show(self):
self.run_command('show 1234')
self.assert_called_anytime('GET', '/shares/1234')
def test_share_export_location_list(self):
self.run_command('share-export-location-list 1234')
self.assert_called_anytime(
'GET', '/shares/1234/export_locations')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_export_location_list_with_columns(self):
self.run_command('share-export-location-list 1234 --columns uuid,path')
self.assert_called_anytime(
'GET', '/shares/1234/export_locations')
cliutils.print_list.assert_called_once_with(mock.ANY, ['Uuid', 'Path'])
def test_share_export_location_show(self):
self.run_command('share-export-location-show 1234 fake_el_uuid')
self.assert_called_anytime(
'GET', '/shares/1234/export_locations/fake_el_uuid')
@ddt.data({'cmd_args': '--driver_options opt1=opt1 opt2=opt2'
' --share_type fake_share_type',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
'share_type': 'fake_share_type',
'share_server_id': None,
}},
{'cmd_args': '--share_type fake_share_type',
'valid_params': {
'driver_options': {},
'share_type': 'fake_share_type',
'share_server_id': None,
}},
{'cmd_args': '',
'valid_params': {
'driver_options': {},
'share_type': None,
'share_server_id': None,
}},
{'cmd_args': '--public',
'valid_params': {
'driver_options': {},
'share_type': None,
'share_server_id': None,
},
'is_public': True,
'version': '--os-share-api-version 2.8',
},
{'cmd_args': '',
'valid_params': {
'driver_options': {},
'share_type': None,
'share_server_id': None,
},
'is_public': False,
'version': '--os-share-api-version 2.8',
},
{'cmd_args': '--driver_options opt1=opt1 opt2=opt2'
' --share_type fake_share_type',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
'share_type': 'fake_share_type',
'share_server_id': None,
},
'version': '--os-share-api-version 2.49',
},
{'cmd_args': '--driver_options opt1=opt1 opt2=opt2'
' --share_type fake_share_type'
' --share_server_id fake_server',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
'share_type': 'fake_share_type',
'share_server_id': 'fake_server',
},
'version': '--os-share-api-version 2.49',
},
{'cmd_args': '--driver_options opt1=opt1 opt2=opt2'
' --share_type fake_share_type'
' --share_server_id fake_server',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
'share_type': 'fake_share_type',
'share_server_id': 'fake_server',
}},
)
@ddt.unpack
def test_manage(self, cmd_args, valid_params, is_public=False,
version=None):
if version is not None:
self.run_command(version
+ ' manage fake_service fake_protocol '
+ ' fake_export_path '
+ cmd_args)
else:
self.run_command(' manage fake_service fake_protocol '
+ ' fake_export_path '
+ cmd_args)
expected = {
'share': {
'service_host': 'fake_service',
'protocol': 'fake_protocol',
'export_path': 'fake_export_path',
'name': None,
'description': None,
'is_public': is_public,
'share_server_id': valid_params['share_server_id'],
}
}
expected['share'].update(valid_params)
self.assert_called('POST', '/shares/manage', body=expected)
def test_manage_invalid_param_share_server_id(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'--os-share-api-version 2.48'
+ ' manage fake_service fake_protocol '
+ ' fake_export_path '
+ ' --driver_options opt1=opt1 opt2=opt2'
+ ' --share_type fake_share_type'
+ ' --share_server_id fake_server')
def test_share_server_manage_unsupported_version(self):
self.assertRaises(
exceptions.UnsupportedVersion,
self.run_command,
'--os-share-api-version 2.48 ' +
'share-server-manage fake_host fake_share_net_id fake_id')
def test_share_server_manage_invalid_param_subnet_id(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'--os-share-api-version 2.49 ' +
'share-server-manage fake_host fake_share_net_id fake_id ' +
'--share-network-subnet fake_subnet_id')
@ddt.data({'driver_args': '--driver_options opt1=opt1 opt2=opt2',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
}},
{'driver_args': '--driver_options opt1=opt1 opt2=opt2',
'subnet_id': 'fake_subnet_1',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
}},
{'driver_args': '--driver_options opt1=opt1 opt2=opt2',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
},
'version': '2.51',
},
{'driver_args': '--driver_options opt1=opt1 opt2=opt2',
'subnet_id': 'fake_subnet_1',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
},
'version': '2.51',
},
{'driver_args': "",
'valid_params': {
'driver_options': {}
},
'version': '2.51',
},
{'driver_args': '--driver_options opt1=opt1 opt2=opt2',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
},
'version': '2.49',
},
{'driver_args': '',
'valid_params': {
'driver_options': {},
},
'network_id': 'fake_network_id',
'version': '2.49',
},
{'driver_args': "",
'valid_params': {
'driver_options': {}
},
'version': '2.49',
},
)
@ddt.unpack
def test_share_server_manage(self, driver_args, valid_params,
version=None, network_id=None,
subnet_id=None):
subnet_support = (version is None or
api_versions.APIVersion(version) >=
api_versions.APIVersion('2.51'))
network_id = '3456' if network_id is None else network_id
fake_share_network = type(
'FakeShareNetwork', (object,), {'id': network_id})
self.mock_object(
shell_v2, '_find_share_network',
mock.Mock(return_value=fake_share_network))
command = ('share-server-manage '
'%(host)s '
'%(share_network_id)s '
'%(identifier)s '
'%(driver_args)s ' % {
'host': 'fake_host',
'share_network_id': fake_share_network.id,
'identifier': '88-as-23-f3-45',
'driver_args': driver_args,
})
command += '--share-network-subnet %s' % subnet_id if subnet_id else ''
self.run_command(command, version=version)
expected = {
'share_server': {
'host': 'fake_host',
'share_network_id': fake_share_network.id,
'identifier': '88-as-23-f3-45',
'driver_options': driver_args
}
}
if subnet_support:
expected['share_server']['share_network_subnet_id'] = subnet_id
expected['share_server'].update(valid_params)
self.assert_called('POST', '/share-servers/manage', body=expected)
@ddt.data(constants.STATUS_ERROR, constants.STATUS_ACTIVE,
constants.STATUS_MANAGE_ERROR, constants.STATUS_UNMANAGE_ERROR,
constants.STATUS_DELETING, constants.STATUS_CREATING)
def test_share_server_reset_state(self, status):
self.run_command('share-server-reset-state 1234 --state %s ' % status)
expected = {'reset_status': {'status': status}}
self.assert_called('POST', '/share-servers/1234/action', body=expected)
def test_unmanage(self):
self.run_command('unmanage 1234')
self.assert_called('POST', '/shares/1234/action')
def test_share_server_unmanage(self):
self.run_command('share-server-unmanage 1234')
self.assert_called('POST', '/share-servers/1234/action',
body={'unmanage': {'force': False}})
def test_share_server_unmanage_force(self):
self.run_command('share-server-unmanage 1234 --force')
self.assert_called('POST', '/share-servers/1234/action',
body={'unmanage': {'force': True}})
@ddt.data({'cmd_args': '--driver_options opt1=opt1 opt2=opt2',
'valid_params': {
'driver_options': {'opt1': 'opt1', 'opt2': 'opt2'},
}},
{'cmd_args': '',
'valid_params': {
'driver_options': {},
}},
)
@ddt.unpack
@mock.patch.object(shell_v2, '_find_share', mock.Mock())
def test_snapshot_manage(self, cmd_args, valid_params):
shell_v2._find_share.return_value = 'fake_share'
self.run_command('snapshot-manage fake_share fake_provider_location '
+ cmd_args)
expected = {
'snapshot': {
'share_id': 'fake_share',
'provider_location': 'fake_provider_location',
'name': None,
'description': None,
}
}
expected['snapshot'].update(valid_params)
self.assert_called('POST', '/snapshots/manage', body=expected)
def test_snapshot_unmanage(self):
self.run_command('snapshot-unmanage 1234')
self.assert_called('POST', '/snapshots/1234/action',
body={'unmanage': None})
def test_revert_to_snapshot(self):
fake_share_snapshot = type(
'FakeShareSnapshot', (object,), {'id': '5678', 'share_id': '1234'})
self.mock_object(
shell_v2, '_find_share_snapshot',
mock.Mock(return_value=fake_share_snapshot))
self.run_command('revert-to-snapshot 5678')
self.assert_called('POST', '/shares/1234/action',
body={'revert': {'snapshot_id': '5678'}})
def test_delete(self):
self.run_command('delete 1234')
self.assert_called('DELETE', '/shares/1234')
@ddt.data(
'--group sg1313', '--share-group sg1313', '--share_group sg1313')
@mock.patch.object(shell_v2, '_find_share_group', mock.Mock())
def test_delete_with_share_group(self, sg_cmd):
fake_sg = type('FakeShareGroup', (object,), {'id': sg_cmd.split()[-1]})
shell_v2._find_share_group.return_value = fake_sg
self.run_command('delete 1234 %s' % sg_cmd)
self.assert_called('DELETE', '/shares/1234?share_group_id=sg1313')
self.assertTrue(shell_v2._find_share_group.called)
def test_delete_not_found(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'delete fake-not-found'
)
@ddt.data(('share_xyz', ), ('share_abc', 'share_xyz'))
def test_delete_wait(self, shares_to_delete):
fake_shares = [
shares.Share('fake', {'id': share})
for share in shares_to_delete
]
share_not_found_error = ("Delete for share %s failed: No share with "
"a name or ID of '%s' exists.")
shares_are_not_found_errors = [
exceptions.CommandError(share_not_found_error % (share, share))
for share in shares_to_delete
]
self.mock_object(
shell_v2, '_find_share',
mock.Mock(side_effect=(fake_shares + shares_are_not_found_errors)))
self.run_command('delete %s --wait' % ' '.join(shares_to_delete))
shell_v2._find_share.assert_has_calls([
mock.call(self.shell.cs, share) for share in shares_to_delete
])
for share in fake_shares:
uri = '/shares/%s' % share.id
self.assert_called_anytime('DELETE', uri, clear_callstack=False)
def test_list_snapshots(self):
self.run_command('snapshot-list')
self.assert_called('GET', '/snapshots/detail')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_snapshot_list_select_column(self):
self.run_command('snapshot-list --columns id,name')
self.assert_called('GET', '/snapshots/detail')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['Id', 'Name'], sortby_index=None)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_list_snapshots_all_tenants_only_key(self):
self.run_command('snapshot-list --all-tenants')
self.assert_called('GET', '/snapshots/detail?all_tenants=1')
cliutils.print_list.assert_called_once_with(
mock.ANY,
['ID', 'Share ID', 'Status', 'Name', 'Share Size', 'Project ID'],
sortby_index=None)
def test_list_snapshots_all_tenants_key_and_value_1(self):
for separator in self.separators:
self.run_command('snapshot-list --all-tenants' + separator + '1')
self.assert_called(
'GET', '/snapshots/detail?all_tenants=1')
def test_list_snapshots_all_tenants_key_and_value_0(self):
for separator in self.separators:
self.run_command('snapshot-list --all-tenants' + separator + '0')
self.assert_called('GET', '/snapshots/detail')
def test_list_snapshots_filter_by_name(self):
for separator in self.separators:
self.run_command('snapshot-list --name' + separator + '1234')
self.assert_called(
'GET', '/snapshots/detail?name=1234')
def test_list_snapshots_filter_by_status(self):
for separator in self.separators:
self.run_command('snapshot-list --status' + separator + '1234')
self.assert_called(
'GET', '/snapshots/detail?status=1234')
def test_list_snapshots_filter_by_share_id(self):
aliases = ['--share_id', '--share-id']
for alias in aliases:
for separator in self.separators:
self.run_command('snapshot-list ' + alias + separator + '1234')
self.assert_called(
'GET', '/snapshots/detail?share_id=1234')
def test_list_snapshots_only_used(self):
for separator in self.separators:
self.run_command('snapshot-list --usage' + separator + 'used')
self.assert_called('GET', '/snapshots/detail?usage=used')
def test_list_snapshots_only_unused(self):
for separator in self.separators:
self.run_command('snapshot-list --usage' + separator + 'unused')
self.assert_called('GET', '/snapshots/detail?usage=unused')
def test_list_snapshots_any(self):
for separator in self.separators:
self.run_command('snapshot-list --usage' + separator + 'any')
self.assert_called('GET', '/snapshots/detail?usage=any')
def test_list_snapshots_with_limit(self):
for separator in self.separators:
self.run_command('snapshot-list --limit' + separator + '50')
self.assert_called(
'GET', '/snapshots/detail?limit=50')
def test_list_snapshots_with_offset(self):
for separator in self.separators:
self.run_command('snapshot-list --offset' + separator + '50')
self.assert_called(
'GET', '/snapshots/detail?offset=50')
def test_list_snapshots_filter_by_inexact_name(self):
for separator in self.separators:
self.run_command('snapshot-list --name~' + separator +
'fake_name')
self.assert_called(
'GET',
'/snapshots/detail?name~=fake_name')
def test_list_snapshots_filter_by_inexact_description(self):
for separator in self.separators:
self.run_command('snapshot-list --description~' + separator +
'fake_description')
self.assert_called(
'GET',
'/snapshots/detail?description~=fake_description')
def test_list_snapshots_filter_by_inexact_unicode_name(self):
for separator in self.separators:
self.run_command('snapshot-list --name~' + separator +
u'Ñ„Ñ„Ñ„')
self.assert_called(
'GET',
'/snapshots/detail?name~=%D1%84%D1%84%D1%84')
def test_list_snapshots_filter_by_inexact_unicode_description(self):
for separator in self.separators:
self.run_command('snapshot-list --description~' + separator +
u'Ñ„Ñ„Ñ„')
self.assert_called(
'GET',
'/snapshots/detail?description~=%D1%84%D1%84%D1%84')
def test_list_snapshots_with_sort_dir_verify_keys(self):
aliases = ['--sort_dir', '--sort-dir']
for alias in aliases:
for key in constants.SORT_DIR_VALUES:
for separator in self.separators:
self.run_command(
'snapshot-list ' + alias + separator + key)
self.assert_called(
'GET',
'/snapshots/detail?sort_dir=' + key)
def test_list_snapshots_with_fake_sort_dir(self):
self.assertRaises(
ValueError,
self.run_command,
'snapshot-list --sort-dir fake_sort_dir',
)
def test_list_snapshots_with_sort_key_verify_keys(self):
aliases = ['--sort_key', '--sort-key']
for alias in aliases:
for key in constants.SNAPSHOT_SORT_KEY_VALUES:
for separator in self.separators:
self.run_command(
'snapshot-list ' + alias + separator + key)
self.assert_called(
'GET',
'/snapshots/detail?sort_key=' + key)
def test_list_snapshots_with_fake_sort_key(self):
self.assertRaises(
ValueError,
self.run_command,
'snapshot-list --sort-key fake_sort_key',
)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_extra_specs_list(self):
self.run_command('extra-specs-list')
self.assert_called('GET', '/types?is_public=all')
cliutils.print_list.assert_called_once_with(
mock.ANY, ['ID', 'Name', 'all_extra_specs'], mock.ANY)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_extra_specs_list_select_column(self):
self.run_command('extra-specs-list --columns id,name')
self.assert_called('GET', '/types?is_public=all')
cliutils.print_list.assert_called_once_with(
mock.ANY, ['id', 'name'], mock.ANY)
@ddt.data('fake', 'FFFalse', 'trueee')
def test_type_create_invalid_dhss_value(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test ' + value,
)
@ddt.data('True', 'False')
def test_type_create_duplicate_dhss(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test ' + value +
' --extra-specs driver_handles_share_servers=' + value,
)
@ddt.data(*itertools.product(
['snapshot_support', 'create_share_from_snapshot_support'],
['True', 'False'])
)
@ddt.unpack
def test_type_create_duplicate_switch_and_extra_spec(self, key, value):
cmd = ('type-create test True --%(key)s %(value)s --extra-specs '
'%(key)s=%(value)s' % {'key': key, 'value': value})
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_type_create_duplicate_extra_spec_key(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test True --extra-specs'
' a=foo1 a=foo2',
)
@ddt.unpack
@ddt.data({'expected_bool': True, 'text': 'true'},
{'expected_bool': True, 'text': '1'},
{'expected_bool': False, 'text': 'false'},
{'expected_bool': False, 'text': '0'})
def test_type_create(self, expected_bool, text):
expected = {
"share_type": {
"name": "test",
"share_type_access:is_public": True,
"extra_specs": {
"driver_handles_share_servers": expected_bool,
}
}
}
self.run_command('type-create test ' + text)
self.assert_called('POST', '/types', body=expected)
def test_type_create_with_description(self):
expected = {
"share_type": {
"name": "test",
"description": "test_description",
"share_type_access:is_public": True,
"extra_specs": {
"driver_handles_share_servers": False,
}
}
}
self.run_command('type-create test false '
'--description test_description', version='2.41')
self.assert_called('POST', '/types', body=expected)
@ddt.data('2.26', '2.40')
def test_type_create_invalid_description_version(self, version):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test false --description test_description',
version=version
)
@ddt.unpack
@ddt.data(
*([{'expected_bool': True, 'text': v}
for v in ('true', 'True', '1', 'TRUE', 'tRuE')] +
[{'expected_bool': False, 'text': v}
for v in ('false', 'False', '0', 'FALSE', 'fAlSe')])
)
def test_type_create_with_snapshot_support(self, expected_bool, text):
expected = {
"share_type": {
"name": "test",
"share_type_access:is_public": True,
"extra_specs": {
"snapshot_support": expected_bool,
"driver_handles_share_servers": False,
}
}
}
self.run_command('type-create test false --snapshot-support ' + text)
self.assert_called('POST', '/types', body=expected)
@ddt.unpack
@ddt.data({'expected_bool': True,
'snapshot_text': 'true',
'replication_type': 'readable'},
{'expected_bool': False,
'snapshot_text': 'false',
'replication_type': 'writable'})
def test_create_with_extra_specs(self, expected_bool, snapshot_text,
replication_type):
expected = {
"share_type": {
"name": "test",
"share_type_access:is_public": True,
"extra_specs": {
"driver_handles_share_servers": False,
"snapshot_support": expected_bool,
"replication_type": replication_type,
}
}
}
self.run_command('type-create test false --extra-specs'
' snapshot_support=' + snapshot_text +
' replication_type=' + replication_type)
self.assert_called('POST', '/types', body=expected)
@ddt.unpack
@ddt.data(
*([{'expected_bool': True, 'text': v}
for v in ('true', 'True', '1', 'TRUE', 'tRuE')] +
[{'expected_bool': False, 'text': v}
for v in ('false', 'False', '0', 'FALSE', 'fAlSe')])
)
def test_type_create_with_create_share_from_snapshot_support(
self, expected_bool, text):
expected = {
"share_type": {
"name": "test",
"share_type_access:is_public": True,
"extra_specs": {
"driver_handles_share_servers": False,
"snapshot_support": True,
"create_share_from_snapshot_support": expected_bool,
}
}
}
self.run_command('type-create test false --snapshot-support true '
'--create-share-from-snapshot-support ' + text)
self.assert_called('POST', '/types', body=expected)
@ddt.data('snapshot_support', 'create_share_from_snapshot_support')
def test_type_create_invalid_switch_value(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test false --%s fake' % value,
)
@ddt.data('snapshot_support', 'create_share_from_snapshot_support')
def test_type_create_invalid_extra_spec_value(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test false --extra-specs %s=fake' % value,
)
@ddt.unpack
@ddt.data(
*([{'expected_bool': True, 'text': v}
for v in ('true', 'True', '1', 'TRUE', 'tRuE')] +
[{'expected_bool': False, 'text': v}
for v in ('false', 'False', '0', 'FALSE', 'fAlSe')])
)
def test_type_create_with_revert_to_snapshot_support(
self, expected_bool, text):
expected = {
"share_type": {
"name": "test",
"share_type_access:is_public": True,
"extra_specs": {
"driver_handles_share_servers": False,
"snapshot_support": True,
"revert_to_snapshot_support": expected_bool,
}
}
}
self.run_command('type-create test false --snapshot-support true '
'--revert-to-snapshot-support ' + text)
self.assert_called('POST', '/types', body=expected)
@ddt.data('fake', 'FFFalse', 'trueee')
def test_type_create_invalid_revert_to_snapshot_support_value(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test false --revert-to-snapshot-support ' + value,
)
@ddt.unpack
@ddt.data(
*([{'expected_bool': True, 'text': v}
for v in ('true', 'True', '1', 'TRUE', 'tRuE')] +
[{'expected_bool': False, 'text': v}
for v in ('false', 'False', '0', 'FALSE', 'fAlSe')])
)
def test_type_create_with_mount_snapshot_support(
self, expected_bool, text):
expected = {
"share_type": {
"name": "test",
"share_type_access:is_public": True,
"extra_specs": {
"driver_handles_share_servers": False,
"snapshot_support": True,
"revert_to_snapshot_support": False,
"mount_snapshot_support": expected_bool,
}
}
}
self.run_command('type-create test false --snapshot-support true '
'--revert-to-snapshot-support false '
'--mount-snapshot-support ' + text)
self.assert_called('POST', '/types', body=expected)
@ddt.data('fake', 'FFFalse', 'trueee')
def test_type_create_invalid_mount_snapshot_support_value(self, value):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'type-create test false --mount-snapshot-support ' + value,
)
@ddt.data('--is-public', '--is_public')
def test_update(self, alias):
# basic rename with positional arguments
self.run_command('update 1234 --name new-name')
expected = {'share': {'display_name': 'new-name'}}
self.assert_called('PUT', '/shares/1234', body=expected)
# change description only
self.run_command('update 1234 --description=new-description')
expected = {'share': {'display_description': 'new-description'}}
self.assert_called('PUT', '/shares/1234', body=expected)
# update is_public attr
valid_is_public_values = strutils.TRUE_STRINGS + strutils.FALSE_STRINGS
for is_public in valid_is_public_values:
self.run_command('update 1234 %(alias)s %(value)s' % {
'alias': alias,
'value': is_public})
expected = {
'share': {
'is_public': strutils.bool_from_string(is_public,
strict=True),
},
}
self.assert_called('PUT', '/shares/1234', body=expected)
for invalid_val in ['truebar', 'bartrue']:
self.assertRaises(ValueError, self.run_command,
'update 1234 %(alias)s %(value)s' % {
'alias': alias,
'value': invalid_val})
# update all attributes
self.run_command('update 1234 --name new-name '
'--description=new-description '
'%s True' % alias)
expected = {'share': {
'display_name': 'new-name',
'display_description': 'new-description',
'is_public': True,
}}
self.assert_called('PUT', '/shares/1234', body=expected)
self.assertRaises(exceptions.CommandError,
self.run_command, 'update 1234')
def test_rename_snapshot(self):
# basic rename with positional arguments
self.run_command('snapshot-rename 1234 new-name')
expected = {'snapshot': {'display_name': 'new-name'}}
self.assert_called('PUT', '/snapshots/1234', body=expected)
# change description only
self.run_command('snapshot-rename 1234 '
'--description=new-description')
expected = {'snapshot': {'display_description': 'new-description'}}
self.assert_called('PUT', '/snapshots/1234', body=expected)
# snapshot-rename and change description
self.run_command('snapshot-rename 1234 new-name '
'--description=new-description')
expected = {'snapshot': {
'display_name': 'new-name',
'display_description': 'new-description',
}}
self.assert_called('PUT', '/snapshots/1234', body=expected)
# noop, the only all will be the lookup
self.assertRaises(exceptions.CommandError,
self.run_command, 'snapshot-rename 1234')
def test_set_metadata_set(self):
self.run_command('metadata 1234 set key1=val1 key2=val2')
self.assert_called('POST', '/shares/1234/metadata',
{'metadata': {'key1': 'val1', 'key2': 'val2'}})
def test_set_metadata_delete_dict(self):
self.run_command('metadata 1234 unset key1=val1 key2=val2')
self.assert_called('DELETE', '/shares/1234/metadata/key1')
self.assert_called('DELETE', '/shares/1234/metadata/key2', pos=-2)
def test_set_metadata_delete_keys(self):
self.run_command('metadata 1234 unset key1 key2')
self.assert_called('DELETE', '/shares/1234/metadata/key1')
self.assert_called('DELETE', '/shares/1234/metadata/key2', pos=-2)
def test_share_metadata_update_all(self):
self.run_command('metadata-update-all 1234 key1=val1 key2=val2')
self.assert_called('PUT', '/shares/1234/metadata',
{'metadata': {'key1': 'val1', 'key2': 'val2'}})
def test_extract_metadata(self):
# mimic the result of argparse's parse_args() method
class Arguments(object):
def __init__(self, metadata=None):
if metadata is None:
metadata = []
self.metadata = metadata
inputs = [
([], {}),
(["key=value"], {"key": "value"}),
(["key"], {"key": None}),
(["k1=v1", "k2=v2"], {"k1": "v1", "k2": "v2"}),
(["k1=v1", "k2"], {"k1": "v1", "k2": None}),
(["k1", "k2=v2"], {"k1": None, "k2": "v2"})
]
for input in inputs:
args = Arguments(metadata=input[0])
self.assertEqual(shell_v2._extract_metadata(args), input[1])
@ddt.data('--wait', '')
def test_extend_with_wait_option(self, wait_option):
share_to_extend = shares.Share('fake', {'id': '1234',
'status': 'extending'})
already_extended_share = shares.Share('fake', {'id': '1234',
'status': 'available'})
fake_shares = [share_to_extend] * 3
fake_shares.append(already_extended_share)
self.mock_object(shell_v2, '_find_share',
mock.Mock(side_effect=fake_shares))
expected_extend_body = {'extend': {'new_size': 77}}
self.run_command('extend 1234 77 %s' % wait_option)
self.assert_called_anytime('POST', '/shares/1234/action',
body=expected_extend_body,
clear_callstack=False)
if wait_option:
shell_v2._find_share.assert_has_calls(
[mock.call(self.shell.cs, '1234')] * 4)
else:
shell_v2._find_share.assert_called_with(
self.shell.cs, '1234')
def test_reset_state(self):
self.run_command('reset-state 1234')
expected = {'reset_status': {'status': 'available'}}
self.assert_called('POST', '/shares/1234/action', body=expected)
@ddt.data('--wait', '')
def test_shrink_with_wait_option(self, wait_option):
share_to_shrink = shares.Share('fake', {'id': '1234',
'status': 'shrinking'})
already_shrank_share = shares.Share('fake', {'id': '1234',
'status': 'available'})
fake_shares = [share_to_shrink] * 3
fake_shares.append(already_shrank_share)
self.mock_object(shell_v2, '_find_share',
mock.Mock(side_effect=fake_shares))
expected_shrink_body = {'shrink': {'new_size': 77}}
self.run_command('shrink 1234 77 %s' % wait_option)
self.assert_called_anytime('POST', '/shares/1234/action',
body=expected_shrink_body,
clear_callstack=False)
if wait_option:
shell_v2._find_share.assert_has_calls(
[mock.call(self.shell.cs, '1234')] * 4)
else:
shell_v2._find_share.assert_called_with(
self.shell.cs, '1234')
def test_reset_state_with_flag(self):
self.run_command('reset-state --state error 1234')
expected = {'reset_status': {'status': 'error'}}
self.assert_called('POST', '/shares/1234/action', body=expected)
def test_snapshot_reset_state(self):
self.run_command('snapshot-reset-state 1234')
expected = {'reset_status': {'status': 'available'}}
self.assert_called('POST', '/snapshots/1234/action', body=expected)
def test_snapshot_reset_state_with_flag(self):
self.run_command('snapshot-reset-state --state error 1234')
expected = {'reset_status': {'status': 'error'}}
self.assert_called('POST', '/snapshots/1234/action', body=expected)
@ddt.data(
{},
{'--name': 'fake_name'},
{'--description': 'fake_description'},
{'--neutron_net_id': 'fake_neutron_net_id'},
{'--neutron_subnet_id': 'fake_neutron_subnet_id'},
{'--description': 'fake_description',
'--name': 'fake_name',
'--neutron_net_id': 'fake_neutron_net_id',
'--neutron_subnet_id': 'fake_neutron_subnet_id'})
def test_share_network_create(self, data):
cmd = 'share-network-create'
for k, v in data.items():
cmd += ' ' + k + ' ' + v
self.run_command(cmd)
self.assert_called('POST', '/share-networks')
@ddt.unpack
@ddt.data(
{'data': {'--name': 'fake_name'}},
{'data': {'--description': 'fake_description'}},
{'data': {'--neutron_net_id': 'fake_neutron_net_id'},
'version': '2.49',
},
{'data': {'--neutron_subnet_id': 'fake_neutron_subnet_id'},
'version': '2.49',
},
{'data': {
'--description': 'fake_description',
'--name': 'fake_name',
'--neutron_net_id': 'fake_neutron_net_id',
'--neutron_subnet_id': 'fake_neutron_subnet_id'},
'version': '2.49',
},
{'data': {'--name': '""'}},
{'data': {'--description': '""'}},
{'data': {'--neutron_net_id': '""'},
'version': '2.49',
},
{'data': {'--neutron_subnet_id': '""'},
'version': '2.49',
},
{'data': {
'--description': '""',
'--name': '""',
'--neutron_net_id': '""',
'--neutron_subnet_id': '""'},
'version': '2.49',
})
def test_share_network_update(self, data, version=None):
cmd = 'share-network-update 1111'
expected = dict()
for k, v in data.items():
cmd += ' ' + k + ' ' + v
expected[k[2:]] = v
expected = dict(share_network=expected)
self.run_command(cmd, version=version)
self.assert_called('PUT', '/share-networks/1111', body=expected)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list(self):
self.run_command('share-network-list')
self.assert_called(
'GET',
'/share-networks/detail',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_select_column(self):
self.run_command('share-network-list --columns id')
self.assert_called(
'GET',
'/share-networks/detail',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['Id'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_all_tenants(self):
self.run_command('share-network-list --all-tenants')
self.assert_called(
'GET',
'/share-networks/detail?all_tenants=1',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
@mock.patch.object(shell_v2, '_find_security_service', mock.Mock())
def test_share_network_list_filter_by_security_service(self):
ss = type('FakeSecurityService', (object,), {'id': 'fake-ss-id'})
shell_v2._find_security_service.return_value = ss
for command in ['--security_service', '--security-service']:
self.run_command('share-network-list %(command)s %(ss_id)s' %
{'command': command,
'ss_id': ss.id})
self.assert_called(
'GET',
'/share-networks/detail?security_service_id=%s' % ss.id,
)
shell_v2._find_security_service.assert_called_with(mock.ANY, ss.id)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_project_id_aliases(self):
for command in ['--project-id', '--project_id']:
self.run_command('share-network-list %s 1234' % command)
self.assert_called(
'GET',
'/share-networks/detail?project_id=1234',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_created_before_aliases(self):
for command in ['--created-before', '--created_before']:
self.run_command('share-network-list %s 2001-01-01' % command)
self.assert_called(
'GET',
'/share-networks/detail?created_before=2001-01-01',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_created_since_aliases(self):
for command in ['--created-since', '--created_since']:
self.run_command('share-network-list %s 2001-01-01' % command)
self.assert_called(
'GET',
'/share-networks/detail?created_since=2001-01-01',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_neutron_net_id_aliases(self):
for command in ['--neutron-net-id', '--neutron-net_id',
'--neutron_net-id', '--neutron_net_id']:
self.run_command('share-network-list %s fake-id' % command)
self.assert_called(
'GET',
'/share-networks/detail?neutron_net_id=fake-id',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_neutron_subnet_id_aliases(self):
for command in ['--neutron-subnet-id', '--neutron-subnet_id',
'--neutron_subnet-id', '--neutron_subnet_id']:
self.run_command('share-network-list %s fake-id' % command)
self.assert_called(
'GET',
'/share-networks/detail?neutron_subnet_id=fake-id',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_network_type_aliases(self):
for command in ['--network_type', '--network-type']:
self.run_command('share-network-list %s local' % command)
self.assert_called(
'GET',
'/share-networks/detail?network_type=local',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_segmentation_id_aliases(self):
for command in ['--segmentation-id', '--segmentation_id']:
self.run_command('share-network-list %s 1234' % command)
self.assert_called(
'GET',
'/share-networks/detail?segmentation_id=1234',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_ip_version_aliases(self):
for command in ['--ip-version', '--ip_version']:
self.run_command('share-network-list %s 4' % command)
self.assert_called(
'GET',
'/share-networks/detail?ip_version=4',
)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_list_all_filters(self):
filters = {
'name': 'fake-name',
'project-id': '1234',
'created-since': '2001-01-01',
'created-before': '2002-02-02',
'neutron-net-id': 'fake-net',
'neutron-subnet-id': 'fake-subnet',
'network-type': 'local',
'segmentation-id': '5678',
'cidr': 'fake-cidr',
'ip-version': '4',
'offset': 10,
'limit': 20,
}
command_str = 'share-network-list'
for key, value in filters.items():
command_str += ' --%(key)s=%(value)s' % {'key': key,
'value': value}
self.run_command(command_str)
query = utils.safe_urlencode(sorted([(k.replace('-', '_'), v) for
(k, v) in filters.items()]))
self.assert_called(
'GET',
'/share-networks/detail?%s' % query,
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['id', 'name'])
def test_share_network_list_filter_by_inexact_name(self):
for separator in self.separators:
self.run_command('share-network-list --name~' + separator +
'fake_name')
self.assert_called(
'GET',
'/share-networks/detail?name~=fake_name')
def test_share_network_list_filter_by_inexact_description(self):
for separator in self.separators:
self.run_command('share-network-list --description~' + separator +
'fake_description')
self.assert_called(
'GET',
'/share-networks/detail?description~=fake_description')
def test_share_network_list_filter_by_inexact_unicode_name(self):
for separator in self.separators:
self.run_command('share-network-list --name~' + separator +
u'Ñ„Ñ„Ñ„')
self.assert_called(
'GET',
'/share-networks/detail?name~=%D1%84%D1%84%D1%84')
def test_share_network_list_filter_by_inexact_unicode_description(self):
for separator in self.separators:
self.run_command('share-network-list --description~' + separator +
u'Ñ„Ñ„Ñ„')
self.assert_called(
'GET',
'/share-networks/detail?description~=%D1%84%D1%84%D1%84')
def test_share_network_security_service_add(self):
self.run_command('share-network-security-service-add fake_share_nw '
'fake_security_service')
self.assert_called(
'POST',
'/share-networks/1234/action',
)
def test_share_network_security_service_remove(self):
self.run_command('share-network-security-service-remove fake_share_nw '
'fake_security_service')
self.assert_called(
'POST',
'/share-networks/1234/action',
)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_network_security_service_list_select_column(self):
self.run_command('share-network-security-service-list '
'fake_share_nw --column id,name')
self.assert_called(
'GET',
'/security-services/detail?share_network_id=1234',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['Id', 'Name'])
def test_share_network_security_service_list_by_name(self):
self.run_command('share-network-security-service-list fake_share_nw')
self.assert_called(
'GET',
'/security-services/detail?share_network_id=1234',
)
def test_share_network_security_service_list_by_name_not_found(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'share-network-security-service-list inexistent_share_nw',
)
def test_share_network_security_service_list_by_name_multiple(self):
self.assertRaises(
exceptions.CommandError,
self.run_command,
'share-network-security-service-list duplicated_name',
)
def test_share_network_security_service_list_by_id(self):
self.run_command('share-network-security-service-list 1111')
self.assert_called(
'GET',
'/security-services/detail?share_network_id=1111',
)
@ddt.data(
{},
{'--neutron_net_id': 'fake_neutron_net_id',
'--neutron_subnet_id': 'fake_neutron_subnet_id'},
{'--availability-zone': 'fake_availability_zone_id'},
{'--neutron_net_id': 'fake_neutron_net_id',
'--neutron_subnet_id': 'fake_neutron_subnet_id',
'--availability-zone': 'fake_availability_zone_id'})
def test_share_network_subnet_add(self, data):
fake_share_network = type(
'FakeShareNetwork', (object,), {'id': '1234'})
self.mock_object(
shell_v2, '_find_share_network',
mock.Mock(return_value=fake_share_network))
cmd = 'share-network-subnet-create'
for k, v in data.items():
cmd += ' ' + k + ' ' + v
cmd += ' ' + fake_share_network.id
self.run_command(cmd)
shell_v2._find_share_network.assert_called_once_with(
mock.ANY, fake_share_network.id)
self.assert_called('POST', '/share-networks/1234/subnets')
@ddt.data(
{'--neutron_net_id': 'fake_neutron_net_id'},
{'--neutron_subnet_id': 'fake_neutron_subnet_id'},
{'--neutron_net_id': 'fake_neutron_net_id',
'--availability-zone': 'fake_availability_zone_id'},
{'--neutron_subnet_id': 'fake_neutron_subnet_id',
'--availability-zone': 'fake_availability_zone_id'})
def test_share_network_subnet_add_invalid_param(self, data):
cmd = 'share-network-subnet-create'
for k, v in data.items():
cmd += ' ' + k + ' ' + v
cmd += ' fake_network_id'
self.assertRaises(
exceptions.CommandError,
self.run_command,
cmd)
def test_share_network_subnet_add_invalid_share_network(self):
cmd = 'share-network-subnet-create not-found-id'
self.assertRaises(
exceptions.CommandError,
self.run_command,
cmd)
@ddt.data(('fake_subnet1', ),
('fake_subnet1', 'fake_subnet2'))
def test_share_network_subnet_delete(self, subnet_ids):
fake_share_network = type(
'FakeShareNetwork', (object,), {'id': '1234'})
self.mock_object(
shell_v2, '_find_share_network',
mock.Mock(return_value=fake_share_network))
fake_share_network_subnets = [
share_network_subnets.ShareNetworkSubnet(
'fake', {'id': subnet_id}, True)
for subnet_id in subnet_ids
]
self.run_command(
'share-network-subnet-delete %(network_id)s %(subnet_ids)s' % {
'network_id': fake_share_network.id,
'subnet_ids': ' '.join(subnet_ids)
})
shell_v2._find_share_network.assert_called_once_with(
mock.ANY, fake_share_network.id)
for subnet in fake_share_network_subnets:
self.assert_called_anytime(
'DELETE', '/share-networks/1234/subnets/%s' % subnet.id,
clear_callstack=False)
def test_share_network_subnet_delete_invalid_share_network(self):
command = 'share-network-subnet-delete %(net_id)s %(subnet_id)s' % {
'net_id': 'not-found-id',
'subnet_id': '1234',
}
self.assertRaises(
exceptions.CommandError,
self.run_command,
command)
def test_share_network_subnet_delete_invalid_share_network_subnet(self):
fake_share_network = type(
'FakeShareNetwork', (object,), {'id': '1234'})
self.mock_object(
shell_v2, '_find_share_network',
mock.Mock(return_value=fake_share_network))
command = 'share-network-subnet-delete %(net_id)s %(subnet_id)s' % {
'net_id': fake_share_network.id,
'subnet_id': 'not-found-id',
}
self.assertRaises(
exceptions.CommandError,
self.run_command,
command)
@mock.patch.object(cliutils, 'print_dict', mock.Mock())
def test_share_network_subnet_show(self):
fake_share_network = type(
'FakeShareNetwork', (object,), {'id': '1234'})
self.mock_object(
shell_v2, '_find_share_network',
mock.Mock(return_value=fake_share_network))
args = {
'share_net_id': fake_share_network.id,
'subnet_id': 'fake_subnet_id',
}
self.run_command(
'share-network-subnet-show %(share_net_id)s %(subnet_id)s' % args)
self.assert_called(
'GET',
'/share-networks/%(share_net_id)s/subnets/%(subnet_id)s' % args,
)
cliutils.print_dict.assert_called_once_with(mock.ANY)
def test_share_network_subnet_show_invalid_share_network(self):
command = 'share-network-subnet-show %(net_id)s %(subnet_id)s' % {
'net_id': 'not-found-id',
'subnet_id': 1234,
}
self.assertRaises(
exceptions.CommandError,
self.run_command,
command)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_share_server_list_select_column(self):
self.run_command('share-server-list --columns id,host,status')
self.assert_called('GET', '/share-servers')
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['Id', 'Host', 'Status'])
def test_create_share(self):
# Use only required fields
self.run_command("create nfs 1")
self.assert_called("POST", "/shares", body=self.create_share_body)
def test_create_public_share(self):
expected = self.create_share_body.copy()
expected['share']['is_public'] = True
self.run_command("create --public nfs 1")
self.assert_called("POST", "/shares", body=expected)
def test_create_with_share_network(self):
# Except required fields added share network
sn = "fake-share-network"
with mock.patch.object(shell_v2, "_find_share_network",
mock.Mock(return_value=sn)):
self.run_command("create nfs 1 --share-network %s" % sn)
expected = self.create_share_body.copy()
expected['share']['share_network_id'] = sn
self.assert_called("POST", "/shares", body=expected)
shell_v2._find_share_network.assert_called_once_with(mock.ANY, sn)
def test_create_with_metadata(self):
# Except required fields added metadata
self.run_command("create nfs 1 --metadata key1=value1 key2=value2")
expected = self.create_share_body.copy()
expected['share']['metadata'] = {"key1": "value1", "key2": "value2"}
self.assert_called("POST", "/shares", body=expected)
def test_create_with_wait(self):
self.run_command("create nfs 1 --wait")
expected = self.create_share_body.copy()
self.assert_called_anytime(
"POST", "/shares", body=expected, clear_callstack=False)
self.assert_called("GET", "/shares/1234")
def test_allow_access_cert(self):
self.run_command("access-allow 1234 cert client.example.com")
expected = {
"allow_access": {
"access_type": "cert",
"access_to": "client.example.com",
}
}
self.assert_called("POST", "/shares/1234/action", body=expected)
def test_allow_access_cert_error_gt64(self):
common_name = 'x' * 65
self.assertRaises(exceptions.CommandError, self.run_command,
("access-allow 1234 cert %s" % common_name))
def test_allow_access_cert_error_zero(self):
cmd = mock.Mock()
cmd.split = mock.Mock(side_effect=lambda: ['access-allow', '1234',
'cert', ''])
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
cmd.split.assert_called_once_with()
def test_allow_access_cert_error_whitespace(self):
cmd = mock.Mock()
cmd.split = mock.Mock(side_effect=lambda: ['access-allow', '1234',
'cert', ' '])
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
cmd.split.assert_called_once_with()
def test_allow_access_with_access_level(self):
aliases = ['--access_level', '--access-level']
expected = {
"allow_access": {
"access_type": "ip",
"access_to": "10.0.0.6",
"access_level": "ro",
}
}
for alias in aliases:
for s in self.separators:
self.run_command(
"access-allow " + alias + s + "ro 1111 ip 10.0.0.6")
self.assert_called("POST", "/shares/1111/action",
body=expected)
def test_allow_access_with_valid_access_levels(self):
expected = {
"allow_access": {
"access_type": "ip",
"access_to": "10.0.0.6",
}
}
for level in ['rw', 'ro']:
expected["allow_access"]['access_level'] = level
self.run_command(
"access-allow --access-level " + level + " 1111 ip 10.0.0.6")
self.assert_called("POST", "/shares/1111/action",
body=expected)
def test_allow_access_with_invalid_access_level(self):
self.assertRaises(SystemExit, self.run_command,
"access-allow --access-level fake 1111 ip 10.0.0.6")
def test_allow_access_with_metadata(self):
expected = {
"allow_access": {
"access_type": "ip",
"access_to": "10.0.0.6",
"metadata": {"key1": "v1", "key2": "v2"},
}
}
self.run_command(
"access-allow 2222 ip 10.0.0.6 --metadata key1=v1 key2=v2",
version="2.45")
self.assert_called("POST", "/shares/2222/action", body=expected)
def test_set_access_metadata(self):
expected = {
"metadata": {
"key1": "v1",
"key2": "v2",
}
}
self.run_command(
"access-metadata 9999 set key1=v1 key2=v2",
version="2.45")
self.assert_called("PUT", "/share-access-rules/9999/metadata",
body=expected)
def test_unset_access_metadata(self):
self.run_command(
"access-metadata 9999 unset key1",
version="2.45")
self.assert_called("DELETE", "/share-access-rules/9999/metadata/key1")
@ddt.data("1.0", "2.0", "2.44")
def test_allow_access_with_metadata_not_support_version(self, version):
self.assertRaises(
exceptions.CommandError,
self.run_command,
"access-allow 2222 ip 10.0.0.6 --metadata key1=v1",
version=version,
)
@mock.patch.object(cliutils, 'print_list', mock.Mock())
@ddt.data(*set(["2.44", "2.45", api_versions.MAX_VERSION]))
def test_access_list(self, version):
self.run_command("access-list 1111", version=version)
version = api_versions.APIVersion(version)
cliutils.print_list.assert_called_with(
mock.ANY,
['id', 'access_type', 'access_to', 'access_level', 'state',
'access_key', 'created_at', 'updated_at'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
@ddt.data(*set(["2.44", "2.45", api_versions.MAX_VERSION]))
def test_access_list_select_column(self, version):
self.run_command("access-list 1111 --columns id,access_type",
version=version)
cliutils.print_list.assert_called_with(
mock.ANY,
['Id', 'Access_Type'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_snapshot_access_list(self):
self.run_command("snapshot-access-list 1234")
self.assert_called('GET', '/snapshots/1234/access-list')
cliutils.print_list.assert_called_with(
mock.ANY, ['id', 'access_type', 'access_to', 'state'])
@mock.patch.object(cliutils, 'print_dict', mock.Mock())
def test_snapshot_access_allow(self):
self.run_command("snapshot-access-allow 1234 ip 1.1.1.1")
self.assert_called('POST', '/snapshots/1234/action')
cliutils.print_dict.assert_called_with(
{'access_type': 'ip', 'access_to': '1.1.1.1'})
def test_snapshot_access_deny(self):
self.run_command("snapshot-access-deny 1234 fake_id")
self.assert_called('POST', '/snapshots/1234/action')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_snapshot_export_location_list(self):
self.run_command('snapshot-export-location-list 1234')
self.assert_called(
'GET', '/snapshots/1234/export-locations')
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_snapshot_instance_export_location_list(self):
self.run_command('snapshot-instance-export-location-list 1234')
self.assert_called(
'GET', '/snapshot-instances/1234/export-locations')
@mock.patch.object(cliutils, 'print_dict', mock.Mock())
def test_snapshot_instance_export_location_show(self):
self.run_command('snapshot-instance-export-location-show 1234 '
'fake_el_id')
self.assert_called(
'GET', '/snapshot-instances/1234/export-locations/fake_el_id')
cliutils.print_dict.assert_called_once_with(
{'path': '/fake_path', 'id': 'fake_id'})
@mock.patch.object(cliutils, 'print_dict', mock.Mock())
def test_snapshot_export_location_show(self):
self.run_command('snapshot-export-location-show 1234 fake_el_id')
self.assert_called('GET',
'/snapshots/1234/export-locations/fake_el_id')
cliutils.print_dict.assert_called_once_with(
{'path': '/fake_path', 'id': 'fake_id'})
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_security_service_list(self):
self.run_command('security-service-list')
self.assert_called(
'GET',
'/security-services',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['id', 'name', 'status', 'type'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_security_service_list_select_column(self):
self.run_command('security-service-list --columns name,type')
self.assert_called(
'GET',
'/security-services',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['Name', 'Type'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
@mock.patch.object(shell_v2, '_find_share_network', mock.Mock())
def test_security_service_list_filter_share_network(self):
class FakeShareNetwork(object):
id = 'fake-sn-id'
sn = FakeShareNetwork()
shell_v2._find_share_network.return_value = sn
for command in ['--share-network', '--share_network']:
self.run_command('security-service-list %(command)s %(sn_id)s' %
{'command': command,
'sn_id': sn.id})
self.assert_called(
'GET',
'/security-services?share_network_id=%s' % sn.id,
)
shell_v2._find_share_network.assert_called_with(mock.ANY, sn.id)
cliutils.print_list.assert_called_with(
mock.ANY,
fields=['id', 'name', 'status', 'type'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_security_service_list_detailed(self):
self.run_command('security-service-list --detailed')
self.assert_called(
'GET',
'/security-services/detail',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['id', 'name', 'status', 'type', 'share_networks'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_security_service_list_all_tenants(self):
self.run_command('security-service-list --all-tenants')
self.assert_called(
'GET',
'/security-services?all_tenants=1',
)
cliutils.print_list.assert_called_once_with(
mock.ANY,
fields=['id', 'name', 'status', 'type'])
@mock.patch.object(cliutils, 'print_list', mock.Mock())
def test_security_service_list_all_filters(self):
filters = {
'status': 'new',
'name': 'fake-name',
'type': 'ldap',
'user': 'fake-user',
'dns-ip': '1.1.1.1',
'ou': 'fake-ou',
'server': 'fake-server',
'domain': 'fake-domain',
'offset': 10,
'limit': 20,
}
command_str = 'security-service-list'
for key, value in filters.items():
command_str += ' --%(key)s=%(value)s' % {'key': key,
'value': value}
self.run_command(command_str)
self.assert_called(
'GET',
'/security-services?dns_ip=1.1.1.1&domain=fake-domain&limit=20'
'&name=fake-name&offset=10&ou=fake-ou&server=fake-server'