openstack
/
tacker
Code Issues Proposed changes

Add RBAC tests for VNF LCM APIs 

Current tests do not have good test coverage of VNF LCM
APIs policies. Either tests for policies do not exist or
if they exist then they do not cover the actual negative
and positive testing.

Basically this commit does the following:

* Add RBAC tests:
  As we are implementing the project personas (project member
  and reader role) in policies, we need to have the enough
  testing coverage of existing policy behavior and to know
  that with new defaults how the access permissions will
  looks like.

* Pass correct target to oslo policy:
  Currently, APIs are not passing the right targets to oslo
  policy, means VNF instance project_id was not passed as target.
  We need to pass the project_id so that we can check the 'onwer'
  permission correctly at RBAC level and RBAC checks pass and
  request goes to fetch the data from DB where project_id
  is checked. For example, GET VNF API requests by a non
  admin user does not check if requester users is from same
  project of requested VNF or not and request pass the oslo
  policy checks and make DB request. Passing the right project_id
  in oslo policy will return the request (if projectA request projectB
  VNF) from policy checks itself. This can be seen in modified
  test_controller.py tests.

Partial implement blueprint implement-project-personas

Change-Id: Ib022397f715c6aa08718a6867d2f2ea19c517c00
This commit is contained in:
Ghanshyam Mann 2024-02-08 23:05:56 -08:00 committed by Ghanshyam
parent 9eac5d363f
commit c2ef23210f
3 changed files with 690 additions and 196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
tacker/api/vnflcm/v1/controller.py
View File

@ -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(

341
tacker/tests/unit/policies/test_vnf_lcm.py
View File

@ -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)

459
tacker/tests/unit/vnflcm/test_controller.py
View File

File diff suppressed because it is too large Load Diff