Add security add/update services for in-use share networks

This change adds two new commands to the client:
- 'share-network-security-service-update'
- 'share-network-security-service-update-check'
- 'share-network-security-service-add-check'
- 'share-network-reset-state'

Partially Implements: bp add-security-service-in-use-share-networks
Co-Authored-By: Carlos Eduardo <ces.eduardo98@gmail.com>
Co-Authored-By: Andre Beltrami <debeltrami@gmail.com>
Co-Authored-By: Douglas Viroel <viroel@gmail.com>

Depends-On: Id121ba942c7840a7cd7574f08a524fd4dbe06f64
Change-Id: Ib7b67b4c0ce9b16e1ee4a8291cb48c0b1c7ca367
This commit is contained in:
debeltrami 2020-12-07 19:42:45 +00:00 committed by Douglas Viroel
parent 8d71f1ee03
commit 1075210b95
7 changed files with 630 additions and 36 deletions

View File

@ -27,7 +27,7 @@ from manilaclient import utils
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
MAX_VERSION = '2.57' MAX_VERSION = '2.63'
MIN_VERSION = '2.0' MIN_VERSION = '2.0'
DEPRECATED_VERSION = '1.0' DEPRECATED_VERSION = '1.0'
_VERSIONED_METHOD_MAP = {} _VERSIONED_METHOD_MAP = {}

View File

@ -587,6 +587,71 @@ class ManilaCLIClient(base.CLIClient):
share_networks = utils.listing(share_networks_raw) share_networks = utils.listing(share_networks_raw)
return share_networks return share_networks
def share_network_reset_state(self, id=None, state=None,
microversion=None):
cmd = 'share-network-reset-state %s ' % id
if state:
cmd += '--state %s' % state
share_network_raw = self.manila(cmd, microversion=microversion)
share_network = utils.listing(share_network_raw)
return share_network
def share_network_security_service_add(
self, share_network_id, security_service_id, microversion=None):
cmd = ('share-network-security-service-add %(network_id)s '
'%(service_id)s' % {'network_id': share_network_id,
'service_id': security_service_id})
self.manila(cmd, microversion=microversion)
def share_network_security_service_update(
self, share_network_id, current_security_service_id,
new_security_service_id, microversion=None):
cmd = (
'share-network-security-service-update %(network_id)s '
'%(current_service_id)s %(new_security_service_id)s' % {
'network_id': share_network_id,
'current_service_id': current_security_service_id,
'new_security_service_id': new_security_service_id})
self.manila(cmd, microversion=microversion)
def share_network_security_service_add_check(
self, share_network_id, security_service_id,
reset=False, microversion=None):
cmd = (
'share-network-security-service-add-check %(network_id)s '
'%(security_service_id)s' % {
'network_id': share_network_id,
'security_service_id': security_service_id})
if reset:
cmd += '--reset %s' % reset
return output_parser.details(
self.manila(cmd, microversion=microversion))
def share_network_security_service_update_check(
self, share_network_id, current_security_service_id,
new_security_service_id, reset=False, microversion=None):
cmd = (
'share-network-security-service-update-check %(network_id)s '
'%(current_security_service_id)s %(new_security_service_id)s ' % {
'network_id': share_network_id,
'current_security_service_id': current_security_service_id,
'new_security_service_id': new_security_service_id})
if reset:
cmd += '--reset %s' % reset
return output_parser.details(
self.manila(cmd, microversion=microversion))
def share_network_security_service_list(
self, share_network_id, microversion=None):
cmd = ('share-network-security-service-list %s' % share_network_id)
share_networks_raw = self.manila(cmd, microversion=microversion)
network_services = utils.listing(share_networks_raw)
return network_services
def is_share_network_deleted(self, share_network, microversion=None): def is_share_network_deleted(self, share_network, microversion=None):
"""Says whether share network is deleted or not. """Says whether share network is deleted or not.

View File

@ -18,10 +18,16 @@ import ast
import ddt import ddt
from tempest.lib.common.utils import data_utils from tempest.lib.common.utils import data_utils
from tempest.lib import exceptions as tempest_lib_exc from tempest.lib import exceptions as tempest_lib_exc
import time
from manilaclient import config
from manilaclient import exceptions
from manilaclient.tests.functional import base from manilaclient.tests.functional import base
from manilaclient.tests.functional import utils from manilaclient.tests.functional import utils
SECURITY_SERVICE_UPDATE_VERSION = '2.63'
CONF = config.CONF
@ddt.ddt @ddt.ddt
class ShareNetworksReadWriteTest(base.BaseTestCase): class ShareNetworksReadWriteTest(base.BaseTestCase):
@ -293,3 +299,256 @@ class ShareNetworksReadWriteTest(base.BaseTestCase):
filters=filters) filters=filters)
self.assertGreater(len(share_networks), 0) self.assertGreater(len(share_networks), 0)
def test_share_network_reset_status(self):
share_network = self.create_share_network(
client=self.user_client,
name='cool_net_name',
description='fakedescription',
neutron_net_id='fake_neutron_net_id',
neutron_subnet_id='fake_neutron_subnet_id',
)
# Admin operation
self.admin_client.share_network_reset_state(
share_network['id'], 'error',
microversion=SECURITY_SERVICE_UPDATE_VERSION)
self.user_client.wait_for_resource_status(
share_network['id'], 'error',
microversion=SECURITY_SERVICE_UPDATE_VERSION,
resource_type="share_network")
def test_share_network_security_service_add(self):
share_network = self.create_share_network(
client=self.user_client,
name='cool_net_name',
description='fakedescription',
neutron_net_id='fake_neutron_net_id',
neutron_subnet_id='fake_neutron_subnet_id',
)
new_security_service = self.create_security_service(
client=self.user_client)
check_result = (
self.user_client.share_network_security_service_add_check(
share_network['id'],
security_service_id=new_security_service['id']))
self.assertEqual(check_result['compatible'], 'True')
self.user_client.share_network_security_service_add(
share_network['id'], new_security_service['id'])
network_services = (
self.user_client.share_network_security_service_list(
share_network['id']))
self.assertEqual(len(network_services), 1)
self.assertEqual(
network_services[0]['id'], new_security_service['id'])
def test_share_network_security_service_update(self):
share_network = self.create_share_network(
client=self.user_client,
name='cool_net_name',
description='fakedescription',
neutron_net_id='fake_neutron_net_id',
neutron_subnet_id='fake_neutron_subnet_id',
)
current_name = 'current'
new_name = 'new'
current_security_service = self.create_security_service(
client=self.user_client, name=current_name)
new_security_service = self.create_security_service(
client=self.user_client, name=new_name)
check_result = (
self.user_client.share_network_security_service_add_check(
share_network['id'], current_security_service['id']))
self.assertEqual(check_result['compatible'], 'True')
self.user_client.share_network_security_service_add(
share_network['id'], current_security_service['id'])
network_services = (
self.user_client.share_network_security_service_list(
share_network['id']))
self.assertEqual(len(network_services), 1)
self.assertEqual(network_services[0]['name'], current_name)
check_result = (
self.user_client.share_network_security_service_update_check(
share_network['id'], current_security_service['id'],
new_security_service['id']))
self.assertEqual(check_result['compatible'], 'True')
self.user_client.share_network_security_service_update(
share_network['id'], current_security_service['id'],
new_security_service['id'])
network_services = (
self.user_client.share_network_security_service_list(
share_network['id']))
self.assertEqual(len(network_services), 1)
self.assertEqual(network_services[0]['name'], new_name)
class ShareNetworkSecurityServiceCheckReadWriteTests(base.BaseTestCase):
protocol = None
def setUp(self):
super(ShareNetworkSecurityServiceCheckReadWriteTests, self).setUp()
if self.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % self.protocol
raise self.skipException(message)
self.client = self.get_user_client()
if not self.client.share_network:
message = "Can run only with DHSS=True mode"
raise self.skipException(message)
def _wait_for_update_security_service_compatible_result(
self, share_network, current_security_service,
new_security_service=None):
compatible_expected_result = 'True'
check_is_compatible = 'None'
tentatives = 0
# There might be a delay from the time the check is requested until
# the backend has performed all checks
while check_is_compatible != compatible_expected_result:
tentatives += 1
if not new_security_service:
check_is_compatible = (
self.user_client.share_network_security_service_add_check(
share_network['id'],
current_security_service['id']))['compatible']
else:
check_is_compatible = (
(self.user_client.
share_network_security_service_update_check(
share_network['id'],
current_security_service['id'],
new_security_service['id'])))['compatible']
if tentatives > 3:
timeout_message = (
"Share network security service add/update check did not "
"reach 'compatible=True' within 15 seconds.")
raise exceptions.TimeoutException(message=timeout_message)
time.sleep(5)
def test_check_if_security_service_can_be_added_to_share_network_in_use(
self):
share_network = self.create_share_network(
client=self.user_client,
description='fakedescription',
neutron_net_id='fake_neutron_net_id',
neutron_subnet_id='fake_neutron_subnet_id',
)
# Create a share so we can be sure that a share server will exist and
# the check will be performed in the backends
self.create_share(
self.protocol, client=self.user_client,
share_network=share_network['id'])
current_security_service = self.create_security_service(
client=self.user_client)
check_result = (
self.user_client.share_network_security_service_add_check(
share_network['id'],
current_security_service['id']))
self.assertEqual(check_result['compatible'], 'None')
self._wait_for_update_security_service_compatible_result(
share_network, current_security_service)
def test_add_and_update_security_service_when_share_network_is_in_use(
self):
share_network = self.create_share_network(
client=self.user_client,
name='cool_net_name',
description='fakedescription',
neutron_net_id='fake_neutron_net_id',
neutron_subnet_id='fake_neutron_subnet_id',
)
# Create a share so we can be sure that a share server will exist and
# the check will be performed in the backends
self.create_share(
self.protocol, name='fake_share_name',
share_network=share_network['id'], client=self.user_client)
current_security_service = self.create_security_service(
client=self.user_client, name='current_security_service')
new_security_service = self.create_security_service(
client=self.user_client, name='new_security_service')
check_result = (
self.user_client.share_network_security_service_add_check(
share_network['id'], current_security_service['id']))
self.assertEqual(check_result['compatible'], 'None')
self._wait_for_update_security_service_compatible_result(
share_network, current_security_service)
self.user_client.share_network_security_service_add(
share_network['id'], current_security_service['id'])
network_services = (
self.user_client.share_network_security_service_list(
share_network['id']))
self.assertEqual(len(network_services), 1)
self.assertEqual(
network_services[0]['name'], current_security_service['name'])
self.user_client.wait_for_resource_status(
share_network['id'], 'active',
microversion=SECURITY_SERVICE_UPDATE_VERSION,
resource_type="share_network")
check_result = (
self.user_client.share_network_security_service_update_check(
share_network['id'], current_security_service['id'],
new_security_service['id']))
self.assertEqual(check_result['compatible'], 'None')
self._wait_for_update_security_service_compatible_result(
share_network, current_security_service,
new_security_service=new_security_service)
self.user_client.share_network_security_service_update(
share_network['id'], current_security_service['id'],
new_security_service['id'])
network_services = (
self.user_client.share_network_security_service_list(
share_network['id']))
self.assertEqual(len(network_services), 1)
self.assertEqual(
network_services[0]['name'], new_security_service['name'])
self.user_client.wait_for_resource_status(
share_network['id'], 'active',
microversion=SECURITY_SERVICE_UPDATE_VERSION,
resource_type="share_network")
base_security_service_check = ShareNetworkSecurityServiceCheckReadWriteTests
class ShareNetworkSecServiceCheckRWNFSTest(base_security_service_check):
protocol = 'nfs'
class ShareNetworkSecServiceCheckRWTestsCIFSTest(base_security_service_check):
protocol = 'cifs'

View File

@ -145,35 +145,30 @@ class ShareNetworkTest(utils.TestCase):
def test_add_security_service(self): def test_add_security_service(self):
security_service = 'fake security service' security_service = 'fake security service'
share_nw = 'fake share nw' share_nw = 'fake share nw'
expected_path = (share_networks.RESOURCE_PATH + '/action') % share_nw expected_path = 'add_security_service'
expected_body = { expected_body = {
'add_security_service': { 'security_service_id': security_service,
'security_service_id': security_service,
},
} }
with mock.patch.object(self.manager, '_create', mock.Mock()): with mock.patch.object(self.manager, '_action', mock.Mock()):
self.manager.add_security_service(share_nw, security_service) self.manager.add_security_service(share_nw, security_service)
self.manager._create.assert_called_once_with( self.manager._action.assert_called_once_with(
expected_path, expected_path,
expected_body, share_nw,
share_networks.RESOURCE_NAME) expected_body)
def test_add_security_service_to_share_nw_object(self): def test_add_security_service_to_share_nw_object(self):
security_service = self._FakeSecurityService() security_service = self._FakeSecurityService()
share_nw = self._FakeShareNetwork() share_nw = self._FakeShareNetwork()
expected_path = ((share_networks.RESOURCE_PATH + expected_path = 'add_security_service'
'/action') % share_nw.id)
expected_body = { expected_body = {
'add_security_service': { 'security_service_id': security_service.id,
'security_service_id': security_service.id,
},
} }
with mock.patch.object(self.manager, '_create', mock.Mock()): with mock.patch.object(self.manager, '_action', mock.Mock()):
self.manager.add_security_service(share_nw, security_service) self.manager.add_security_service(share_nw, security_service)
self.manager._create.assert_called_once_with( self.manager._action.assert_called_once_with(
expected_path, expected_path,
expected_body, share_nw,
share_networks.RESOURCE_NAME) expected_body)
def test_remove_security_service(self): def test_remove_security_service(self):
security_service = 'fake security service' security_service = 'fake security service'
@ -207,3 +202,80 @@ class ShareNetworkTest(utils.TestCase):
expected_path, expected_path,
expected_body, expected_body,
share_networks.RESOURCE_NAME) share_networks.RESOURCE_NAME)
def test_update_share_network_security_service(self):
current_security_service = self._FakeSecurityService()
new_security_service = self._FakeSecurityService()
share_nw = self._FakeShareNetwork()
expected_path = 'update_security_service'
expected_body = {
'current_service_id': current_security_service.id,
'new_service_id': new_security_service.id
}
with mock.patch.object(self.manager, '_action', mock.Mock()):
self.manager.update_share_network_security_service(
share_nw, current_security_service, new_security_service)
self.manager._action.assert_called_once_with(
expected_path,
share_nw,
expected_body)
def test_share_network_reset_state(self):
share_nw = self._FakeShareNetwork()
state = 'active'
expected_path = 'reset_status'
expected_body = {
'status': state,
}
with mock.patch.object(self.manager, '_action', mock.Mock()):
self.manager.reset_state(
share_nw, state)
self.manager._action.assert_called_once_with(
expected_path,
share_nw,
expected_body)
def test_share_network_security_service_update_check(self):
current_security_service = self._FakeSecurityService()
new_security_service = self._FakeSecurityService()
share_nw = self._FakeShareNetwork()
expected_path = 'update_security_service_check'
expected_body = {
'current_service_id': current_security_service.id,
'new_service_id': new_security_service.id,
'reset_operation': False
}
with mock.patch.object(self.manager, '_action', mock.Mock()):
self.manager.update_share_network_security_service_check(
share_nw, current_security_service, new_security_service)
self.manager._action.assert_called_once_with(
expected_path,
share_nw,
expected_body)
def test_add_security_service_check(self):
current_security_service = self._FakeSecurityService()
share_nw = self._FakeShareNetwork()
expected_path = 'add_security_service_check'
expected_body = {
'security_service_id': current_security_service.id,
'reset_operation': False
}
with mock.patch.object(self.manager, '_action', mock.Mock()):
self.manager.add_security_service_check(
share_nw, current_security_service, False)
self.manager._action.assert_called_once_with(
expected_path,
share_nw,
expected_body)

View File

@ -22,6 +22,7 @@ RESOURCES_PATH = '/share-networks'
RESOURCE_PATH = "/share-networks/%s" RESOURCE_PATH = "/share-networks/%s"
RESOURCE_NAME = 'share_network' RESOURCE_NAME = 'share_network'
RESOURCES_NAME = 'share_networks' RESOURCES_NAME = 'share_networks'
ACTION_PATH = RESOURCE_PATH + '/action'
class ShareNetwork(common_base.Resource): class ShareNetwork(common_base.Resource):
@ -115,24 +116,6 @@ class ShareNetworkManager(base.ManagerWithFind):
return self._create(RESOURCES_PATH, body, RESOURCE_NAME) return self._create(RESOURCES_PATH, body, RESOURCE_NAME)
def add_security_service(self, share_network, security_service):
"""Associate given security service with a share network.
:param share_network: share network name, id or ShareNetwork instance
:param security_service: name, id or SecurityService instance
:rtype: :class:`ShareNetwork`
"""
body = {
'add_security_service': {
'security_service_id': common_base.getid(security_service),
},
}
return self._create(
RESOURCE_PATH % common_base.getid(share_network) + '/action',
body,
RESOURCE_NAME,
)
def remove_security_service(self, share_network, security_service): def remove_security_service(self, share_network, security_service):
"""Dissociate security service from a share network. """Dissociate security service from a share network.
@ -246,3 +229,96 @@ class ShareNetworkManager(base.ManagerWithFind):
path = RESOURCES_PATH + query_string path = RESOURCES_PATH + query_string
return self._list(path, RESOURCES_NAME) return self._list(path, RESOURCES_NAME)
def _action(self, action, share_network, info=None):
"""Perform a share network 'action'.
:param action: text with action name.
:param share_network: either share_network object or text with its ID.
:param info: dict with data for specified 'action'.
"""
body = {action: info}
self.run_hooks('modify_body_for_action', body)
url = ACTION_PATH % common_base.getid(share_network)
return self.api.client.post(url, body=body)
def add_security_service(self, share_network, security_service):
"""Associate given security service with a share network.
:param share_network: share network name, id or ShareNetwork instance
:param security_service: name, id or SecurityService instance
:rtype: :class:`ShareNetwork`
"""
info = {
'security_service_id': common_base.getid(security_service),
}
return self._action('add_security_service', share_network, info)
@api_versions.wraps("2.63")
def add_security_service_check(self, share_network, security_service,
reset_operation=False):
"""Associate given security service with a share network.
:param share_network: share network name, id or ShareNetwork instance
:param security_service: name, id or SecurityService instance
:param reset_operation: start over the check operation
:rtype: :class:`ShareNetwork`
"""
info = {
'security_service_id': common_base.getid(security_service),
'reset_operation': reset_operation,
}
return self._action('add_security_service_check', share_network, info)
@api_versions.wraps("2.63")
def update_share_network_security_service(self, share_network,
current_security_service,
new_security_service):
"""Update current security service to new one of a given share network.
:param share_network: share network name, id or ShareNetwork instance
:param current_security_service: current name, id or
SecurityService instance that will be changed
:param new_security_service: new name, id or
SecurityService instance that will be updated
:rtype: :class:`ShareNetwork`
"""
info = {
'current_service_id': common_base.getid(current_security_service),
'new_service_id': common_base.getid(new_security_service)}
return self._action('update_security_service', share_network, info)
@api_versions.wraps("2.63")
def update_share_network_security_service_check(
self, share_network, current_security_service,
new_security_service, reset_operation=False):
"""Validates if the security service update is supported by all hosts.
:param share_network: share network name, id or ShareNetwork instance
:param current_security_service: current name, id or
SecurityService instance that will be changed
:param new_security_service: new name, id or
:param reset_operation: start over the check operation
SecurityService instance that will be updated
:rtype: :class:`ShareNetwork`
"""
info = {
'current_service_id': common_base.getid(current_security_service),
'new_service_id': common_base.getid(new_security_service),
'reset_operation': reset_operation
}
return self._action('update_security_service_check',
share_network, info)
@api_versions.wraps("2.63")
def reset_state(self, share_network, state):
"""Reset state of a share network.
:param share_network: either share_network object or text with its ID
or name.
:param state: text with new state to set for share network.
"""
return self._action('reset_status', share_network,
{"status": state})

View File

@ -3553,6 +3553,33 @@ def do_share_network_security_service_add(cs, args):
cs.share_networks.add_security_service(share_network, security_service) cs.share_networks.add_security_service(share_network, security_service)
@cliutils.arg(
'share_network',
metavar='<share-network>',
help='Share network name or ID.')
@cliutils.arg(
'security_service',
metavar='<security-service>',
help='Security service name or ID to associate with.')
@cliutils.arg(
'--reset',
metavar='<True|False>',
choices=['True', 'False'],
required=False,
default=False,
help='Reset and restart the check operation.'
'(Optional, Default=False)')
@api_versions.wraps("2.63")
def do_share_network_security_service_add_check(cs, args):
"""Associate security service with share network."""
share_network = _find_share_network(cs, args.share_network)
security_service = _find_security_service(cs, args.security_service)
add_sec_service_result = cs.share_networks.add_security_service_check(
share_network, security_service, reset_operation=args.reset)
# result[0] is response code, result[1] is dict body
cliutils.print_dict(add_sec_service_result[1])
@cliutils.arg( @cliutils.arg(
'share_network', 'share_network',
metavar='<share-network>', metavar='<share-network>',
@ -3594,6 +3621,87 @@ def do_share_network_security_service_list(cs, args):
cliutils.print_list(security_services, fields=fields) cliutils.print_list(security_services, fields=fields)
@cliutils.arg(
'share_network',
metavar='<share-network>',
help='Share network name or ID.')
@cliutils.arg(
'current_security_service',
metavar='<current-security-service>',
help='Current security service name or ID.')
@cliutils.arg(
'new_security_service',
metavar='<new-security-service>',
help='New security service name or ID.')
@api_versions.wraps("2.63")
def do_share_network_security_service_update(cs, args):
"""Update a current security service to a new security service."""
share_network = _find_share_network(cs, args.share_network)
current_security_service = _find_security_service(
cs, args.current_security_service)
new_security_service = _find_security_service(
cs, args.new_security_service)
cs.share_networks.update_share_network_security_service(
share_network, current_security_service, new_security_service)
@cliutils.arg(
'share_network',
metavar='<share-network>',
help='Share network name or ID.')
@cliutils.arg(
'current_security_service',
metavar='<current-security-service>',
help='Current security service name or ID.')
@cliutils.arg(
'new_security_service',
metavar='<new-security-service>',
help='New security service name or ID.')
@cliutils.arg(
'--reset',
metavar='<True|False>',
choices=['True', 'False'],
required=False,
default=False,
help='Reset and start again the check operation.'
'(Optional, Default=False)')
@api_versions.wraps("2.63")
def do_share_network_security_service_update_check(cs, args):
"""Check if a security service update on the share network is supported.
This call can be repeated until a successful result is obtained.
"""
share_network = _find_share_network(cs, args.share_network)
current_security_service = _find_security_service(
cs, args.current_security_service)
new_security_service = _find_security_service(
cs, args.new_security_service)
share_network_update_check = (
cs.share_networks.update_share_network_security_service_check(
share_network, current_security_service, new_security_service,
reset_operation=args.reset))
# result[0] is response code, result[1] is dict body
cliutils.print_dict(share_network_update_check[1])
@cliutils.arg(
'share_network',
metavar='<share-network>',
help='Share network name or ID.')
@cliutils.arg(
'--state',
metavar='<state>',
default=constants.STATUS_ACTIVE,
help=('Indicate which state to assign the share network. Options include '
'active, error, network change. If no state is provided, active '
'will be used.'))
@api_versions.wraps("2.63")
def do_share_network_reset_state(cs, args):
"""Explicitly update the state of a share network (Admin only)."""
share_network = _find_share_network(cs, args.share_network)
cs.share_networks.reset_state(share_network, args.state)
@cliutils.arg( @cliutils.arg(
'share_network', 'share_network',
metavar='<share-network>', metavar='<share-network>',

View File

@ -0,0 +1,14 @@
---
features:
- |
Added support for updating and adding security services to in use share
networks. The command ``share-network-security-service-update`` was
added to the client. Before each of these commands is executed, make
sure to run the correspondent check command, being either
``share-network-security-service-add-check`` or
``share-network-security-service-update-check``. These commands will
check if the desired share network can have security services added or
updated, based on the cloud support. Also, these commands can be used
for both to request a check and to check the outcome.
The command ``share-network-reset-state`` was also implemented in case
there is need to update the share network ``status`` field.