NFP (contrib) - Over the Cloud Network Function Controller
This changeset implements over the cloud network function controller. The network function controller provides the API server to consume the REST APIs from under the cloud components required to provision the network functions rendered over the cloud. It includes a pecan REST server which receives REST calls from under the cloud and sends corresponding RPC calls to the configurator. Change-Id: I74ee6d9ce7d7a362721b5d3b72ee9b4ed1993995 Implements: blueprint gbp-network-services-framework
This commit is contained in:
Suresh Dharavath
Hemanth Ravi
parent
7dc8e9d5ee
commit
3921f16311
@ -2,6 +2,6 @@
|
||||
test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
|
||||
OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \
|
||||
OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-60} \
|
||||
${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./gbpservice/neutron/tests/unit} $LISTOPT $IDOPTION
|
||||
${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./gbpservice} $LISTOPT $IDOPTION
|
||||
test_id_option=--load-list $IDFILE
|
||||
test_list_option=--list
|
||||
|
||||
0
gbpservice/contrib/__init__.py
Normal file
0
gbpservice/contrib/__init__.py
Normal file
0
gbpservice/contrib/nfp/__init__.py
Normal file
0
gbpservice/contrib/nfp/__init__.py
Normal file
0
gbpservice/contrib/nfp/configurator/__init__.py
Normal file
0
gbpservice/contrib/nfp/configurator/__init__.py
Normal file
@ -0,0 +1,271 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import oslo_serialization.jsonutils as jsonutils
|
||||
|
||||
from neutron.common import rpc as n_rpc
|
||||
from oslo_config import cfg
|
||||
from oslo_log import log as logging
|
||||
import oslo_messaging
|
||||
import pecan
|
||||
|
||||
from gbpservice.nfp.pecan import base_controller
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
n_rpc.init(cfg.CONF)
|
||||
|
||||
|
||||
class Controller(base_controller.BaseController):
|
||||
"""Implements all the APIs Invoked by HTTP requests.
|
||||
|
||||
Implements following HTTP methods.
|
||||
-get
|
||||
-post
|
||||
-put
|
||||
According to the HTTP request received from config-agent this class make
|
||||
call/cast to configurator and return response to config-agent
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, method_name):
|
||||
try:
|
||||
self.method_name = method_name
|
||||
self.services = pecan.conf['cloud_services']
|
||||
self.rpc_routing_table = {}
|
||||
for service in self.services:
|
||||
self._entry_to_rpc_routing_table(service)
|
||||
|
||||
super(Controller, self).__init__()
|
||||
except Exception as err:
|
||||
msg = (
|
||||
"Failed to initialize Controller class %s." %
|
||||
str(err).capitalize())
|
||||
LOG.error(msg)
|
||||
|
||||
def _entry_to_rpc_routing_table(self, service):
|
||||
"""Prepares routing table based on the uservice configuration.
|
||||
This routing table is used to route the rpcs to all interested
|
||||
uservices. Key used for routing is the uservice[apis].
|
||||
|
||||
:param uservice
|
||||
e.g uservice = {'service_name': 'configurator',
|
||||
'topic': 'configurator',
|
||||
'reporting_interval': '10', # in seconds
|
||||
'apis': ['CONFIGURATION', 'EVENT']
|
||||
}
|
||||
Returns: None
|
||||
|
||||
Prepares: self.rpc_routing_table
|
||||
e.g self.rpc_routing_table = {'CONFIGURATION': [rpc_client, ...],
|
||||
'EVENT': [rpc_client, ...]
|
||||
}
|
||||
"""
|
||||
for api in service['apis']:
|
||||
if api not in self.rpc_routing_table:
|
||||
self.rpc_routing_table[api] = []
|
||||
|
||||
self.rpc_routing_table[api].append(CloudService(**service))
|
||||
|
||||
@pecan.expose(method='GET', content_type='application/json')
|
||||
def get(self):
|
||||
"""Method of REST server to handle request get_notifications.
|
||||
|
||||
This method send an RPC call to configurator and returns Notification
|
||||
data to config-agent
|
||||
|
||||
Returns: Dictionary that contains Notification data
|
||||
|
||||
"""
|
||||
|
||||
try:
|
||||
if self.method_name == 'get_notifications':
|
||||
routing_key = 'CONFIGURATION'
|
||||
uservice = self.rpc_routing_table[routing_key]
|
||||
notification_data = uservice[0].rpcclient.call(
|
||||
self.method_name)
|
||||
msg = ("NOTIFICATION_DATA sent to config_agent %s"
|
||||
% notification_data)
|
||||
LOG.info(msg)
|
||||
return jsonutils.dumps(notification_data)
|
||||
|
||||
except Exception as err:
|
||||
pecan.response.status = 400
|
||||
msg = ("Failed to get handle request=%s. Reason=%s."
|
||||
% (self.method_name, str(err).capitalize()))
|
||||
LOG.error(msg)
|
||||
error_data = self._format_description(msg)
|
||||
return jsonutils.dumps(error_data)
|
||||
|
||||
@pecan.expose(method='POST', content_type='application/json')
|
||||
def post(self, **body):
|
||||
"""Method of REST server to handle all the post requests.
|
||||
|
||||
This method sends an RPC cast to configurator according to the
|
||||
HTTP request.
|
||||
|
||||
:param body: This method excepts dictionary as a parameter in HTTP
|
||||
request and send this dictionary to configurator with RPC cast.
|
||||
|
||||
Returns: None
|
||||
|
||||
"""
|
||||
|
||||
try:
|
||||
body = None
|
||||
if pecan.request.is_body_readable:
|
||||
body = pecan.request.json_body
|
||||
if self.method_name == 'network_function_event':
|
||||
routing_key = 'VISIBILITY'
|
||||
else:
|
||||
routing_key = 'CONFIGURATION'
|
||||
for uservice in self.rpc_routing_table[routing_key]:
|
||||
uservice.rpcclient.cast(self.method_name, body)
|
||||
msg = ('Sent RPC to %s' % (uservice.topic))
|
||||
LOG.info(msg)
|
||||
|
||||
msg = ("Successfully served HTTP request %s" % self.method_name)
|
||||
LOG.info(msg)
|
||||
|
||||
except Exception as err:
|
||||
pecan.response.status = 400
|
||||
msg = ("Failed to serve HTTP post request %s %s."
|
||||
% (self.method_name, str(err).capitalize()))
|
||||
# extra_import = ("need to remove this import %s" % config)
|
||||
# LOG.debug(extra_import)
|
||||
LOG.error(msg)
|
||||
error_data = self._format_description(msg)
|
||||
return jsonutils.dumps(error_data)
|
||||
|
||||
@pecan.expose(method='PUT', content_type='application/json')
|
||||
def put(self, **body):
|
||||
"""Method of REST server to handle all the put requests.
|
||||
|
||||
This method sends an RPC cast to configurator according to the
|
||||
HTTP request.
|
||||
|
||||
:param body: This method excepts dictionary as a parameter in HTTP
|
||||
request and send this dictionary to configurator with RPC cast.
|
||||
|
||||
Returns: None
|
||||
|
||||
"""
|
||||
try:
|
||||
body = None
|
||||
if pecan.request.is_body_readable:
|
||||
body = pecan.request.json_body
|
||||
if self.method_name == 'network_function_event':
|
||||
routing_key = 'VISIBILITY'
|
||||
else:
|
||||
routing_key = 'CONFIGURATION'
|
||||
for uservice in self.rpc_routing_table[routing_key]:
|
||||
uservice.rpcclient.cast(self.method_name, body)
|
||||
msg = ('Sent RPC to %s' % (uservice.topic))
|
||||
LOG.info(msg)
|
||||
msg = ("Successfully served HTTP request %s" % self.method_name)
|
||||
LOG.info(msg)
|
||||
|
||||
except Exception as err:
|
||||
pecan.response.status = 400
|
||||
msg = ("Failed to serve HTTP put request %s %s."
|
||||
% (self.method_name, str(err).capitalize()))
|
||||
LOG.error(msg)
|
||||
error_data = self._format_description(msg)
|
||||
return jsonutils.dumps(error_data)
|
||||
|
||||
def _format_description(self, msg):
|
||||
"""This methgod formats error description.
|
||||
|
||||
:param msg: An error message that is to be formatted
|
||||
|
||||
Returns: error_data dictionary
|
||||
"""
|
||||
|
||||
error_data = {'failure_desc': {'msg': msg}}
|
||||
return error_data
|
||||
|
||||
|
||||
class RPCClient(object):
|
||||
"""Implements call/cast methods used in REST Controller.
|
||||
|
||||
Implements following methods.
|
||||
-call
|
||||
-cast
|
||||
This class send an RPC call/cast to configurator according to the data sent
|
||||
by Controller class of REST server.
|
||||
|
||||
"""
|
||||
|
||||
API_VERSION = '1.0'
|
||||
|
||||
def __init__(self, topic):
|
||||
|
||||
self.topic = topic
|
||||
target = oslo_messaging.Target(
|
||||
topic=self.topic,
|
||||
version=self.API_VERSION)
|
||||
self.client = n_rpc.get_client(target)
|
||||
|
||||
def call(self, method_name):
|
||||
"""Method for sending call request on behalf of REST Controller.
|
||||
|
||||
This method sends an RPC call to configurator.
|
||||
|
||||
Returns: Notification data sent by configurator.
|
||||
|
||||
"""
|
||||
cctxt = self.client.prepare(version=self.API_VERSION,
|
||||
topic=self.topic)
|
||||
return cctxt.call(self, method_name)
|
||||
|
||||
def cast(self, method_name, request_data):
|
||||
"""Method for sending cast request on behalf of REST Controller.
|
||||
|
||||
This method sends an RPC cast to configurator according to the
|
||||
method_name passed by COntroller class of REST server.
|
||||
|
||||
:param method_name:method name can be any of the following.
|
||||
|
||||
|
||||
Returns: None.
|
||||
|
||||
"""
|
||||
cctxt = self.client.prepare(version=self.API_VERSION,
|
||||
topic=self.topic)
|
||||
|
||||
return cctxt.cast(self,
|
||||
method_name,
|
||||
request_data=request_data)
|
||||
|
||||
def to_dict(self):
|
||||
"""This function return empty dictionary.
|
||||
|
||||
For making RPC call/cast it internally requires context class that
|
||||
contains to_dict() function. Here we are sending context inside
|
||||
request data so we are passing class itself as a context that
|
||||
contains to_dict() function.
|
||||
|
||||
Returns: Dictionary.
|
||||
|
||||
"""
|
||||
return {}
|
||||
|
||||
|
||||
class CloudService(object):
|
||||
""" CloudService keeps all information of uservice along with initialized
|
||||
RPCClient object using which rpc is routed to over the cloud service.
|
||||
"""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self.service_name = kwargs.get('service_name')
|
||||
self.topic = kwargs.get('topic')
|
||||
self.reporting_interval = kwargs.get('reporting_interval')
|
||||
self.rpcclient = RPCClient(topic=self.topic)
|
||||
@ -0,0 +1,63 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import pecan
|
||||
|
||||
from gbpservice.contrib.nfp.configurator.advanced_controller import (
|
||||
controller)
|
||||
|
||||
"""This class forwards HTTP request to controller class.
|
||||
|
||||
This class create an object of Controller class with appropriate
|
||||
parameter according to the path of HTTP request. According to the
|
||||
parameter passed to Controller class it sends an RPC call/cast to
|
||||
configurator.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
class ControllerResolver(object):
|
||||
|
||||
create_network_function_device_config = controller.Controller(
|
||||
"create_network_function_device_config")
|
||||
delete_network_function_device_config = controller.Controller(
|
||||
"delete_network_function_device_config")
|
||||
update_network_function_device_config = controller.Controller(
|
||||
"update_network_function_device_config")
|
||||
create_network_function_config = controller.Controller(
|
||||
"create_network_function_config")
|
||||
delete_network_function_config = controller.Controller(
|
||||
"delete_network_function_config")
|
||||
update_network_function_config = controller.Controller(
|
||||
"update_network_function_config")
|
||||
get_notifications = controller.Controller("get_notifications")
|
||||
network_function_event = controller.Controller("network_function_event")
|
||||
get_requests = controller.Controller("get_requests")
|
||||
|
||||
""" This class forwards HTTP requests starting with /v1/nfp.
|
||||
|
||||
All HTTP requests with path starting from /v1
|
||||
land here. This class forward request with path starting from /v1/nfp
|
||||
to ControllerResolver.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
class V1Controller(object):
|
||||
|
||||
nfp = ControllerResolver()
|
||||
|
||||
@pecan.expose()
|
||||
def get(self):
|
||||
return {'versions': [{'status': 'CURRENT',
|
||||
'updated': '2014-12-11T00:00:00Z',
|
||||
'id': 'v1'}]}
|
||||
0
gbpservice/contrib/tests/__init__.py
Normal file
0
gbpservice/contrib/tests/__init__.py
Normal file
0
gbpservice/contrib/tests/unit/__init__.py
Normal file
0
gbpservice/contrib/tests/unit/__init__.py
Normal file
0
gbpservice/contrib/tests/unit/nfp/__init__.py
Normal file
0
gbpservice/contrib/tests/unit/nfp/__init__.py
Normal file
@ -0,0 +1,275 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
import mock
|
||||
import os
|
||||
import oslo_serialization.jsonutils as jsonutils
|
||||
import pecan
|
||||
PECAN_CONFIG_FILE = (os.getcwd() +
|
||||
"/gbpservice/nfp/pecan/api/config.py")
|
||||
|
||||
pecan.set_config(PECAN_CONFIG_FILE, overwrite=True)
|
||||
|
||||
import webtest
|
||||
import zlib
|
||||
|
||||
from neutron.tests import base
|
||||
from pecan import rest
|
||||
|
||||
from gbpservice.nfp.pecan import constants
|
||||
|
||||
setattr(pecan, 'mode', constants.advanced)
|
||||
|
||||
from gbpservice.contrib.nfp.configurator.advanced_controller import controller
|
||||
from gbpservice.nfp.pecan.api import root_controller
|
||||
reload(root_controller)
|
||||
|
||||
|
||||
class ControllerTestCase(base.BaseTestCase, rest.RestController):
|
||||
"""
|
||||
This class contains all the unittest cases for REST server of configurator.
|
||||
|
||||
This class tests success and failure cases for all the HTTP requests which
|
||||
are implemented in REST server. run_tests.sh file is used for running all
|
||||
the tests in this class. All the methods of this class started with test
|
||||
prefix called and on success it will print ok and on failure it will
|
||||
print the error trace.
|
||||
|
||||
"""
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""A class method called before tests in an individual class run
|
||||
|
||||
"""
|
||||
rootController = root_controller.RootController()
|
||||
ControllerTestCase.app = webtest.TestApp(
|
||||
pecan.make_app(rootController))
|
||||
ControllerTestCase.data = {'info': {'service_type': 'firewall',
|
||||
'service_vendor': 'vyos',
|
||||
'context': {}},
|
||||
'config': [{'resource': 'firewall',
|
||||
'resource_data': {}}]
|
||||
}
|
||||
|
||||
def test_get_notifications(self):
|
||||
"""Tests HTTP get request get_notifications.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'call') as rpc_mock:
|
||||
rpc_mock.return_value = jsonutils.dumps(self.data)
|
||||
response = self.app.get(
|
||||
'/v1/nfp/get_notifications'
|
||||
)
|
||||
rpc_mock.assert_called_with('get_notifications')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_post_create_network_function_device_config(self):
|
||||
"""Tests HTTP post request create_network_function_device_config.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
response = self.app.post(
|
||||
'/v1/nfp/create_network_function_device_config',
|
||||
zlib.compress(jsonutils.dumps(self.data)),
|
||||
content_type='application/octet-stream')
|
||||
rpc_mock.assert_called_with(
|
||||
'create_network_function_device_config', self.data)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_post_create_network_function_config(self):
|
||||
"""Tests HTTP post request create_network_function_config.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
response = self.app.post(
|
||||
'/v1/nfp/create_network_function_config',
|
||||
zlib.compress(jsonutils.dumps(self.data)),
|
||||
content_type='application/octet-stream')
|
||||
rpc_mock.assert_called_with(
|
||||
'create_network_function_config', self.data)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_post_delete_network_function_device_config(self):
|
||||
"""Tests HTTP post request delete_network_function_device_config.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
response = self.app.post(
|
||||
'/v1/nfp/delete_network_function_device_config',
|
||||
zlib.compress(jsonutils.dumps(self.data)),
|
||||
content_type='application/octet-stream')
|
||||
rpc_mock.assert_called_with(
|
||||
'delete_network_function_device_config', self.data)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_post_delete_network_function_config(self):
|
||||
"""Tests HTTP post request delete_network_function_config.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
response = self.app.post(
|
||||
'/v1/nfp/delete_network_function_config',
|
||||
zlib.compress(jsonutils.dumps(self.data)),
|
||||
content_type='application/octet-stream')
|
||||
rpc_mock.assert_called_with(
|
||||
'delete_network_function_config', self.data)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_put_update_network_function_device_config(self):
|
||||
"""Tests HTTP put request update_network_function_device_config.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
response = self.app.put(
|
||||
'/v1/nfp/update_network_function_device_config',
|
||||
zlib.compress(jsonutils.dumps(self.data)),
|
||||
content_type='application/octet-stream')
|
||||
rpc_mock.assert_called_with(
|
||||
'update_network_function_device_config', self.data)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_put_update_network_function_config(self):
|
||||
"""Tests HTTP put request update_network_function_config.
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
response = self.app.put(
|
||||
'/v1/nfp/update_network_function_config',
|
||||
zlib.compress(jsonutils.dumps(self.data)),
|
||||
content_type='application/octet-stream')
|
||||
rpc_mock.assert_called_with(
|
||||
'update_network_function_config', self.data)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_post_create_network_function_device_config_fail(self):
|
||||
"""Tests failure case of HTTP post request
|
||||
create_network_function_device_config
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
rpc_mock.return_value = Exception
|
||||
response = self.app.post(
|
||||
'/v1/nfp/create_network_function_device_config',
|
||||
expect_errors=True)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_post_create_network_function_config_fail(self):
|
||||
"""Tests failure case of HTTP post request
|
||||
create_network_function_config
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
rpc_mock.return_value = Exception
|
||||
response = self.app.post(
|
||||
'/v1/nfp/create_network_function_config',
|
||||
expect_errors=True)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_post_delete_network_function_device_config_fail(self):
|
||||
"""Tests failure case of HTTP post request
|
||||
delete_network_function_device_config
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
rpc_mock.return_value = Exception
|
||||
response = self.app.post(
|
||||
'/v1/nfp/delete_network_function_device_config',
|
||||
expect_errors=True)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_post_delete_network_function_config_fail(self):
|
||||
"""Tests failure case of HTTP post request
|
||||
delete_network_function_config
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
rpc_mock.return_value = Exception
|
||||
response = self.app.post(
|
||||
'/v1/nfp/delete_network_function_config',
|
||||
expect_errors=True)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_put_update_network_function_device_config_fail(self):
|
||||
"""Tests failure case of HTTP put request
|
||||
update_network_function_device_config
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
rpc_mock.return_value = Exception
|
||||
response = self.app.post(
|
||||
'/v1/nfp/update_network_function_device_config',
|
||||
expect_errors=True)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_put_update_network_function_config_fail(self):
|
||||
"""Tests failure case of HTTP put request
|
||||
update_network_function_config
|
||||
|
||||
Returns: none
|
||||
|
||||
"""
|
||||
|
||||
with mock.patch.object(
|
||||
controller.RPCClient, 'cast') as rpc_mock:
|
||||
rpc_mock.return_value = Exception
|
||||
response = self.app.post(
|
||||
'/v1/nfp/update_network_function_config',
|
||||
expect_errors=True)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
Loading…
Reference in New Issue
Block a user