openstack/tacker
Code Issues Proposed changes

OAuth 2.0 Mutual-TLS Support

Browse Source 
Provide the option to use mutual TLS client authentication when
accessing external servers from Tacker. Oauth2MtlsAuthHandle has been
added to support Mutual-TLS client authentication for access from Tacker
to external NFVO servers and notification endpoints using user-provided
Mutual-TLS client certificates.

Implements: blueprint support-oauth2-mtls
Change-Id: Ib1b33bccac85ba8c68aeebd460876bb38a4917fa
This commit is contained in:
Yusuke Niimi 2022-10-14 08:56:06 +00:00
parent 692ec77688
commit 4e699aec35
21 changed files with 598 additions and 215 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.zuul.yaml
View File

@ -349,6 +349,7 @@
token_endpoint: http://127.0.0.1:9990
client_id: 229ec984de7547b2b662e968961af5a4
client_password: devstack
use_client_secret_basic: True
tox_envlist: dsvm-functional-sol-separated-nfvo-v2
# TODO(manpreetk): As per the 2023.1 testing runtime, we need to run atleast

10
releasenotes/notes/bp-support-oauth2-mtls-1ef757cca82059cf.yaml Normal file
View File

@ -0,0 +1,10 @@
---
features:
- |
[`blueprint support-oauth2-mtls <https://blueprints.launchpad.net/keystone/+spec/support-oauth2-mtls>`_]
Provide the option to use mutual TLS client authentication when accessing
external servers from Tacker. OAuth2MtlsAuthHandle has been added to
support Mutual-TLS client authentication for access from Tacker to external
NFVO servers and notification endpoints using user-provided Mutual-TLS
client certificates.

2
tacker/alarm_receiver.py
View File

@ -17,7 +17,6 @@ from urllib import parse
from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import strutils
from tacker._i18n import _
from tacker.vnfm.monitor_drivers.token import Token
@ -49,7 +48,6 @@ def config_opts():
class AlarmReceiver(wsgi.Middleware):
def process_request(self, req):
LOG.debug('Process request: %s', strutils.mask_password(req))
if req.method != 'POST':
return
url = req.url

19
tacker/sol_refactored/api/schemas/common_types.py
View File

@ -122,7 +122,7 @@ _IpAddresses = {
'additionalProperties': True
}
# SOL013 8.3.4
# SOL013 v3.5.1 8.3.4
SubscriptionAuthentication = {
'type': 'object',
'properties': {
@ -133,7 +133,7 @@ SubscriptionAuthentication = {
'enum': [
'BASIC',
'OAUTH2_CLIENT_CREDENTIALS',
'TLS_CERT']
'OAUTH2_CLIENT_CERT']
}
},
'paramsBasic': {
@ -150,6 +150,21 @@ SubscriptionAuthentication = {
'clientPassword': {'type': 'string'},
'tokenEndpoint': {'type': 'string'}
}
},
'paramsOauth2ClientCert': {
'type': 'object',
'properties': {
'clientId': {'type': 'string'},
'certificateRef': {'type': 'object',
'properties': {
'type': {'type': 'string'},
'value': {'type': 'string'}
},
'required': ['type', 'value']
},
'tokenEndpoint': {'type': 'string'}
},
'required': ['clientId', 'certificateRef', 'tokenEndpoint']
}
},
'required': ['authType'],

20
tacker/sol_refactored/api/schemas/vnflcm_v2.py
View File

@ -157,7 +157,7 @@ ChangeExtVnfConnectivityRequest_V200 = {
'additionalProperties': True,
}
# SOL013 8.3.4
# SOL013 v3.5.1 8.3.4
_SubscriptionAuthentication = {
'type': 'object',
'properties': {
@ -168,7 +168,7 @@ _SubscriptionAuthentication = {
'enum': [
'BASIC',
'OAUTH2_CLIENT_CREDENTIALS',
'TLS_CERT']
'OAUTH2_CLIENT_CERT']
}
},
'paramsBasic': {
@ -191,6 +191,22 @@ _SubscriptionAuthentication = {
# NOTE: must be specified since the way to specify them out of
# band is not supported.
'required': ['clientId', 'clientPassword', 'tokenEndpoint']
},
'paramsOauth2ClientCert': {
'type': 'object',
'properties': {
'clientId': {'type': 'string'},
'certificateRef': {
'type': 'object',
'properties': {
'type': {'type': 'string'},
'value': {'type': 'string'}
},
'required': ['type', 'value']
},
'tokenEndpoint': {'type': 'string'}
},
'required': ['clientId', 'certificateRef', 'tokenEndpoint']
}
},
'required': ['authType'],

22
tacker/sol_refactored/common/config.py
View File

@ -50,6 +50,14 @@ VNFM_OPTS = [
default=0, # 0 means no paging
help=_('Paged response size of the query result '
'for VNF LCM operation occurrences.')),
cfg.StrOpt('notification_mtls_ca_cert_file',
default='',
help=_('CA Certificate file used by OAuth2.0 mTLS '
'authentication.')),
cfg.StrOpt('notification_mtls_client_cert_file',
default='',
help=_('Client Certificate file used by OAuth2.0 mTLS '
'authentication.')),
cfg.IntOpt('notify_connect_retries',
default=0, # 0 means no retry
help=_('Number of retries that should be attempted for '
@ -121,13 +129,25 @@ NFVO_OPTS = [
cfg.StrOpt('vnf_package_cache_dir',
default='/opt/stack/data/tacker/vnf_package_cache',
help=_('Vnf package content cache directory.')),
cfg.StrOpt('mtls_ca_cert_file',
default='',
help=_('CA Certificate file used by OAuth2.0 mTLS '
'authentication.')),
cfg.StrOpt('mtls_client_cert_file',
default='',
help=_('Client Certificate file used by OAuth2.0 mTLS '
'authentication.')),
cfg.BoolOpt('test_callback_uri',
default=True,
help=_('Check to get notification from callback Uri.')),
cfg.ListOpt('test_grant_zone_list',
default=["nova"],
help=_('Zones used for test which returned in Grant '
'response.'))
'response.')),
cfg.BoolOpt('use_client_secret_basic',
default=False,
help=_('Use password authenticatiojn if True, '
'use certificate authentication if False.'))
]
CONF.register_opts(NFVO_OPTS, 'v2_nfvo')

4
tacker/sol_refactored/common/exceptions.py
View File

@ -378,6 +378,10 @@ class OIDCAuthFailed(SolHttpError400):
" Detail: %(detail)s")
class AuthTypeNotFound(SolHttpError400):
message = _("AuthType parameter %(auth_type)s is not found")
class HelmOperationFailed(SolHttpError422):
title = 'Helm operation failed'
# detail set in the code

82
tacker/sol_refactored/common/http_client.py
View File

@ -15,6 +15,7 @@
import abc
import urllib
from keystoneauth1 import adapter
from keystoneauth1 import http_basic
@ -188,10 +189,10 @@ class NoAuthHandle(AuthHandle):
return session.Session(auth=auth, verify=False)
class Oauth2AuthPlugin(plugin.FixedEndpointPlugin):
class OAuth2AuthPlugin(plugin.FixedEndpointPlugin):
def __init__(self, endpoint, token_endpoint, client_id, client_password):
super(Oauth2AuthPlugin, self).__init__(endpoint)
super(OAuth2AuthPlugin, self).__init__(endpoint)
self.token_endpoint = token_endpoint
self.client_id = client_id
self.client_password = client_password
@ -230,10 +231,85 @@ class OAuth2AuthHandle(AuthHandle):
self.client_password = client_password
def get_auth(self, context=None):
return Oauth2AuthPlugin(self.endpoint, self.token_endpoint,
return OAuth2AuthPlugin(self.endpoint, self.token_endpoint,
self.client_id, self.client_password)
def get_session(self, auth, service_type):
_session = session.Session(auth=auth, verify=False)
return adapter.Adapter(session=_session,
service_type=service_type)
class CertAuthMtlsHandle(AuthHandle):
def __init__(self, endpoint, verify_cert, client_cert):
self.endpoint = endpoint
self.verify_cert = verify_cert
self.client_cert = client_cert
def get_auth(self, context=None):
return noauth.NoAuth(endpoint=self.endpoint)
def get_session(self, auth, service_type):
return session.Session(auth=auth, verify=self.verify_cert,
cert=self.client_cert)
class OAuth2MtlsAuthPlugin(plugin.FixedEndpointPlugin):
def __init__(self, endpoint, token_endpoint, client_id,
verify_cert, client_cert):
super(OAuth2MtlsAuthPlugin, self).__init__(endpoint)
self.token_endpoint = token_endpoint
self.client_id = client_id
self.verify_cert = verify_cert
self.client_cert = client_cert
def get_token(self, session, **kwargs):
auth = CertAuthMtlsHandle(self.endpoint, self.verify_cert,
self.client_cert)
client = HttpClient(auth)
url = f'{self.token_endpoint}'
data = {
'grant_type': 'client_credentials',
'client_id': self.client_id
}
data = urllib.parse.urlencode(data)
resp, resp_body = client.do_request(url, "POST",
body=data, content_type='application/x-www-form-urlencoded')
if resp.status_code != 200:
LOG.error("get OAuth2 mTLS token failed: %d" % resp.status_code)
return
return resp_body['access_token']
def get_headers(self, session, **kwargs):
token = self.get_token(session)
if not token:
return None
auth = 'Bearer %s' % token
return {'Authorization': auth}
class OAuth2MtlsAuthHandle(AuthHandle):
def __init__(self, endpoint, token_endpoint, client_id,
verify_cert, client_cert):
self.endpoint = endpoint
self.token_endpoint = token_endpoint
self.client_id = client_id
self.verify_cert = verify_cert
self.client_cert = client_cert
def get_auth(self, context=None):
return OAuth2MtlsAuthPlugin(self.endpoint, self.token_endpoint,
self.client_id, self.verify_cert, self.client_cert)
def get_session(self, auth, service_type):
_session = session.Session(auth=auth, verify=self.verify_cert,
cert=self.client_cert)
return adapter.Adapter(session=_session,
service_type=service_type)

6
tacker/sol_refactored/common/pm_job_utils.py
View File

@ -103,6 +103,12 @@ def _get_notification_auth_handle(pm_job):
param = pm_job.authentication.paramsOauth2ClientCredentials
return http_client.OAuth2AuthHandle(
None, param.tokenEndpoint, param.clientId, param.clientPassword)
if pm_job.authentication.obj_attr_is_set('paramsOauth2ClientCert'):
param = pm_job.authentication.paramsOauth2ClientCert
verify_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
client_cert = CONF.v2_vnfm.notification_mtls_client_cert_file
return http_client.OAuth2MtlsAuthHandle(None,
param.tokenEndpoint, param.clientId, verify_cert, client_cert)
return None

74
tacker/sol_refactored/common/subscription_utils.py
View File

@ -51,16 +51,28 @@ def subsc_href(subsc_id, endpoint):
def _get_notification_auth_handle(subsc):
if not subsc.obj_attr_is_set('authentication'):
auth_req = subsc.get('authentication', None)
if auth_req:
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if 'OAUTH2_CLIENT_CERT' in auth.authType:
param = subsc.authentication.paramsOauth2ClientCert
verify_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
client_cert = CONF.v2_vnfm.notification_mtls_client_cert_file
return http_client.OAuth2MtlsAuthHandle(None,
param.tokenEndpoint, param.clientId, verify_cert, client_cert)
elif 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
param = subsc.authentication.paramsOauth2ClientCredentials
return http_client.OAuth2AuthHandle(None,
param.tokenEndpoint, param.clientId, param.clientPassword)
elif 'BASIC' in auth.authType:
param = subsc.authentication.paramsBasic
return http_client.BasicAuthHandle(param.userName, param.password)
else:
raise sol_ex.AuthTypeNotFound(auth.authType)
else:
return http_client.NoAuthHandle()
elif subsc.authentication.obj_attr_is_set('paramsBasic'):
param = subsc.authentication.paramsBasic
return http_client.BasicAuthHandle(param.userName, param.password)
elif subsc.authentication.obj_attr_is_set(
'paramsOauth2ClientCredentials'):
param = subsc.authentication.paramsOauth2ClientCredentials
return http_client.OAuth2AuthHandle(None,
param.tokenEndpoint, param.clientId, param.clientPassword)
# not reach here
@ -273,3 +285,47 @@ def make_delete_inst_notif_data(subsc, inst, endpoint):
# vnfLcmOpOcc: is not necessary
)
return notif_data
def check_http_client_auth(auth_req):
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if 'OAUTH2_CLIENT_CERT' in auth.authType:
oauth2_mtls_req = auth_req.get('paramsOauth2ClientCert')
if oauth2_mtls_req is None:
msg = "paramsOauth2ClientCert must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsOauth2ClientCert = (
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId=oauth2_mtls_req.get('clientId'),
certificateRef=oauth2_mtls_req.get('certificateRef'),
tokenEndpoint=oauth2_mtls_req.get('tokenEndpoint')
)
)
elif 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
if oauth2_req is None:
msg = "paramsOauth2ClientCredentials must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsOauth2ClientCredentials = (
objects.SubscriptionAuthentication_ParamsOauth2(
clientId=oauth2_req.get('clientId'),
clientPassword=oauth2_req.get('clientPassword'),
tokenEndpoint=oauth2_req.get('tokenEndpoint')
)
)
elif 'BASIC' in auth.authType:
basic_req = auth_req.get('paramsBasic')
if basic_req is None:
msg = "ParamsBasic must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsBasic = (
objects.SubscriptionAuthentication_ParamsBasic(
userName=basic_req.get('userName'),
password=basic_req.get('password')
)
)
else:
raise sol_ex.AuthTypeNotFound(auth.authType)
return auth

48
tacker/sol_refactored/controller/vnffm_v1.py
View File

@ -24,7 +24,9 @@ from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import coordinate
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import fm_alarm_utils
from tacker.sol_refactored.common import fm_subscription_utils as subsc_utils
from tacker.sol_refactored.common import fm_subscription_utils\
as fm_subsc_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.controller import vnffm_view
from tacker.sol_refactored.nfvo import nfvo_client
from tacker.sol_refactored import objects
@ -117,47 +119,15 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
auth_req = body.get('authentication')
if auth_req:
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if 'BASIC' in auth.authType:
basic_req = auth_req.get('paramsBasic')
if basic_req is None:
msg = "ParamsBasic must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsBasic = (
objects.SubscriptionAuthentication_ParamsBasic(
userName=basic_req.get('userName'),
password=basic_req.get('password')
)
)
if 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
if oauth2_req is None:
msg = "paramsOauth2ClientCredentials must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsOauth2ClientCredentials = (
objects.SubscriptionAuthentication_ParamsOauth2(
clientId=oauth2_req.get('clientId'),
clientPassword=oauth2_req.get('clientPassword'),
tokenEndpoint=oauth2_req.get('tokenEndpoint')
)
)
if 'TLS_CERT' in auth.authType:
msg = "'TLS_CERT' is not supported at the moment."
raise sol_ex.InvalidSubscription(sol_detail=msg)
subsc.authentication = auth
subsc.authentication = subsc_utils.check_http_client_auth(auth_req)
if CONF.v2_nfvo.test_callback_uri:
subsc_utils.test_notification(subsc)
fm_subsc_utils.test_notification(subsc)
subsc.create(context)
resp_body = self._subsc_view.detail(subsc)
self_href = subsc_utils.subsc_href(subsc.id, self.endpoint)
self_href = fm_subsc_utils.subsc_href(subsc.id, self.endpoint)
return sol_wsgi.SolResponse(
201, resp_body, version=api_version.CURRENT_FM_VERSION,
@ -173,7 +143,7 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
page_size = CONF.v2_vnfm.subscription_page_size
pager = self._subsc_view.parse_pager(request, page_size)
subscs = subsc_utils.get_subsc_all(request.context,
subscs = fm_subsc_utils.get_subsc_all(request.context,
marker=pager.marker)
resp_body = self._subsc_view.detail_list(subscs, filters, None, pager)
@ -183,7 +153,7 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
link=pager.get_link())
def subscription_show(self, request, id):
subsc = subsc_utils.get_subsc(request.context, id)
subsc = fm_subsc_utils.get_subsc(request.context, id)
resp_body = self._subsc_view.detail(subsc)
@ -192,7 +162,7 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
def subscription_delete(self, request, id):
context = request.context
subsc = subsc_utils.get_subsc(request.context, id)
subsc = fm_subsc_utils.get_subsc(request.context, id)
subsc.delete(context)

34
tacker/sol_refactored/controller/vnflcm_v2.py
View File

@ -486,39 +486,7 @@ class VnfLcmControllerV2(sol_wsgi.SolAPIController):
auth_req = body.get('authentication')
if auth_req:
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if 'BASIC' in auth.authType:
basic_req = auth_req.get('paramsBasic')
if basic_req is None:
msg = "ParamsBasic must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsBasic = (
objects.SubscriptionAuthentication_ParamsBasic(
userName=basic_req.get('userName'),
password=basic_req.get('password')
)
)
if 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
if oauth2_req is None:
msg = "paramsOauth2ClientCredentials must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsOauth2ClientCredentials = (
objects.SubscriptionAuthentication_ParamsOauth2(
clientId=oauth2_req.get('clientId'),
clientPassword=oauth2_req.get('clientPassword'),
tokenEndpoint=oauth2_req.get('tokenEndpoint')
)
)
if 'TLS_CERT' in auth.authType:
msg = "'TLS_CERT' is not supported at the moment."
raise sol_ex.InvalidSubscription(sol_detail=msg)
subsc.authentication = auth
subsc.authentication = subsc_utils.check_http_client_auth(auth_req)
if CONF.v2_nfvo.test_callback_uri:
subsc_utils.test_notification(subsc)

42
tacker/sol_refactored/controller/vnfpm_v2.py
View File

@ -28,6 +28,7 @@ from tacker.sol_refactored.common import coordinate
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import monitoring_plugin_base as plugin
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.common import vnf_instance_utils as inst_utils
from tacker.sol_refactored.controller import vnfpm_view
from tacker.sol_refactored.nfvo import nfvo_client
@ -58,42 +59,6 @@ OBJ_TYPE_TO_METRIC_LISt = {
}
def _check_http_client_auth(auth_req):
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if 'BASIC' in auth.authType:
basic_req = auth_req.get('paramsBasic')
if basic_req is None:
msg = "ParamsBasic must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsBasic = (
objects.SubscriptionAuthentication_ParamsBasic(
userName=basic_req.get('userName'),
password=basic_req.get('password')
)
)
if 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
if oauth2_req is None:
msg = "paramsOauth2ClientCredentials must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
auth.paramsOauth2ClientCredentials = (
objects.SubscriptionAuthentication_ParamsOauth2(
clientId=oauth2_req.get('clientId'),
clientPassword=oauth2_req.get('clientPassword'),
tokenEndpoint=oauth2_req.get('tokenEndpoint')
)
)
if 'TLS_CERT' in auth.authType:
msg = "'TLS_CERT' is not supported at the moment."
raise sol_ex.InvalidSubscription(sol_detail=msg)
return auth
def _check_performance_metric_or_group(
obj_type, metric_group, performance_metric):
# Check whether the object_type is consistent with the corresponding
@ -186,7 +151,8 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
# authentication
auth_req = body.get('authentication')
if auth_req:
pm_job.authentication = _check_http_client_auth(auth_req)
pm_job.authentication = subsc_utils.check_http_client_auth(
auth_req)
# metadata
metadata = body.get('metadata')
@ -246,7 +212,7 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
if body.get("callbackUri"):
pm_job.callbackUri = body.get("callbackUri")
if body.get("authentication"):
pm_job.authentication = _check_http_client_auth(
pm_job.authentication = subsc_utils.check_http_client_auth(
body.get("authentication"))
if CONF.v2_nfvo.test_callback_uri:

18
tacker/sol_refactored/nfvo/nfvo_client.py
View File

@ -47,11 +47,19 @@ class NfvoClient(object):
if CONF.v2_nfvo.use_external_nfvo:
self.is_local = False
self.endpoint = CONF.v2_nfvo.endpoint
auth_handle = http_client.OAuth2AuthHandle(
self.endpoint,
CONF.v2_nfvo.token_endpoint,
CONF.v2_nfvo.client_id,
CONF.v2_nfvo.client_password)
if CONF.v2_nfvo.use_client_secret_basic:
auth_handle = http_client.OAuth2AuthHandle(
self.endpoint,
CONF.v2_nfvo.token_endpoint,
CONF.v2_nfvo.client_id,
CONF.v2_nfvo.client_password)
else:
auth_handle = http_client.OAuth2MtlsAuthHandle(
self.endpoint,
CONF.v2_nfvo.token_endpoint,
CONF.v2_nfvo.client_id,
CONF.v2_nfvo.mtls_ca_cert_file,
CONF.v2_nfvo.mtls_client_cert_file)
self.client = http_client.HttpClient(auth_handle)
self.grant_api_version = CONF.v2_nfvo.grant_api_version
self.vnfpkgm_api_version = CONF.v2_nfvo.vnfpkgm_api_version

32
tacker/sol_refactored/objects/common/subscription_authentication.py
View File

@ -31,6 +31,7 @@ class SubscriptionAuthentication(base.TackerObject,
valid_values=[
'BASIC',
'OAUTH2_CLIENT_CREDENTIALS',
'OAUTH2_CLIENT_CERT',
'TLS_CERT',
],
nullable=False),
@ -38,6 +39,9 @@ class SubscriptionAuthentication(base.TackerObject,
'SubscriptionAuthentication_ParamsBasic', nullable=True),
'paramsOauth2ClientCredentials': fields.ObjectField(
'SubscriptionAuthentication_ParamsOauth2', nullable=True),
'paramsOauth2ClientCert': fields.ObjectField(
'SubscriptionAuthentication_ParamsOauth2ClientCert',
nullable=True),
}
@ -70,3 +74,31 @@ class SubscriptionAuthentication_ParamsOauth2(
'clientPassword': fields.StringField(nullable=True),
'tokenEndpoint': fields.UriField(nullable=True),
}
@base.TackerObjectRegistry.register
class SubscriptionAuthentication_ParamsOauth2ClientCert(
base.TackerObject, base.TackerObjectDictCompat):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'clientId': fields.StringField(nullable=False),
'cerficateRef': fields.ObjectField(
'ParamsOauth2ClientCert_CertificateRef', nullable=False),
'tokenEndpoint': fields.UriField(nullable=False),
}
@base.TackerObjectRegistry.register
class ParamsOauth2ClientCert_CertificateRef(
base.TackerObject, base.TackerObjectDictCompat):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'type': fields.StringField(nullable=False),
'value': fields.StringField(nullable=False),
}

34
tacker/tests/unit/sol_refactored/common/test_pm_job_utils.py
View File

@ -136,9 +136,7 @@ class TestPmJobUtils(base.BaseTestCase):
id='pm_job_1',
authentication=pm_job_1_auth)
result = pm_job_utils._get_notification_auth_handle(pm_job_1)
res = type(result).__name__
name = type(http_client.BasicAuthHandle('test', 'test')).__name__
self.assertEqual(name, res)
self.assertIsInstance(result, http_client.BasicAuthHandle)
pm_job_2 = objects.PmJobV2(
id='pm_job_2',
@ -153,19 +151,27 @@ class TestPmJobUtils(base.BaseTestCase):
)
)
result = pm_job_utils._get_notification_auth_handle(pm_job_2)
res = type(result).__name__
name = type(http_client.OAuth2AuthHandle(
None, 'tokenEndpoint', 'test', 'test')).__name__
self.assertEqual(name, res)
self.assertIsInstance(result, http_client.OAuth2AuthHandle)
pm_job_3 = objects.PmJobV2(id='pm_job_3',
authentication=(
objects.SubscriptionAuthentication(
authType=["TLS_CERT"],
))
)
pm_job_3 = objects.PmJobV2(
id='pm_job_3',
authentication=objects.SubscriptionAuthentication(
authType=["OAUTH2_CLIENT_CERT"],
paramsOauth2ClientCert=(
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId='test',
certificateRef=objects.
ParamsOauth2ClientCert_CertificateRef(
type='x5t#256',
value='03c6e188d1fe5d3da8c9bc9a8dc531a2'
'b3ecf812b03aede9bec7ba1b410b6b64'
),
tokenEndpoint='http://127.0.0.1/token'
))
)
)
result = pm_job_utils._get_notification_auth_handle(pm_job_3)
self.assertEqual(None, result)
self.assertIsInstance(result, http_client.OAuth2MtlsAuthHandle)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification(self, mock_do_request):

82
tacker/tests/unit/sol_refactored/common/test_subscription_utils.py
View File

@ -74,6 +74,7 @@ class TestSubscriptionUtils(base.BaseTestCase):
id='sub-2', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['BASIC'],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test')))
@ -84,6 +85,7 @@ class TestSubscriptionUtils(base.BaseTestCase):
id='sub-3', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['OAUTH2_CLIENT_CREDENTIALS'],
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test', clientPassword='test',
@ -92,6 +94,25 @@ class TestSubscriptionUtils(base.BaseTestCase):
# execute oauth2
subsc_utils.send_notification(subsc_oauth2, notif_data_no_auth)
subsc_oauth2_mtls = objects.LccnSubscriptionV2(
id='sub-4', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=["OAUTH2_CLIENT_CERT"],
paramsOauth2ClientCert=(
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId='test',
certificateRef=objects.
ParamsOauth2ClientCert_CertificateRef(
type='x5t#256',
value='03c6e188d1fe5d3da8c9bc9a8dc531a2'
'b3ecf812b03aede9bec7ba1b410b6b64'
),
tokenEndpoint='http://127.0.0.1/token'))))
# execute oauth2 mtls
subsc_utils.send_notification(subsc_oauth2_mtls, notif_data_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_error_code(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
@ -389,3 +410,64 @@ class TestSubscriptionUtils(base.BaseTestCase):
self.assertEqual('subsc-1', result.subscriptionId)
self.assertEqual('test-instance', result.vnfInstanceId)
def test_check_http_client_auth(self):
auth_req_1 = {
'authType': ['BASIC'],
'paramsBasic': {
'userName': 'test',
'password': 'test'
},
}
result = subsc_utils.check_http_client_auth(auth_req_1)
self.assertEqual(['BASIC'], result.authType)
auth_req_2 = {
'authType': ['OAUTH2_CLIENT_CREDENTIALS'],
'paramsOauth2ClientCredentials': {
'clientId': 'test',
'clientPassword': 'test',
'tokenEndpoint':
'http://127.0.0.1/token'
}
}
result = subsc_utils.check_http_client_auth(auth_req_2)
self.assertEqual(['OAUTH2_CLIENT_CREDENTIALS'], result.authType)
auth_req_3 = {
'authType': ['OAUTH2_CLIENT_CERT'],
'paramsOauth2ClientCert': {
'clientId': 'test',
'certificateRef': {
'type': 'x5t#256',
'value': '03c6e188d1fe5d3da8c9bc9a8dc531a2'
'b3ecf812b03aede9bec7ba1b410b6b64'
},
'tokenEndpoint': 'http://127.0.0.1/token'
}
}
result = subsc_utils.check_http_client_auth(auth_req_3)
self.assertEqual(['OAUTH2_CLIENT_CERT'], result.authType)
def test_check_http_client_auth_error(self):
auth_req_1 = {
'authType': ['BASIC'],
'paramsBasic': None
}
self.assertRaises(sol_ex.InvalidSubscription,
subsc_utils.check_http_client_auth,
auth_req=auth_req_1)
auth_req_2 = {
'authType': ['OAUTH2_CLIENT_CREDENTIALS'],
}
self.assertRaises(sol_ex.InvalidSubscription,
subsc_utils.check_http_client_auth,
auth_req=auth_req_2)
auth_req_3 = {
'authType': ['OAUTH2_CLIENT_CERT']
}
self.assertRaises(sol_ex.InvalidSubscription,
subsc_utils.check_http_client_auth,
auth_req=auth_req_3)

84
tacker/tests/unit/sol_refactored/controller/test_vnffm_v1.py
View File

@ -111,14 +111,61 @@ class TestVnffmV1(base.BaseTestCase):
@mock.patch.object(objects.base.TackerPersistentObject, 'create')
@mock.patch.object(subsc_utils, 'test_notification')
def test_subscription_create(self, mock_test, mock_create):
body = {
body_1 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["BASIC", "OAUTH2_CLIENT_CREDENTIALS"],
"authType": ["BASIC", "OAUTH2_CLIENT_CREDENTIALS",
"OAUTH2_CLIENT_CERT"],
"paramsBasic": {
"userName": "test",
"password": "test"
},
"paramsOauth2ClientCredentials": {
"clientId": "test",
"clientPassword": "test",
"tokenEndpoint": "https://127.0.0.1/token"
},
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
},
"filter": fakes_for_fm.fm_subsc_example['filter']
}
result = self.controller.subscription_create(
request=self.request, body=body_1)
self.assertEqual(201, result.status)
self.assertEqual(body_1['callbackUri'], result.body['callbackUri'])
self.assertEqual(body_1['filter'], result.body['filter'])
self.assertIsNone(result.body.get('authentication'))
body_2 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["BASIC"],
"paramsBasic": {
"userName": "test",
"password": "test"
}
},
"filter": fakes_for_fm.fm_subsc_example['filter']
}
result = self.controller.subscription_create(
request=self.request, body=body_2)
self.assertEqual(201, result.status)
self.assertEqual(body_2['callbackUri'], result.body['callbackUri'])
self.assertEqual(body_2['filter'], result.body['filter'])
self.assertIsNone(result.body.get('authentication'))
body_3 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["OAUTH2_CLIENT_CREDENTIALS"],
"paramsOauth2ClientCredentials": {
"clientId": "test",
"clientPassword": "test",
@ -128,10 +175,33 @@ class TestVnffmV1(base.BaseTestCase):
"filter": fakes_for_fm.fm_subsc_example['filter']
}
result = self.controller.subscription_create(
request=self.request, body=body)
request=self.request, body=body_3)
self.assertEqual(201, result.status)
self.assertEqual(body['callbackUri'], result.body['callbackUri'])
self.assertEqual(body['filter'], result.body['filter'])
self.assertEqual(body_3['callbackUri'], result.body['callbackUri'])
self.assertEqual(body_3['filter'], result.body['filter'])
self.assertIsNone(result.body.get('authentication'))
body_4 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["OAUTH2_CLIENT_CERT"],
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
},
"filter": fakes_for_fm.fm_subsc_example['filter']
}
result = self.controller.subscription_create(
request=self.request, body=body_4)
self.assertEqual(201, result.status)
self.assertEqual(body_4['callbackUri'], result.body['callbackUri'])
self.assertEqual(body_4['filter'], result.body['filter'])
self.assertIsNone(result.body.get('authentication'))
def test_invalid_subscripion(self):
@ -161,13 +231,13 @@ class TestVnffmV1(base.BaseTestCase):
body = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["TLS_CERT"]
"authType": ["OAUTH2_CLIENT_CERT"]
}
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
self.controller.subscription_create, request=self.request,
body=body)
self.assertEqual("'TLS_CERT' is not supported at the moment.",
self.assertEqual("paramsOauth2ClientCert must be specified.",
ex.detail)
@mock.patch.object(subsc_utils, 'get_subsc_all')

91
tacker/tests/unit/sol_refactored/controller/test_vnflcm_v2.py
View File

@ -539,8 +539,8 @@ class TestVnflcmV2(db_base.SqlTestCase):
body=body)
self.assertEqual(202, result.status)
def test_invalid_subscripion(self):
body = {
def test_invalid_subscription(self):
body_1 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["BASIC"]
@ -548,10 +548,10 @@ class TestVnflcmV2(db_base.SqlTestCase):
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
self.controller.subscription_create, request=self.request,
body=body)
body=body_1)
self.assertEqual("ParamsBasic must be specified.", ex.detail)
body = {
body_2 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["OAUTH2_CLIENT_CREDENTIALS"]
@ -559,32 +559,77 @@ class TestVnflcmV2(db_base.SqlTestCase):
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
self.controller.subscription_create, request=self.request,
body=body)
body=body_2)
self.assertEqual("paramsOauth2ClientCredentials must be specified.",
ex.detail)
body = {
body_3 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["TLS_CERT"]
"authType": ["OAUTH2_CLIENT_CERT"]
}
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
self.controller.subscription_create, request=self.request,
body=body)
self.assertEqual("'TLS_CERT' is not supported at the moment.",
body=body_3)
self.assertEqual("paramsOauth2ClientCert must be specified.",
ex.detail)
@mock.patch.object(subsc_utils, 'test_notification')
def test_subscription_create_201(self, mock_test):
body = {
body_1 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["BASIC", "OAUTH2_CLIENT_CREDENTIALS"],
"authType": ["BASIC", "OAUTH2_CLIENT_CREDENTIALS",
"OAUTH2_CLIENT_CERT"],
"paramsBasic": {
"userName": "test",
"password": "test"
},
"paramsOauth2ClientCredentials": {
"clientId": "test",
"clientPassword": "test",
"tokenEndpoint": "https://127.0.0.1/token"
},
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
},
"filter": {
"operationTypes": [fields.LcmOperationType.INSTANTIATE]
}
}
result = self.controller.subscription_create(
request=self.request, body=body_1)
self.assertEqual(201, result.status)
body_2 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["BASIC"],
"paramsBasic": {
"userName": "test",
"password": "test"
}
},
"filter": {
"operationTypes": [fields.LcmOperationType.INSTANTIATE]
}
}
result = self.controller.subscription_create(
request=self.request, body=body_2)
self.assertEqual(201, result.status)
body_3 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["OAUTH2_CLIENT_CREDENTIALS"],
"paramsOauth2ClientCredentials": {
"clientId": "test",
"clientPassword": "test",
@ -596,7 +641,29 @@ class TestVnflcmV2(db_base.SqlTestCase):
}
}
result = self.controller.subscription_create(
request=self.request, body=body)
request=self.request, body=body_3)
self.assertEqual(201, result.status)
body_4 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
"authType": ["OAUTH2_CLIENT_CERT"],
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
},
"filter": {
"operationTypes": [fields.LcmOperationType.INSTANTIATE]
}
}
result = self.controller.subscription_create(
request=self.request, body=body_4)
self.assertEqual(201, result.status)
@mock.patch.object(subsc_utils, 'get_subsc_all')

45
tacker/tests/unit/sol_refactored/controller/test_vnfpm_v2.py
View File

@ -45,51 +45,6 @@ class TestVnfpmV2(base.BaseTestCase):
self.endpoint = CONF.v2_vnfm.endpoint
self._pm_job_view = vnfpm_view.PmJobViewBuilder(self.endpoint)
def test_check_http_client_auth(self):
auth_req = {
'authType': ['BASIC'],
'paramsBasic': None
}
self.assertRaises(sol_ex.InvalidSubscription,
vnfpm_v2._check_http_client_auth,
auth_req=auth_req)
auth_req_1 = {
'authType': ['BASIC'],
'paramsBasic': {
'userName': 'test',
'password': 'test'
},
}
result = vnfpm_v2._check_http_client_auth(auth_req_1)
self.assertEqual(['BASIC'], result.authType)
auth_req_2 = {
'authType': ['OAUTH2_CLIENT_CREDENTIALS'],
'paramsOauth2ClientCredentials': {
'clientId': 'test',
'clientPassword': 'test',
'tokenEndpoint':
'http://127.0.0.1/token'
}
}
result = vnfpm_v2._check_http_client_auth(auth_req_2)
self.assertEqual(['OAUTH2_CLIENT_CREDENTIALS'], result.authType)
auth_req_3 = {
'authType': ['OAUTH2_CLIENT_CREDENTIALS'],
}
self.assertRaises(sol_ex.InvalidSubscription,
vnfpm_v2._check_http_client_auth,
auth_req=auth_req_3)
auth_req_4 = {
'authType': ['TLS_CERT']
}
self.assertRaises(sol_ex.InvalidSubscription,
vnfpm_v2._check_http_client_auth,
auth_req=auth_req_4)
def test_check_performance_metric_or_group(self):
vnfpm_v2._check_performance_metric_or_group(
obj_type='Vnf',

63
tacker/tests/unit/sol_refactored/nfvo/test_nfvo_client.py
View File

@ -17,6 +17,7 @@ import requests
from unittest import mock
import zipfile
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import uuidutils
@ -314,11 +315,24 @@ class TestNfvoClient(base.BaseTestCase):
self.nfvo_client.grant_api_version = '1.4.0'
self.nfvo_client.vnfpkgm_api_version = '2.1.0'
cfg.CONF.set_override("use_external_nfvo", True, group="v2_nfvo")
cfg.CONF.set_override("mtls_ca_cert_file", "/path/to/cacert",
group="v2_nfvo")
cfg.CONF.set_override("mtls_client_cert_file", "/path/to/clientcert",
group="v2_nfvo")
cfg.CONF.set_override("token_endpoint", "http://127.0.0.1:9990/token",
group="v2_nfvo")
cfg.CONF.set_override("client_id", "test", group="v2_nfvo")
self.addCleanup(mock.patch.stopall)
mock.patch('os.makedirs').start()
self.nfvo_client_mtls = nfvo_client.NfvoClient()
@mock.patch.object(local_nfvo.LocalNfvo, 'onboarded_show')
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_get_vnf_package_info_vnfd(
self, mock_request, mock_onboarded_show):
# local nfvo
cfg.CONF.clear_override("use_external_nfvo", group="v2_nfvo")
self.nfvo_client.is_local = True
mock_onboarded_show.return_value = vnfd_utils.Vnfd(
vnfd_id=SAMPLE_VNFD_ID)
@ -326,47 +340,77 @@ class TestNfvoClient(base.BaseTestCase):
self.context, SAMPLE_VNFD_ID)
self.assertEqual(SAMPLE_VNFD_ID, result.vnfd_id)
# external nfvo
# external nfvo oauth2
cfg.CONF.clear_override("mtls_client_cert_file", group="v2_nfvo")
cfg.CONF.set_override("use_external_nfvo", True, group="v2_nfvo")
self.nfvo_client.is_local = False
mock_request.return_value = (requests.Response(), _vnfpkg_body_example)
result = self.nfvo_client.get_vnf_package_info_vnfd(
self.context, SAMPLE_VNFD_ID)
self.assertEqual(SAMPLE_VNFD_ID, result.vnfdId)
# external nfvo oauth2 mtls
cfg.CONF.set_override("mtls_client_cert_file", "/path/to/clientcert",
group="v2_nfvo")
result = self.nfvo_client_mtls.get_vnf_package_info_vnfd(
self.context, SAMPLE_VNFD_ID)
self.assertEqual(SAMPLE_VNFD_ID, result.vnfdId)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_onboarded_show_vnfd(self, mock_request):
# local nfvo
cfg.CONF.clear_override("use_external_nfvo", group="v2_nfvo")
self.nfvo_client.is_local = True
result = self.nfvo_client.onboarded_show_vnfd(
self.context, SAMPLE_VNFD_ID)
self.assertIsNone(result)
# external nfvo
# external nfvo oauth2
cfg.CONF.clear_override("mtls_client_cert_file", group="v2_nfvo")
cfg.CONF.set_override("use_external_nfvo", True, group="v2_nfvo")
self.nfvo_client.is_local = False
mock_request.return_value = (requests.Response(), 'test')
result = self.nfvo_client.onboarded_show_vnfd(
self.context, SAMPLE_VNFD_ID)
self.assertEqual('test', result)
# external nfvo oauth2 mtls
cfg.CONF.set_override("mtls_client_cert_file", "/path/to/clientcert",
group="v2_nfvo")
result = self.nfvo_client_mtls.onboarded_show_vnfd(
self.context, SAMPLE_VNFD_ID)
self.assertEqual('test', result)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_onboarded_package_content(self, mock_request):
# local nfvo
cfg.CONF.clear_override("use_external_nfvo", group="v2_nfvo")
self.nfvo_client.is_local = True
result = self.nfvo_client.onboarded_package_content(
self.context, SAMPLE_VNFD_ID)
self.assertIsNone(result)
# external nfvo
# external nfvo oauth2
cfg.CONF.clear_override("mtls_client_cert_file", group="v2_nfvo")
cfg.CONF.set_override("use_external_nfvo", True, group="v2_nfvo")
self.nfvo_client.is_local = False
mock_request.return_value = (requests.Response(), 'test')
result = self.nfvo_client.onboarded_package_content(
self.context, SAMPLE_VNFD_ID)
self.assertEqual('test', result)
# external nfvo oauth2 mtls
cfg.CONF.set_override("mtls_client_cert_file", "/path/to/clientcert",
group="v2_nfvo")
result = self.nfvo_client_mtls.onboarded_package_content(
self.context, SAMPLE_VNFD_ID)
self.assertEqual('test', result)
@mock.patch.object(local_nfvo.LocalNfvo, 'grant')
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_grant(self, mock_request, mock_grant):
# local nfvo
cfg.CONF.clear_override("use_external_nfvo", group="v2_nfvo")
self.nfvo_client.is_local = True
mock_grant.return_value = objects.GrantV1.from_dict(_grant_res)
grant_req = objects.GrantRequestV1.from_dict(_inst_grant_req_example)
@ -377,6 +421,8 @@ class TestNfvoClient(base.BaseTestCase):
'vimAssets']['softwareImages'][0]['vimSoftwareImageId'])
# external nfvo
cfg.CONF.clear_override("mtls_client_cert_file", group="v2_nfvo")
cfg.CONF.set_override("use_external_nfvo", True, group="v2_nfvo")
self.nfvo_client.is_local = False
mock_request.return_value = (requests.Response(), _grant_res)
grant_res = self.nfvo_client.grant(self.context, grant_req)
@ -385,11 +431,21 @@ class TestNfvoClient(base.BaseTestCase):
self.assertEqual('44fd5841-ce98-4d54-a828-6ef51f941c8a', result[
'vimAssets']['softwareImages'][0]['vimSoftwareImageId'])
# external nfvo oauth2 mtls
cfg.CONF.set_override("mtls_client_cert_file", "/path/to/clientcert",
group="v2_nfvo")
grant_res = self.nfvo_client_mtls.grant(self.context, grant_req)
result = grant_res.to_dict()
self.assertIsNotNone(result['addResources'])
self.assertEqual('44fd5841-ce98-4d54-a828-6ef51f941c8a', result[
'vimAssets']['softwareImages'][0]['vimSoftwareImageId'])
@mock.patch.object(objects.base.TackerPersistentObject, 'get_all')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(local_nfvo.LocalNfvo, 'recv_inst_create_notification')
def test_send_inst_create_notification(
self, mock_recv, mock_send, mock_subscs):
cfg.CONF.clear_override("use_external_nfvo", group="v2_nfvo")
self.nfvo_client.is_local = True
inst = objects.VnfInstanceV2(id='test-instance')
mock_subscs.return_value = [objects.LccnSubscriptionV2(id='subsc-1')]
@ -403,6 +459,7 @@ class TestNfvoClient(base.BaseTestCase):
@mock.patch.object(local_nfvo.LocalNfvo, 'recv_inst_delete_notification')
def test_send_inst_delete_notification(
self, mock_recv, mock_send, mock_subscs):
cfg.CONF.clear_override("use_external_nfvo", group="v2_nfvo")
self.nfvo_client.is_local = True
inst = objects.VnfInstanceV2(id='test-instance')
mock_subscs.return_value = [objects.LccnSubscriptionV2(id='subsc-1')]