Checking security service is not used while deleting
Added db method share_network_get_all_by_security_service that returns all share networks to which security service is assigned. Added check to delete method in security service api that gets list of share networks that security service is assigned at and raises error if this list is not empty. Closes-bug: 1304473 Change-Id: Ibb73f77095b035f33376aa1791448de3a1ed939b
This commit is contained in:
parent
c483abf775
commit
b2fd5e6223
@ -109,3 +109,34 @@ class SecServicesMappingNegativeTest(base.BaseSharesTest):
|
||||
self.assertRaises(exceptions.BadRequest,
|
||||
self.cl.add_sec_service_to_share_network,
|
||||
sn["id"], ss["id"])
|
||||
|
||||
@test.attr(type=["gate", "smoke", "negative"])
|
||||
def test_try_delete_ss_that_assigned_to_sn(self):
|
||||
# create share network
|
||||
data = self.generate_share_network_data()
|
||||
|
||||
resp, sn = self.create_share_network(client=self.cl, **data)
|
||||
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
|
||||
self.assertDictContainsSubset(data, sn)
|
||||
|
||||
# create security service
|
||||
data = self.generate_security_service_data()
|
||||
|
||||
resp, ss = self.create_security_service(client=self.cl, **data)
|
||||
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
|
||||
self.assertDictContainsSubset(data, ss)
|
||||
|
||||
# Add security service to share network
|
||||
resp, __ = self.cl.add_sec_service_to_share_network(sn["id"],
|
||||
ss["id"])
|
||||
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
|
||||
|
||||
# Try delete ss, that has been assigned to some sn
|
||||
self.assertRaises(exceptions.Unauthorized,
|
||||
self.cl.delete_security_service,
|
||||
ss["id"], )
|
||||
|
||||
# remove seurity service from share-network
|
||||
resp, __ = self.cl.remove_sec_service_from_share_network(sn["id"],
|
||||
ss["id"])
|
||||
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
|
||||
|
@ -86,13 +86,18 @@ class SecurityServiceController(wsgi.Controller):
|
||||
|
||||
try:
|
||||
security_service = db.security_service_get(context, id)
|
||||
except exception.NotFound:
|
||||
raise exc.HTTPNotFound()
|
||||
|
||||
share_nets = db.share_network_get_all_by_security_service(
|
||||
context, id)
|
||||
if share_nets:
|
||||
# Cannot delete security service
|
||||
# if it is assigned to share networks
|
||||
raise exc.HTTPForbidden()
|
||||
policy.check_policy(context, RESOURCE_NAME,
|
||||
'delete', security_service)
|
||||
db.security_service_delete(context, id)
|
||||
except exception.NotFound:
|
||||
raise exc.HTTPNotFound()
|
||||
except exception.InvalidShare:
|
||||
raise exc.HTTPForbidden()
|
||||
|
||||
return webob.Response(status_int=202)
|
||||
|
||||
|
@ -514,6 +514,12 @@ def share_network_get_all_by_project(context, project_id):
|
||||
return IMPL.share_network_get_all_by_project(context, project_id)
|
||||
|
||||
|
||||
def share_network_get_all_by_security_service(context, share_network_id):
|
||||
"""Get all share network DB records for the given project."""
|
||||
return IMPL.share_network_get_all_by_security_service(
|
||||
context, share_network_id)
|
||||
|
||||
|
||||
def share_network_add_security_service(context, id, security_service_id):
|
||||
return IMPL.share_network_add_security_service(context,
|
||||
id,
|
||||
|
@ -1568,6 +1568,16 @@ def share_network_get_all_by_project(context, project_id, user_id=None,
|
||||
return query.all()
|
||||
|
||||
|
||||
@require_context
|
||||
def share_network_get_all_by_security_service(context, share_network_id):
|
||||
session = get_session()
|
||||
return model_query(context, models.ShareNetwork, session=session).\
|
||||
join(models.ShareNetworkSecurityServiceAssociation,
|
||||
models.ShareNetwork.id ==
|
||||
models.ShareNetworkSecurityServiceAssociation.share_network_id).\
|
||||
filter_by(security_service_id=share_network_id, deleted=False).all()
|
||||
|
||||
|
||||
@require_context
|
||||
def share_network_add_security_service(context, id, security_service_id):
|
||||
session = get_session()
|
||||
|
@ -90,7 +90,9 @@ class ShareApiTest(test.TestCase):
|
||||
def test_security_service_delete(self):
|
||||
db.security_service_delete = mock.Mock()
|
||||
db.security_service_get = mock.Mock()
|
||||
req = fakes.HTTPRequest.blank('/shares/1')
|
||||
db.share_network_get_all_by_security_service = mock.Mock(
|
||||
return_value=[])
|
||||
req = fakes.HTTPRequest.blank('/security_services/1')
|
||||
resp = self.controller.delete(req, 1)
|
||||
db.security_service_delete.assert_called_once_with(
|
||||
req.environ['manila.context'], 1)
|
||||
@ -104,6 +106,14 @@ class ShareApiTest(test.TestCase):
|
||||
req,
|
||||
1)
|
||||
|
||||
def test_security_service_delete_has_share_networks(self):
|
||||
db.security_service_get = mock.Mock()
|
||||
db.share_network_get_all_by_security_service = mock.Mock(
|
||||
return_value=[{'share_network': 'fake_share_network'}])
|
||||
req = fakes.HTTPRequest.blank('/security_services/1')
|
||||
self.assertRaises(webob.exc.HTTPForbidden, self.controller.delete,
|
||||
req, 1)
|
||||
|
||||
def test_security_service_update_name(self):
|
||||
new = self.security_service.copy()
|
||||
updated = self.security_service.copy()
|
||||
|
Loading…
Reference in New Issue
Block a user