Merge "Add RBAC tests for VNF LCM APIs"
This commit is contained in:
Zuul
Gerrit Code Review
commit
d19541340a
|
|
@ -608,12 +608,13 @@ class VnfLcmController(wsgi.Controller):
|
|||
@wsgi.expected_errors((http_client.FORBIDDEN, http_client.NOT_FOUND))
|
||||
def show(self, request, id):
|
||||
context = request.environ['tacker.context']
|
||||
if not CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'show')
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'show',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'show',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
return self._view_builder.show(vnf_instance)
|
||||
|
||||
|
|
@ -708,6 +709,9 @@ class VnfLcmController(wsgi.Controller):
|
|||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'delete',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'delete',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
vnf = self._get_vnf(context, id)
|
||||
self._delete(context, vnf_instance, vnf)
|
||||
|
|
@ -779,14 +783,15 @@ class VnfLcmController(wsgi.Controller):
|
|||
@validation.schema(vnf_lcm.instantiate)
|
||||
def instantiate(self, request, id, body):
|
||||
context = request.environ['tacker.context']
|
||||
if not CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'instantiate')
|
||||
|
||||
vnf = self._get_vnf(context, id)
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'instantiate',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'instantiate',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
return self._instantiate(context, vnf_instance, vnf, body)
|
||||
|
||||
|
|
@ -834,14 +839,15 @@ class VnfLcmController(wsgi.Controller):
|
|||
@validation.schema(vnf_lcm.terminate)
|
||||
def terminate(self, request, id, body):
|
||||
context = request.environ['tacker.context']
|
||||
if not CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'terminate')
|
||||
|
||||
vnf = self._get_vnf(context, id)
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'terminate',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'terminate',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
return self._terminate(context, vnf_instance, vnf, body)
|
||||
|
||||
@check_vnf_status_and_error_point(action="heal",
|
||||
|
|
@ -894,14 +900,15 @@ class VnfLcmController(wsgi.Controller):
|
|||
@validation.schema(vnf_lcm.heal)
|
||||
def heal(self, request, id, body):
|
||||
context = request.environ['tacker.context']
|
||||
if not CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'heal')
|
||||
|
||||
vnf = self._get_vnf(context, id)
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'heal',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'heal',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
if vnf_instance.instantiation_state not in \
|
||||
[fields.VnfInstanceState.INSTANTIATED]:
|
||||
|
|
@ -941,12 +948,13 @@ class VnfLcmController(wsgi.Controller):
|
|||
@wsgi.expected_errors((http_client.FORBIDDEN, http_client.NOT_FOUND))
|
||||
def update(self, request, id, body):
|
||||
context = request.environ['tacker.context']
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'update_vnf',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'update_vnf')
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'update_vnf',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
# get body
|
||||
body_data = {}
|
||||
|
|
@ -1425,8 +1433,6 @@ class VnfLcmController(wsgi.Controller):
|
|||
http_client.NOT_FOUND, http_client.CONFLICT))
|
||||
def scale(self, request, id, body):
|
||||
context = request.environ['tacker.context']
|
||||
if not CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'scale')
|
||||
|
||||
try:
|
||||
vnf_info = self._vnfm_plugin.get_vnf(context, id)
|
||||
|
|
@ -1437,6 +1443,9 @@ class VnfLcmController(wsgi.Controller):
|
|||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'scale',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'scale',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
if not vnf_instance.instantiated_vnf_info.scale_status:
|
||||
return self._make_problem_detail(
|
||||
'NOT SCALE VNF', 409, title='NOT SCALE VNF')
|
||||
|
|
@ -1485,10 +1494,14 @@ class VnfLcmController(wsgi.Controller):
|
|||
http_client.NOT_FOUND, http_client.CONFLICT))
|
||||
def rollback(self, request, id):
|
||||
context = request.environ['tacker.context']
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'rollback')
|
||||
|
||||
try:
|
||||
vnf_lcm_op_occs = objects.VnfLcmOpOcc.get_by_id(context, id)
|
||||
vnf_instance = self._get_vnf_instance(
|
||||
context, vnf_lcm_op_occs.vnf_instance_id)
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'rollback',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
if vnf_lcm_op_occs.operation_state != 'FAILED_TEMP':
|
||||
return self._make_problem_detail(
|
||||
'OperationState IS NOT FAILED_TEMP',
|
||||
|
|
@ -1513,8 +1526,6 @@ class VnfLcmController(wsgi.Controller):
|
|||
|
||||
vnf_info = self._get_rollback_vnf(
|
||||
context, vnf_lcm_op_occs.vnf_instance_id)
|
||||
vnf_instance = self._get_vnf_instance(
|
||||
context, vnf_lcm_op_occs.vnf_instance_id)
|
||||
|
||||
inst_vnf_info = vnf_instance.instantiated_vnf_info
|
||||
if inst_vnf_info is not None:
|
||||
|
|
@ -1537,6 +1548,8 @@ class VnfLcmController(wsgi.Controller):
|
|||
except webob.exc.HTTPNotFound as inst_e:
|
||||
return self._make_problem_detail(
|
||||
str(inst_e), 404, title='VNF NOT FOUND')
|
||||
except exceptions.PolicyNotAuthorized:
|
||||
raise
|
||||
except Exception as e:
|
||||
LOG.error(traceback.format_exc())
|
||||
return self._make_problem_detail(
|
||||
|
|
@ -1573,7 +1586,6 @@ class VnfLcmController(wsgi.Controller):
|
|||
"""
|
||||
|
||||
context = request.environ['tacker.context']
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'cancel')
|
||||
req_body = utils.convert_camelcase_to_snakecase(body)
|
||||
_ = objects.CancelMode(context=context, **req_body)
|
||||
|
||||
|
|
@ -1581,9 +1593,13 @@ class VnfLcmController(wsgi.Controller):
|
|||
vnf_lcm_op_occs = objects.VnfLcmOpOcc.get_by_id(context, id)
|
||||
vnf_instance = self._get_vnf_instance(
|
||||
context, vnf_lcm_op_occs.vnf_instance_id)
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'cancel',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
except (webob.exc.HTTPNotFound, exceptions.NotFound) as e:
|
||||
return self._make_problem_detail(
|
||||
str(e), 404, title='VNF NOT FOUND')
|
||||
except exceptions.PolicyNotAuthorized:
|
||||
raise
|
||||
except Exception as e:
|
||||
LOG.error(traceback.format_exc())
|
||||
return self._make_problem_detail(
|
||||
|
|
@ -1652,26 +1668,30 @@ class VnfLcmController(wsgi.Controller):
|
|||
http_client.NOT_FOUND))
|
||||
def fail(self, request, id):
|
||||
context = request.environ['tacker.context']
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'fail')
|
||||
|
||||
try:
|
||||
vnf_lcm_op_occs = objects.VnfLcmOpOcc.get_by_id(context, id)
|
||||
operation = vnf_lcm_op_occs.operation
|
||||
|
||||
vnf_instance_id = vnf_lcm_op_occs.vnf_instance_id
|
||||
vnf_instance = self._get_vnf_instance(context, vnf_instance_id)
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'fail',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
|
||||
if (vnf_lcm_op_occs.operation_state
|
||||
!= fields.LcmOccsOperationState.FAILED_TEMP):
|
||||
return self._make_problem_detail(
|
||||
'Transitions to FAIL from state %s is not allowed' %
|
||||
vnf_lcm_op_occs.operation_state, 409, title='Conflict')
|
||||
|
||||
vnf_instance_id = vnf_lcm_op_occs.vnf_instance_id
|
||||
vnf_instance = self._get_vnf_instance(context, vnf_instance_id)
|
||||
except webob.exc.HTTPNotFound as e:
|
||||
return self._make_problem_detail(
|
||||
str(e), 404, title='VNF NOT FOUND')
|
||||
except exceptions.NotFound as e:
|
||||
return self._make_problem_detail(
|
||||
str(e), 404, title='VNF LCM NOT FOUND')
|
||||
except exceptions.PolicyNotAuthorized:
|
||||
raise
|
||||
except Exception as e:
|
||||
LOG.error(traceback.format_exc())
|
||||
return self._make_problem_detail(
|
||||
|
|
@ -1741,7 +1761,6 @@ class VnfLcmController(wsgi.Controller):
|
|||
http_client.NOT_FOUND, http_client.CONFLICT))
|
||||
def retry(self, request, id):
|
||||
context = request.environ['tacker.context']
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'retry')
|
||||
|
||||
try:
|
||||
vnf_lcm_op_occs = objects.VnfLcmOpOcc.get_by_id(context, id)
|
||||
|
|
@ -1753,14 +1772,6 @@ class VnfLcmController(wsgi.Controller):
|
|||
return self._make_problem_detail(str(exc),
|
||||
500, title='Internal Server Error')
|
||||
|
||||
# operation state checking
|
||||
if vnf_lcm_op_occs.operation_state != \
|
||||
fields.LcmOccsOperationState.FAILED_TEMP:
|
||||
error_msg = ('Cannot proceed with operation_state %s'
|
||||
% vnf_lcm_op_occs.operation_state)
|
||||
return self._make_problem_detail(error_msg,
|
||||
409, title='Conflict')
|
||||
|
||||
# get vnf
|
||||
try:
|
||||
vnf = self._get_vnf(context, vnf_lcm_op_occs.vnf_instance_id)
|
||||
|
|
@ -1776,16 +1787,28 @@ class VnfLcmController(wsgi.Controller):
|
|||
try:
|
||||
vnf_instance = objects.VnfInstance.get_by_id(
|
||||
context, vnf_lcm_op_occs.vnf_instance_id)
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'retry',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
except exceptions.VnfInstanceNotFound:
|
||||
msg = (_("Can not find requested vnf instance: %s")
|
||||
% vnf_lcm_op_occs.vnf_instance_id)
|
||||
return self._make_problem_detail(msg,
|
||||
404, title='Not Found')
|
||||
except exceptions.PolicyNotAuthorized:
|
||||
raise
|
||||
except Exception as exc:
|
||||
LOG.exception(exc)
|
||||
return self._make_problem_detail(str(exc),
|
||||
500, title='Internal Server Error')
|
||||
|
||||
# operation state checking
|
||||
if vnf_lcm_op_occs.operation_state != \
|
||||
fields.LcmOccsOperationState.FAILED_TEMP:
|
||||
error_msg = ('Cannot proceed with operation_state %s'
|
||||
% vnf_lcm_op_occs.operation_state)
|
||||
return self._make_problem_detail(error_msg,
|
||||
409, title='Conflict')
|
||||
|
||||
operation = vnf_lcm_op_occs.operation
|
||||
body = jsonutils.loads(vnf_lcm_op_occs.operation_params)
|
||||
vnf['before_error_point'] = vnf_lcm_op_occs.error_point
|
||||
|
|
@ -1907,14 +1930,15 @@ class VnfLcmController(wsgi.Controller):
|
|||
@validation.schema(vnf_lcm.change_ext_conn)
|
||||
def change_ext_conn(self, request, id, body):
|
||||
context = request.environ['tacker.context']
|
||||
if not CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'change_ext_conn')
|
||||
|
||||
vnf = self._get_vnf(context, id)
|
||||
vnf_instance = self._get_vnf_instance(context, id)
|
||||
if CONF.oslo_policy.enhanced_tacker_policy:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'change_ext_conn',
|
||||
target=self._get_policy_target(vnf_instance))
|
||||
else:
|
||||
context.can(vnf_lcm_policies.VNFLCM % 'change_ext_conn',
|
||||
target={'project_id': vnf_instance.tenant_id})
|
||||
if (vnf_instance.instantiation_state !=
|
||||
fields.VnfInstanceState.INSTANTIATED):
|
||||
return self._make_problem_detail(
|
||||
|
|
|
|||
|
|
@ -16,16 +16,24 @@
|
|||
from unittest import mock
|
||||
|
||||
from tacker.api.vnflcm.v1 import controller
|
||||
import tacker.conductor.conductorrpc.vnf_lcm_rpc as vnf_lcm_rpc
|
||||
from tacker import objects
|
||||
from tacker.objects import fields
|
||||
from tacker.policies import vnf_lcm as policies
|
||||
from tacker.tests.unit.db import utils
|
||||
from tacker.tests.unit import fake_request
|
||||
import tacker.tests.unit.nfvo.test_nfvo_plugin as test_nfvo_plugin
|
||||
from tacker.tests.unit.policies import base as base_test
|
||||
from tacker.tests.unit.vnflcm import fakes
|
||||
from tacker.tests import uuidsentinel
|
||||
from tacker.vnfm import vim_client
|
||||
|
||||
|
||||
class FakePlugin(mock.Mock):
|
||||
|
||||
def get_vnf(self, *args, **kwargs):
|
||||
return utils.get_dummy_vnf(status='ACTIVE')
|
||||
|
||||
|
||||
class VNFLCMPolicyTest(base_test.BasePolicyTest):
|
||||
"""Test VNF LCM APIs policies with all possible context.
|
||||
|
||||
|
|
@ -39,7 +47,7 @@ class VNFLCMPolicyTest(base_test.BasePolicyTest):
|
|||
super(VNFLCMPolicyTest, self).setUp()
|
||||
self.patcher = mock.patch(
|
||||
'tacker.manager.TackerManager.get_service_plugins',
|
||||
return_value={'VNFM': test_nfvo_plugin.FakeVNFMPlugin()})
|
||||
return_value={'VNFM': FakePlugin()})
|
||||
self.mock_manager = self.patcher.start()
|
||||
self.controller = controller.VnfLcmController()
|
||||
self.vim_info = {
|
||||
|
|
@ -50,15 +58,29 @@ class VNFLCMPolicyTest(base_test.BasePolicyTest):
|
|||
'tenant': 'test',
|
||||
'extra': {}
|
||||
}
|
||||
# Below user's context will be allowed to create the VNF
|
||||
# in their project.
|
||||
self.create_authorized_contexts = [
|
||||
# Below user's context will be allowed to create VNF or a few of
|
||||
# the VNF operations in their project.
|
||||
self.project_authorized_contexts = [
|
||||
self.legacy_admin_context, self.project_admin_context,
|
||||
self.project_member_context, self.project_reader_context,
|
||||
self.project_foo_context, self.other_project_member_context,
|
||||
self.other_project_reader_context
|
||||
]
|
||||
self.create_unauthorized_contexts = []
|
||||
self.project_unauthorized_contexts = []
|
||||
|
||||
# Admin or any user in same project will be allowed to get,
|
||||
# instantiate, terminate etc operations of VNF of their project.
|
||||
self.project_member_authorized_contexts = [
|
||||
self.legacy_admin_context, self.project_admin_context,
|
||||
self.project_member_context, self.project_reader_context,
|
||||
self.project_foo_context
|
||||
]
|
||||
# User from other project will not be allowed to get or perform
|
||||
# the other project's VNF operations.
|
||||
self.project_member_unauthorized_contexts = [
|
||||
self.other_project_member_context,
|
||||
self.other_project_reader_context
|
||||
]
|
||||
|
||||
@mock.patch.object(vim_client.VimClient, "get_vim")
|
||||
@mock.patch.object(objects.VnfPackage, 'get_by_id')
|
||||
|
|
@ -98,8 +120,311 @@ class VNFLCMPolicyTest(base_test.BasePolicyTest):
|
|||
'metadata': {'key': 'value'}}
|
||||
req = fake_request.HTTPRequest.blank('/vnf_instances')
|
||||
rule_name = policies.VNFLCM % 'create'
|
||||
self.common_policy_check(self.create_authorized_contexts,
|
||||
self.create_unauthorized_contexts,
|
||||
self.common_policy_check(self.project_authorized_contexts,
|
||||
self.project_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.create,
|
||||
req, body=body)
|
||||
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_show_vnf(self, mock_vnf_by_id):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s' % uuidsentinel.instance_id)
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
rule_name = policies.VNFLCM % 'show'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.show,
|
||||
req, uuidsentinel.instance_id)
|
||||
|
||||
@mock.patch.object(objects.VnfInstanceList, "get_by_marker_filter")
|
||||
def test_index_vnf(self, mock_vnf_list):
|
||||
req = fake_request.HTTPRequest.blank('/vnf_instances')
|
||||
vnf_instance_1 = fakes.return_vnf_instance()
|
||||
vnf_instance_2 = fakes.return_vnf_instance()
|
||||
mock_vnf_list.return_value = [vnf_instance_1, vnf_instance_2]
|
||||
rule_name = policies.VNFLCM % 'index'
|
||||
self.common_policy_check(self.project_authorized_contexts,
|
||||
self.project_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.index,
|
||||
req)
|
||||
|
||||
@mock.patch.object(objects.VNF, "vnf_index_list")
|
||||
@mock.patch.object(objects.VnfInstanceList, "vnf_instance_list")
|
||||
@mock.patch.object(objects.VnfPackageVnfd, 'get_vnf_package_vnfd')
|
||||
@mock.patch.object(vnf_lcm_rpc.VNFLcmRPCAPI, "update")
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_update_vnf(
|
||||
self, mock_vnf_by_id, mock_update,
|
||||
mock_vnf_package_vnf_get_vnf_package_vnfd,
|
||||
mock_vnf_instance_list,
|
||||
mock_vnf_index_list,):
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf_index_list.return_value = fakes._get_vnf()
|
||||
mock_vnf_instance_list.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf_package_vnf_get_vnf_package_vnfd.return_value =\
|
||||
fakes.return_vnf_package_vnfd()
|
||||
|
||||
body = {"vnfInstanceName": "new_instance_name",
|
||||
"vnfInstanceDescription": "new_instance_discription",
|
||||
"vnfdId": "2c69a161-0000-4b0f-bcf8-391f8fc76600",
|
||||
"vnfConfigurableProperties": {
|
||||
"test": "test_value"
|
||||
},
|
||||
"vnfcInfoModificationsDeleteIds": ["test1"],
|
||||
"metadata": {"testkey": "test_value"},
|
||||
"vimConnectionInfo": {"id": "testid"}}
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s' % uuidsentinel.instance_id)
|
||||
rule_name = policies.VNFLCM % 'update_vnf'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.update,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_vnf')
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._instantiate')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_instantiate_vnf(
|
||||
self, mock_vnf_by_id, mock_instantiate, mock_vnf):
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf.return_value = utils.get_dummy_vnf()
|
||||
mock_instantiate.return_value = {
|
||||
'status': 202, 'Location': 'vnf status check link'}
|
||||
body = {"flavourId": "simple",
|
||||
"instantiationLevelId": "instantiation_level_1"}
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s/instantiate' % uuidsentinel.instance_id)
|
||||
rule_name = policies.VNFLCM % 'instantiate'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.instantiate,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_vnf')
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._terminate')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_terminate_vnf(self, mock_vnf_by_id, mock_terminate, mock_vnf):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s/terminate' % uuidsentinel.instance_id)
|
||||
body = {'terminationType': 'GRACEFUL',
|
||||
'gracefulTerminationTimeout': 10}
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf.return_value = utils.get_dummy_vnf()
|
||||
mock_terminate.return_value = {
|
||||
'status': 202, 'Location': 'vnf status check link'}
|
||||
rule_name = policies.VNFLCM % 'terminate'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.terminate,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_vnf')
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._delete')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_delete_vnf(self, mock_vnf_by_id, mock_delete, mock_vnf):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s' % uuidsentinel.instance_id)
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf.return_value = utils.get_dummy_vnf()
|
||||
rule_name = policies.VNFLCM % 'delete'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.delete,
|
||||
req, uuidsentinel.instance_id)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_vnf')
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._heal')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_heal_vnf(self, mock_vnf_by_id, mock_heal, mock_vnf):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s/heal' % uuidsentinel.instance_id)
|
||||
body = {'cause': 'healing'}
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf.return_value = utils.get_dummy_vnf()
|
||||
rule_name = policies.VNFLCM % 'heal'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.heal,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._scale')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_scale_vnf(self, mock_vnf_by_id, mock_scale):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_instances/%s/scale' % uuidsentinel.instance_id)
|
||||
body = {
|
||||
"type": "SCALE_OUT",
|
||||
"aspectId": "SP1",
|
||||
"numberOfSteps": 1,
|
||||
"additionalParams": {
|
||||
"test": "test_value"}}
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
rule_name = policies.VNFLCM % 'scale'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.scale,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "get_by_id")
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_rollback_vnf')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_rollback_vnf(self, mock_vnf_by_id, mock_rollback,
|
||||
mock_lcm_by_id):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_lcm_op_occs/%s/rollback' % uuidsentinel.instance_id)
|
||||
mock_lcm_by_id.return_value = fakes.vnflcm_rollback_active()
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
rule_name = policies.VNFLCM % 'rollback'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.rollback,
|
||||
req, uuidsentinel.instance_id)
|
||||
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "get_by_id")
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
@mock.patch.object(objects.VnfInstance, "save")
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "save")
|
||||
def test_cancel_vnf(self, mock_save_occ, mock_vnf, mock_vnf_by_id,
|
||||
mock_lcm_by_id):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_lcm_op_occs/%s/cancel' % uuidsentinel.instance_id)
|
||||
body = {'cancelMode': 'FORCEFUL'}
|
||||
mock_lcm_by_id.return_value = fakes.vnflcm_cancel_insta()
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
rule_name = policies.VNFLCM % 'cancel'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.cancel,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "get_by_id")
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
@mock.patch.object(objects.VnfInstance, "save")
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "save")
|
||||
def test_fail_vnf(self, mock_save_occ, mock_vnf, mock_vnf_by_id,
|
||||
mock_lcm_by_id):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_lcm_op_occs/%s/fail' % uuidsentinel.instance_id)
|
||||
mock_lcm_by_id.return_value = fakes.vnflcm_fail_insta()
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
rule_name = policies.VNFLCM % 'fail'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.fail,
|
||||
req, uuidsentinel.instance_id)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_vnf')
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "get_by_id")
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
@mock.patch.object(controller.VnfLcmController, "_instantiate")
|
||||
def test_retry_vnf(self, mock_instantiate, mock_vnf_by_id,
|
||||
mock_lcm_by_id, mock_vnf):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_lcm_op_occs/%s/fail' % uuidsentinel.instance_id)
|
||||
mock_lcm_by_id.return_value = fakes.vnflcm_op_occs_retry_data()
|
||||
mock_vnf.return_value = utils.get_dummy_vnf()
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
tenant_id=self.project_id)
|
||||
rule_name = policies.VNFLCM % 'retry'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.retry,
|
||||
req, uuidsentinel.instance_id)
|
||||
|
||||
@mock.patch.object(objects.VnfLcmOpOcc, "get_by_id")
|
||||
def test_show_lcm_op_occs(self, mock_lcm_by_id):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnf_lcm_op_occs/%s' % uuidsentinel.instance_id)
|
||||
mock_lcm_by_id.return_value = fakes.return_vnf_lcm_opoccs_obj()
|
||||
rule_name = policies.VNFLCM % 'show_lcm_op_occs'
|
||||
self.common_policy_check(self.project_authorized_contexts,
|
||||
self.project_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.show_lcm_op_occs,
|
||||
req, uuidsentinel.instance_id)
|
||||
|
||||
@mock.patch.object(objects.VnfLcmOpOccList, "get_by_marker_filter")
|
||||
def test_list_lcm_op_occs(self, mock_op_occ_list):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnflcm/v1/vnf_lcm_op_occs')
|
||||
rule_name = policies.VNFLCM % 'list_lcm_op_occs'
|
||||
self.common_policy_check(self.project_authorized_contexts,
|
||||
self.project_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.list_lcm_op_occs,
|
||||
req)
|
||||
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._get_vnf')
|
||||
@mock.patch('tacker.api.vnflcm.v1.controller.'
|
||||
'VnfLcmController._change_ext_conn')
|
||||
@mock.patch.object(objects.VnfInstance, "get_by_id")
|
||||
def test_change_ext_conn_vnf(self, mock_vnf_by_id,
|
||||
mock_change_ext_conn, mock_vnf):
|
||||
req = fake_request.HTTPRequest.blank(
|
||||
'/vnflcm/v1/vnf_instances/%s/change_ext_conn' %
|
||||
uuidsentinel.instance_id)
|
||||
body = fakes.get_change_ext_conn_request_body()
|
||||
mock_vnf_by_id.return_value = fakes.return_vnf_instance(
|
||||
fields.VnfInstanceState.INSTANTIATED,
|
||||
tenant_id=self.project_id)
|
||||
mock_vnf.return_value = utils.get_dummy_vnf()
|
||||
rule_name = policies.VNFLCM % 'change_ext_conn'
|
||||
self.common_policy_check(self.project_member_authorized_contexts,
|
||||
self.project_member_unauthorized_contexts,
|
||||
rule_name,
|
||||
self.controller.change_ext_conn,
|
||||
req, uuidsentinel.instance_id,
|
||||
body=body)
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue