openstack/tacker
Code Issues Proposed changes

Merge "Reduce duplication under SOL v2 API framework"

Browse Source
This commit is contained in:
Zuul
2024-03-17 17:34:34 +00:00
committed by Gerrit Code Review
parent eff547cf3d ef30b1d7e8
commit a0eb01ee44
28 changed files with 148 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
etc/tacker/api-paste.ini
View File

@@ -90,19 +90,19 @@ paste.app_factory = tacker.sol_refactored.api.router:VnflcmVersions.factory
paste.app_factory = tacker.sol_refactored.api.router:VnffmAPIRouterV1.factory paste.app_factory = tacker.sol_refactored.api.router:VnffmAPIRouterV1.factory
[app:prometheus_auto_scaling] [app:prometheus_auto_scaling]
paste.app_factory = tacker.sol_refactored.api.prometheus_plugin_router:AutoScalingRouter.factory paste.app_factory = tacker.sol_refactored.api.router:AutoScalingRouter.factory
[app:prometheus_auto_healing] [app:prometheus_auto_healing]
paste.app_factory = tacker.sol_refactored.api.prometheus_plugin_router:AutoHealingRouter.factory paste.app_factory = tacker.sol_refactored.api.router:AutoHealingRouter.factory
[app:prometheus_fm] [app:prometheus_fm]
paste.app_factory = tacker.sol_refactored.api.prometheus_plugin_router:FmAlertRouter.factory paste.app_factory = tacker.sol_refactored.api.router:FmAlertRouter.factory
[app:prometheus_pm] [app:prometheus_pm]
paste.app_factory = tacker.sol_refactored.api.prometheus_plugin_router:PmEventRouter.factory paste.app_factory = tacker.sol_refactored.api.router:PmEventRouter.factory
[app:prometheus_threshold] [app:prometheus_threshold]
paste.app_factory = tacker.sol_refactored.api.prometheus_plugin_router:PmThresholdRouter.factory paste.app_factory = tacker.sol_refactored.api.router:PmThresholdRouter.factory
[app:server_notification] [app:server_notification]
paste.app_factory = tacker.sol_refactored.api.server_notification_router:ServerNotificationRouter.factory paste.app_factory = tacker.sol_refactored.api.router:ServerNotificationRouter.factory

54
tacker/sol_refactored/api/prometheus_plugin_router.py
View File

@@ -1,54 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
from tacker.sol_refactored.api.policies import vnffm_v1 as vnffm_policy_v1
from tacker.sol_refactored.api.policies import vnfpm_v2 as vnfpm_policy_v2
from tacker.sol_refactored.api import prometheus_plugin_wsgi as prom_wsgi
from tacker.sol_refactored.controller import prometheus_plugin_controller
class PmEventRouter(prom_wsgi.PrometheusPluginAPIRouter):
controller = prom_wsgi.PrometheusPluginResource(
prometheus_plugin_controller.PmEventController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "pm_event"})]
class PmThresholdRouter(prom_wsgi.PrometheusPluginAPIRouter):
controller = prom_wsgi.PrometheusPluginResource(
prometheus_plugin_controller.PmThresholdController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "pm_threshold"})]
class FmAlertRouter(prom_wsgi.PrometheusPluginAPIRouter):
controller = prom_wsgi.PrometheusPluginResource(
prometheus_plugin_controller.FmAlertController(),
policy_name=vnffm_policy_v1.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "alert"})]
class AutoHealingRouter(prom_wsgi.PrometheusPluginAPIRouter):
controller = prom_wsgi.PrometheusPluginResource(
prometheus_plugin_controller.AutoHealingController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "auto_healing"})]
class AutoScalingRouter(prom_wsgi.PrometheusPluginAPIRouter):
controller = prom_wsgi.PrometheusPluginResource(
prometheus_plugin_controller.AutoScalingController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "auto_scaling"})]

46
tacker/sol_refactored/api/prometheus_plugin_validator.py
View File

@@ -1,46 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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 functools
from tacker.api.validation import validators
from tacker.common import exceptions as tacker_ex
from tacker.sol_refactored.common import exceptions as sol_ex
class PrometheusPluginSchemaValidator(validators._SchemaValidator):
def validate(self, *args, **kwargs):
try:
super(PrometheusPluginSchemaValidator, self).validate(
*args, **kwargs)
except tacker_ex.ValidationError as ex:
raise sol_ex.PrometheusPluginValidationError(detail=str(ex))
def schema(request_body_schema):
def add_validator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if 'body' not in kwargs:
raise sol_ex.PrometheusPluginValidationError(
detail="body is missing.")
schema_validator = PrometheusPluginSchemaValidator(
request_body_schema)
schema_validator.validate(kwargs['body'])
return func(*args, **kwargs)
return wrapper
return add_validator

68
tacker/sol_refactored/api/prometheus_plugin_wsgi.py
View File

@@ -1,68 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
from oslo_log import log as logging
from tacker.sol_refactored.api import wsgi as sol_wsgi
import webob
LOG = logging.getLogger(__name__)
class PrometheusPluginResponse(sol_wsgi.SolResponse):
allowed_headers = ['content_type']
def __init__(self, status, body, **kwargs):
self.status = status
self.body = body
self.headers = {}
for hdr in self.allowed_headers:
if hdr in kwargs:
self.headers[hdr] = kwargs[hdr]
class PrometheusPluginErrorResponse(sol_wsgi.SolErrorResponse):
pass
class PrometheusPluginResource(sol_wsgi.SolResource):
@webob.dec.wsgify(RequestClass=sol_wsgi.SolRequest)
def __call__(self, request):
LOG.info("%(method)s %(url)s", {"method": request.method,
"url": request.url})
try:
action, args, accept = self._deserialize_request(request)
self._check_policy(request, action)
result = self._dispatch(request, action, args)
response = result.serialize(accept)
except Exception as ex:
result = PrometheusPluginErrorResponse(ex, request)
try:
response = result.serialize('application/problem+json')
except Exception:
LOG.exception("Unknown error")
return webob.exc.HTTPBadRequest(explanation="Unknown error")
LOG.info("%(url)s returned with HTTP %(status)d",
{"url": request.url, "status": response.status_int})
return response
class PrometheusPluginAPIRouter(sol_wsgi.SolAPIRouter):
pass
class PrometheusPluginAPIController(sol_wsgi.SolAPIController):
pass

51
tacker/sol_refactored/api/router.py
View File

@@ -18,6 +18,8 @@ from tacker.sol_refactored.api.policies import vnffm_v1 as vnffm_policy_v1
from tacker.sol_refactored.api.policies import vnflcm_v2 as vnflcm_policy_v2 from tacker.sol_refactored.api.policies import vnflcm_v2 as vnflcm_policy_v2
from tacker.sol_refactored.api.policies import vnfpm_v2 as vnfpm_policy_v2 from tacker.sol_refactored.api.policies import vnfpm_v2 as vnfpm_policy_v2
from tacker.sol_refactored.api import wsgi as sol_wsgi from tacker.sol_refactored.api import wsgi as sol_wsgi
from tacker.sol_refactored.controller import prometheus_plugin_controller
from tacker.sol_refactored.controller import server_notification
from tacker.sol_refactored.controller import vnffm_v1 from tacker.sol_refactored.controller import vnffm_v1
from tacker.sol_refactored.controller import vnflcm_v2 from tacker.sol_refactored.controller import vnflcm_v2
from tacker.sol_refactored.controller import vnflcm_versions from tacker.sol_refactored.controller import vnflcm_versions
@@ -90,3 +92,52 @@ class VnfPmAPIRouterV2(sol_wsgi.SolAPIRouter):
"GET": "show_threshold", "GET": "show_threshold",
"DELETE": "delete_threshold"}), "DELETE": "delete_threshold"}),
] ]
# The definitions after here are of tacker original APIs.
# Although these APIs are not included in ESTI SOL specification,
# these APIs are (should be) designed as same as SOL APIs and
# use same API frameworks (i.e. modules in this directory).
class PmEventRouter(sol_wsgi.SolAPIRouter):
controller = sol_wsgi.SolResource(
prometheus_plugin_controller.PmEventController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "pm_event"})]
class PmThresholdRouter(sol_wsgi.SolAPIRouter):
controller = sol_wsgi.SolResource(
prometheus_plugin_controller.PmThresholdController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "pm_threshold"})]
class FmAlertRouter(sol_wsgi.SolAPIRouter):
controller = sol_wsgi.SolResource(
prometheus_plugin_controller.FmAlertController(),
policy_name=vnffm_policy_v1.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "alert"})]
class AutoHealingRouter(sol_wsgi.SolAPIRouter):
controller = sol_wsgi.SolResource(
prometheus_plugin_controller.AutoHealingController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "auto_healing"})]
class AutoScalingRouter(sol_wsgi.SolAPIRouter):
controller = sol_wsgi.SolResource(
prometheus_plugin_controller.AutoScalingController(),
policy_name=vnfpm_policy_v2.POLICY_NAME_PROM_PLUGIN)
route_list = [("", {"POST": "auto_scaling"})]
class ServerNotificationRouter(sol_wsgi.SolAPIRouter):
controller = sol_wsgi.SolResource(
server_notification.ServerNotificationController(),
policy_name=vnflcm_policy_v2.SERVER_NOTIFICATION_POLICY_NAME)
route_list = [
("/vnf_instances/{vnf_instance_id}/servers/{server_id}/notify",
{"POST": "notify"})
]

28
tacker/sol_refactored/api/server_notification_router.py
View File

@@ -1,28 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
from tacker.sol_refactored.api.policies import vnflcm_v2
from tacker.sol_refactored.api import server_notification_wsgi as sn_wsgi
from tacker.sol_refactored.controller import server_notification
class ServerNotificationRouter(sn_wsgi.ServerNotificationAPIRouter):
controller = sn_wsgi.ServerNotificationResource(
server_notification.ServerNotificationController(),
policy_name=vnflcm_v2.SERVER_NOTIFICATION_POLICY_NAME)
route_list = [
("/vnf_instances/{vnf_instance_id}/servers/{server_id}/notify",
{"POST": "notify"})
]

51
tacker/sol_refactored/api/server_notification_validator.py
View File

@@ -1,51 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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 functools
from tacker.api.validation import validators
from tacker.common import exceptions as tacker_ex
from tacker.sol_refactored.common import exceptions as sol_ex
# TODO(shimizu-koji): `validators._SchemaValidator` is protected class,
# thus it shouldn't be referred from other modules. This refactoring
# will be done in other patches in the future.
class ServerNotificationSchemaValidator(validators._SchemaValidator):
def validate(self, *args, **kwargs):
try:
super(ServerNotificationSchemaValidator, self).validate(
*args, **kwargs)
except tacker_ex.ValidationError as ex:
raise sol_ex.ServerNotificationValidationError(detail=str(ex))
def schema(request_body_schema):
def add_validator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if 'body' not in kwargs:
raise sol_ex.ServerNotificationValidationError(
detail="body is missing.")
schema_validator = ServerNotificationSchemaValidator(
request_body_schema)
schema_validator.validate(kwargs['body'])
return func(*args, **kwargs)
return wrapper
return add_validator

82
tacker/sol_refactored/api/server_notification_wsgi.py
View File

@@ -1,82 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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 webob
from oslo_log import log as logging
from tacker.sol_refactored.api import wsgi as sol_wsgi
LOG = logging.getLogger(__name__)
class ServerNotificationResponse(sol_wsgi.SolResponse):
allowed_headers = ['content_type']
def __init__(self, status, body, **kwargs):
self.status = status
self.body = body
self.headers = {}
for hdr in self.allowed_headers:
if hdr in kwargs:
self.headers[hdr] = kwargs[hdr]
class ServerNotificationErrorResponse(ServerNotificationResponse):
def __init__(self, ex, _):
status = ex.status if hasattr(ex, 'status') else 'error'
detail = ex.status if hasattr(ex, 'detail') else 'error'
problem_details = {'status': status, 'detail': detail}
if hasattr(ex, 'title'):
problem_details['title'] = ex.title
super(ServerNotificationErrorResponse, self).__init__(
problem_details['status'], problem_details)
class ServerNotificationResource(sol_wsgi.SolResource):
def __init__(self, controller, policy_name=None):
super(ServerNotificationResource, self).__init__(
controller, policy_name=policy_name
)
@webob.dec.wsgify(RequestClass=sol_wsgi.SolRequest)
def __call__(self, request):
LOG.info("%(method)s %(url)s", {"method": request.method,
"url": request.url})
try:
action, args, accept = self._deserialize_request(request)
self._check_policy(request, action)
result = self._dispatch(request, action, args)
response = result.serialize(accept)
except Exception as ex:
result = ServerNotificationErrorResponse(ex, request)
try:
response = result.serialize('application/problem+json')
except Exception:
LOG.exception("Unknown error")
return webob.exc.HTTPBadRequest(explanation="Unknown error")
LOG.info("%(url)s returned with HTTP %(status)d",
{"url": request.url, "status": response.status_int})
return response
class ServerNotificationAPIRouter(sol_wsgi.SolAPIRouter):
pass
class ServerNotificationAPIController(sol_wsgi.SolAPIController):
pass

16
tacker/sol_refactored/api/validator.py
View File

@@ -49,3 +49,19 @@ def schema(request_body_schema, min_version, max_version=None):
return wrapper return wrapper
return add_validator return add_validator
def schema_nover(request_body_schema):
def add_validator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if 'body' not in kwargs:
raise sol_ex.SolValidationError(detail="body is missing")
schema_validator = SolSchemaValidator(request_body_schema)
schema_validator.validate(kwargs['body'])
return func(*args, **kwargs)
return wrapper
return add_validator

21
tacker/sol_refactored/api/wsgi.py
View File

@@ -69,8 +69,6 @@ class SolResponse(object):
for hdr in self.allowed_headers: for hdr in self.allowed_headers:
if kwargs.get(hdr): if kwargs.get(hdr):
self.headers[hdr] = kwargs[hdr] self.headers[hdr] = kwargs[hdr]
self.headers.setdefault('version', api_version.CURRENT_VERSION)
self.headers.setdefault('accept-ranges', 'none')
def serialize(self, content_type): def serialize(self, content_type):
self.headers.setdefault('content_type', content_type) self.headers.setdefault('content_type', content_type)
@@ -132,13 +130,11 @@ class SolResource(object):
self._check_api_version(request, action) self._check_api_version(request, action)
self._check_policy(request, action) self._check_policy(request, action)
result = self._dispatch(request, action, args) result = self._dispatch(request, action, args)
self.controller.set_default_to_response(result, action)
response = result.serialize(accept) response = result.serialize(accept)
except Exception as ex: except Exception as ex:
result = SolErrorResponse(ex, request.best_match_language()) result = SolErrorResponse(ex, request.best_match_language())
if type(self.controller).__name__ == 'VnfFmControllerV1': self.controller.set_default_to_response(result, action)
result.headers['version'] = api_version.CURRENT_FM_VERSION
if type(self.controller).__name__ == 'VnfPmControllerV2':
result.headers['version'] = api_version.CURRENT_PM_VERSION
try: try:
response = result.serialize('application/problem+json') response = result.serialize('application/problem+json')
except Exception: except Exception:
@@ -202,7 +198,7 @@ class SolResource(object):
return request.body_file return request.body_file
else: else:
# assume json format # assume json format
# ex. 'application/json', 'application/mergepatch+json' # ex. 'application/json', 'application/merge-patch+json'
try: try:
return request.json return request.json
except Exception: except Exception:
@@ -275,10 +271,10 @@ class SolAPIController(object):
raise sol_ex.MethodNotAllowed(method=request.method) raise sol_ex.MethodNotAllowed(method=request.method)
def supported_api_versions(self, action): def supported_api_versions(self, action):
# NOTE: support v2 API by default. if a contorller supports # NOTE: if a contorller supports versions header, override
# and/or v1 API, or depending on action, override this method # this method in the subclass. return None means version is
# in the subclass. # not checked.
return api_version.v2_versions return None
def allowed_content_types(self, action): def allowed_content_types(self, action):
# NOTE: if other than 'application/json' is expected depending # NOTE: if other than 'application/json' is expected depending
@@ -291,3 +287,6 @@ class SolAPIController(object):
# NOTE: if other than 'application/json' is expected depending # NOTE: if other than 'application/json' is expected depending
# on action, override this method in the subclass. # on action, override this method in the subclass.
return ['application/json'] return ['application/json']
def set_default_to_response(self, result, action):
pass

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

@@ -448,10 +448,6 @@ class PrometheusPluginSkipped(Exception):
pass pass
class PrometheusPluginValidationError(SolValidationError):
pass
class PrometheusSettingFailed(SolHttpError503): class PrometheusSettingFailed(SolHttpError503):
message = _("Setting PM job on External Monitoring Tool failed.") message = _("Setting PM job on External Monitoring Tool failed.")
@@ -459,7 +455,3 @@ class PrometheusSettingFailed(SolHttpError503):
# server_notification # server_notification
class ServerNotificationNotEnabled(SolHttpError404): class ServerNotificationNotEnabled(SolHttpError404):
message = _("ServerNotification API is not enabled.") message = _("ServerNotification API is not enabled.")
class ServerNotificationValidationError(SolValidationError):
message = _("%(detail)s")

12
tacker/sol_refactored/common/prometheus_plugin.py
View File

@@ -26,8 +26,8 @@ from keystoneauth1 import exceptions as ks_exc
from oslo_log import log as logging from oslo_log import log as logging
from oslo_utils import uuidutils from oslo_utils import uuidutils
from tacker.common import utils from tacker.common import utils
from tacker.sol_refactored.api import prometheus_plugin_validator as validator
from tacker.sol_refactored.api.schemas import prometheus_plugin_schemas from tacker.sol_refactored.api.schemas import prometheus_plugin_schemas
from tacker.sol_refactored.api import validator
from tacker.sol_refactored.common import config as cfg from tacker.sol_refactored.common import config as cfg
from tacker.sol_refactored.common import exceptions as sol_ex 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_alarm_utils
@@ -505,7 +505,7 @@ class PrometheusPluginPm(PrometheusPluginPmBase, mon_base.MonitoringPlugin):
f"doesn't match pmJob.") f"doesn't match pmJob.")
raise sol_ex.PrometheusPluginSkipped() raise sol_ex.PrometheusPluginSkipped()
@validator.schema(prometheus_plugin_schemas.AlertMessage) @validator.schema_nover(prometheus_plugin_schemas.AlertMessage)
def _alert(self, request, body): def _alert(self, request, body):
result = [] result = []
context = request.context context = request.context
@@ -762,7 +762,7 @@ class PrometheusPluginThreshold(PrometheusPluginPmBase,
crossing_direction = "DOWN" crossing_direction = "DOWN"
return crossing_direction return crossing_direction
@validator.schema(prometheus_plugin_schemas.AlertMessage) @validator.schema_nover(prometheus_plugin_schemas.AlertMessage)
def _alert(self, request, body): def _alert(self, request, body):
result = [] result = []
context = request.context context = request.context
@@ -1100,7 +1100,7 @@ class PrometheusPluginFm(PrometheusPlugin, mon_base.MonitoringPlugin):
return [new_alarm] return [new_alarm]
raise sol_ex.PrometheusPluginSkipped() raise sol_ex.PrometheusPluginSkipped()
@validator.schema(prometheus_plugin_schemas.AlertMessage) @validator.schema_nover(prometheus_plugin_schemas.AlertMessage)
def _alert(self, request, body): def _alert(self, request, body):
now = datetime.datetime.now(datetime.timezone.utc) now = datetime.datetime.now(datetime.timezone.utc)
result = [] result = []
@@ -1153,7 +1153,7 @@ class PrometheusPluginAutoHealing(PrometheusPlugin, mon_base.MonitoringPlugin):
self.rpc.enqueue_auto_heal_instance( self.rpc.enqueue_auto_heal_instance(
context, vnf_instance_id, vnfc_info_id) context, vnf_instance_id, vnfc_info_id)
@validator.schema(prometheus_plugin_schemas.AlertMessage) @validator.schema_nover(prometheus_plugin_schemas.AlertMessage)
def _alert(self, request, body): def _alert(self, request, body):
context = request.context context = request.context
alerts = (alert for alert in body['alerts'] if alerts = (alert for alert in body['alerts'] if
@@ -1225,7 +1225,7 @@ class PrometheusPluginAutoScaling(PrometheusPlugin, mon_base.MonitoringPlugin):
def default_callback(self, context, vnf_instance_id, scaling_param): def default_callback(self, context, vnf_instance_id, scaling_param):
self.rpc.trigger_scale(context, vnf_instance_id, scaling_param) self.rpc.trigger_scale(context, vnf_instance_id, scaling_param)
@validator.schema(prometheus_plugin_schemas.AlertMessage) @validator.schema_nover(prometheus_plugin_schemas.AlertMessage)
def _alert(self, request, body): def _alert(self, request, body):
context = request.context context = request.context
alerts = (alert for alert in body['alerts'] if alerts = (alert for alert in body['alerts'] if

13
tacker/sol_refactored/common/server_notification.py
View File

@@ -15,8 +15,7 @@
from oslo_log import log as logging from oslo_log import log as logging
from tacker.sol_refactored.api.schemas import server_notification_schemas from tacker.sol_refactored.api.schemas import server_notification_schemas
from tacker.sol_refactored.api import server_notification_validator\ from tacker.sol_refactored.api import validator
as validator
from tacker.sol_refactored.common import config as cfg from tacker.sol_refactored.common import config as cfg
from tacker.sol_refactored.common import exceptions as sol_ex from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import monitoring_plugin_base as mon_base from tacker.sol_refactored.common import monitoring_plugin_base as mon_base
@@ -69,11 +68,11 @@ class ServerNotification(mon_base.MonitoringPlugin):
'vnfcResourceInfo') or 'vnfcResourceInfo') or
not vnf_instance.instantiatedVnfInfo.obj_attr_is_set( not vnf_instance.instantiatedVnfInfo.obj_attr_is_set(
'vnfcInfo')): 'vnfcInfo')):
raise sol_ex.ServerNotificationValidationError( raise sol_ex.SolValidationError(
detail="access info not found in the vnf instance.") detail="access info not found in the vnf instance.")
if fault_id not in vnf_instance.instantiatedVnfInfo.metadata.get( if fault_id not in vnf_instance.instantiatedVnfInfo.metadata.get(
'ServerNotifierFaultID', []): 'ServerNotifierFaultID', []):
raise sol_ex.ServerNotificationValidationError( raise sol_ex.SolValidationError(
detail="fault_id does not match.") detail="fault_id does not match.")
# Get the list of instantiatedVnfInfo.vnfcInfo[x].id where # Get the list of instantiatedVnfInfo.vnfcInfo[x].id where
@@ -91,16 +90,16 @@ class ServerNotification(mon_base.MonitoringPlugin):
vnfc_ids = list(map(lambda x: x.id, vnfc_info)) vnfc_ids = list(map(lambda x: x.id, vnfc_info))
if len(vnfc_ids) == 0: if len(vnfc_ids) == 0:
raise sol_ex.ServerNotificationValidationError( raise sol_ex.SolValidationError(
detail="target vnfc not found.") detail="target vnfc not found.")
return vnfc_ids return vnfc_ids
@validator.schema(server_notification_schemas.ServerNotification) @validator.schema_nover(server_notification_schemas.ServerNotification)
def notify(self, request, vnf_instance_id, body): def notify(self, request, vnf_instance_id, body):
context = request.context context = request.context
vnf_instance = inst_utils.get_inst(context, vnf_instance_id) vnf_instance = inst_utils.get_inst(context, vnf_instance_id)
if not vnf_instance: if not vnf_instance:
raise sol_ex.ServerNotificationValidationError( raise sol_ex.SolValidationError(
detail="target vnf instance not found.") detail="target vnf instance not found.")
if (not vnf_instance.obj_attr_is_set( if (not vnf_instance.obj_attr_is_set(
'vnfConfigurableProperties') or 'vnfConfigurableProperties') or

22
tacker/sol_refactored/controller/prometheus_plugin_controller.py
View File

@@ -13,7 +13,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from tacker.sol_refactored.api import prometheus_plugin_wsgi as prom_wsgi from tacker.sol_refactored.api import wsgi as sol_wsgi
from tacker.sol_refactored.common import config as cfg from tacker.sol_refactored.common import config as cfg
from tacker.sol_refactored.common import exceptions as sol_ex from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import monitoring_plugin_base as mon_base from tacker.sol_refactored.common import monitoring_plugin_base as mon_base
@@ -21,7 +21,7 @@ from tacker.sol_refactored.common import monitoring_plugin_base as mon_base
CONF = cfg.CONF CONF = cfg.CONF
class PmEventController(prom_wsgi.PrometheusPluginAPIController): class PmEventController(sol_wsgi.SolAPIController):
def pm_event(self, request, body): def pm_event(self, request, body):
if not CONF.prometheus_plugin.performance_management: if not CONF.prometheus_plugin.performance_management:
raise sol_ex.PrometheusPluginNotEnabled( raise sol_ex.PrometheusPluginNotEnabled(
@@ -31,10 +31,10 @@ class PmEventController(prom_wsgi.PrometheusPluginAPIController):
CONF.prometheus_plugin.performance_management_class) CONF.prometheus_plugin.performance_management_class)
mon_base.MonitoringPlugin.get_instance(cls).alert( mon_base.MonitoringPlugin.get_instance(cls).alert(
request=request, body=body) request=request, body=body)
return prom_wsgi.PrometheusPluginResponse(204, None) return sol_wsgi.SolResponse(204, None)
class PmThresholdController(prom_wsgi.PrometheusPluginAPIController): class PmThresholdController(sol_wsgi.SolAPIController):
def pm_threshold(self, request, body): def pm_threshold(self, request, body):
if not CONF.prometheus_plugin.performance_management: if not CONF.prometheus_plugin.performance_management:
raise sol_ex.PrometheusPluginNotEnabled( raise sol_ex.PrometheusPluginNotEnabled(
@@ -44,10 +44,10 @@ class PmThresholdController(prom_wsgi.PrometheusPluginAPIController):
CONF.prometheus_plugin.performance_management_threshold_class) CONF.prometheus_plugin.performance_management_threshold_class)
mon_base.MonitoringPlugin.get_instance(cls).alert( mon_base.MonitoringPlugin.get_instance(cls).alert(
request=request, body=body) request=request, body=body)
return prom_wsgi.PrometheusPluginResponse(204, None) return sol_wsgi.SolResponse(204, None)
class FmAlertController(prom_wsgi.PrometheusPluginAPIController): class FmAlertController(sol_wsgi.SolAPIController):
def alert(self, request, body): def alert(self, request, body):
if not CONF.prometheus_plugin.fault_management: if not CONF.prometheus_plugin.fault_management:
raise sol_ex.PrometheusPluginNotEnabled( raise sol_ex.PrometheusPluginNotEnabled(
@@ -57,10 +57,10 @@ class FmAlertController(prom_wsgi.PrometheusPluginAPIController):
CONF.prometheus_plugin.fault_management_class) CONF.prometheus_plugin.fault_management_class)
mon_base.MonitoringPlugin.get_instance(cls).alert( mon_base.MonitoringPlugin.get_instance(cls).alert(
request=request, body=body) request=request, body=body)
return prom_wsgi.PrometheusPluginResponse(204, None) return sol_wsgi.SolResponse(204, None)
class AutoHealingController(prom_wsgi.PrometheusPluginAPIController): class AutoHealingController(sol_wsgi.SolAPIController):
def auto_healing(self, request, body): def auto_healing(self, request, body):
if not CONF.prometheus_plugin.auto_healing: if not CONF.prometheus_plugin.auto_healing:
raise sol_ex.PrometheusPluginNotEnabled( raise sol_ex.PrometheusPluginNotEnabled(
@@ -70,10 +70,10 @@ class AutoHealingController(prom_wsgi.PrometheusPluginAPIController):
CONF.prometheus_plugin.auto_healing_class) CONF.prometheus_plugin.auto_healing_class)
mon_base.MonitoringPlugin.get_instance(cls).alert( mon_base.MonitoringPlugin.get_instance(cls).alert(
request=request, body=body) request=request, body=body)
return prom_wsgi.PrometheusPluginResponse(204, None) return sol_wsgi.SolResponse(204, None)
class AutoScalingController(prom_wsgi.PrometheusPluginAPIController): class AutoScalingController(sol_wsgi.SolAPIController):
def auto_scaling(self, request, body): def auto_scaling(self, request, body):
if not CONF.prometheus_plugin.auto_scaling: if not CONF.prometheus_plugin.auto_scaling:
raise sol_ex.PrometheusPluginNotEnabled( raise sol_ex.PrometheusPluginNotEnabled(
@@ -83,4 +83,4 @@ class AutoScalingController(prom_wsgi.PrometheusPluginAPIController):
CONF.prometheus_plugin.auto_scaling_class) CONF.prometheus_plugin.auto_scaling_class)
mon_base.MonitoringPlugin.get_instance(cls).alert( mon_base.MonitoringPlugin.get_instance(cls).alert(
request=request, body=body) request=request, body=body)
return prom_wsgi.PrometheusPluginResponse(204, None) return sol_wsgi.SolResponse(204, None)

6
tacker/sol_refactored/controller/server_notification.py
View File

@@ -13,7 +13,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from tacker.sol_refactored.api import server_notification_wsgi as sn_wsgi from tacker.sol_refactored.api import wsgi as sol_wsgi
from tacker.sol_refactored.common import config as cfg from tacker.sol_refactored.common import config as cfg
from tacker.sol_refactored.common import exceptions as sol_ex from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import monitoring_plugin_base as mon_base from tacker.sol_refactored.common import monitoring_plugin_base as mon_base
@@ -21,7 +21,7 @@ from tacker.sol_refactored.common import monitoring_plugin_base as mon_base
CONF = cfg.CONF CONF = cfg.CONF
class ServerNotificationController(sn_wsgi.ServerNotificationAPIController): class ServerNotificationController(sol_wsgi.SolAPIController):
def notify(self, request, vnf_instance_id, server_id, body): def notify(self, request, vnf_instance_id, server_id, body):
if not CONF.server_notification.server_notification: if not CONF.server_notification.server_notification:
raise sol_ex.ServerNotificationNotEnabled() raise sol_ex.ServerNotificationNotEnabled()
@@ -31,4 +31,4 @@ class ServerNotificationController(sn_wsgi.ServerNotificationAPIController):
mon_base.MonitoringPlugin.get_instance(cls).alert( mon_base.MonitoringPlugin.get_instance(cls).alert(
request=request, vnf_instance_id=vnf_instance_id, request=request, vnf_instance_id=vnf_instance_id,
server_id=server_id, body=body) server_id=server_id, body=body)
return sn_wsgi.ServerNotificationResponse(204, None) return sol_wsgi.SolResponse(204, None)

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

@@ -53,10 +53,10 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
def allowed_content_types(self, action): def allowed_content_types(self, action):
if action == 'update': if action == 'update':
# Content-Type of Modify request shall be # Content-Type of Modify request shall be
# 'application/mergepatch+json' according to SOL spec. # 'application/merge-patch+json' according to SOL spec.
# But 'application/json' and 'text/plain' is OK for backward # But 'application/json' and 'text/plain' is OK for backward
# compatibility. # compatibility.
return ['application/mergepatch+json', 'application/json', return ['application/merge-patch+json', 'application/json',
'text/plain'] 'text/plain']
return ['application/json', 'text/plain'] return ['application/json', 'text/plain']
@@ -176,3 +176,7 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
return sol_wsgi.SolResponse(204, None, return sol_wsgi.SolResponse(204, None,
version=api_version.CURRENT_FM_VERSION) version=api_version.CURRENT_FM_VERSION)
def set_default_to_response(self, result, action):
result.headers.setdefault('version', api_version.CURRENT_FM_VERSION)
result.headers.setdefault('accept-ranges', 'none')

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

@@ -650,10 +650,14 @@ class VnfLcmControllerV2(sol_wsgi.SolAPIController):
def allowed_content_types(self, action): def allowed_content_types(self, action):
if action == 'update': if action == 'update':
# Content-Type of Modify request shall be # Content-Type of Modify request shall be
# 'application/mergepatch+json' according to SOL spec. # 'application/merge-patch+json' according to SOL spec.
# But 'application/json' and 'text/plain' is OK for backward # But 'application/json' and 'text/plain' is OK for backward
# compatibility. # compatibility.
return ['application/mergepatch+json', 'application/json', return ['application/merge-patch+json', 'application/json',
'text/plain'] 'text/plain']
else: else:
return ['application/json', 'text/plain'] return ['application/json', 'text/plain']
def set_default_to_response(self, result, action):
result.headers.setdefault('version', api_version.CURRENT_VERSION)
result.headers.setdefault('accept-ranges', 'none')

4
tacker/sol_refactored/controller/vnflcm_versions.py
View File

@@ -29,3 +29,7 @@ class VnfLcmVersionsController(sol_wsgi.SolAPIController):
def supported_api_versions(self, action): def supported_api_versions(self, action):
# support all versions and it is OK there is no Version header. # support all versions and it is OK there is no Version header.
return None return None
def set_default_to_response(self, result, action):
result.headers.setdefault('version', api_version.CURRENT_VERSION)
result.headers.setdefault('accept-ranges', 'none')

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

@@ -278,15 +278,15 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
def allowed_content_types(self, action): def allowed_content_types(self, action):
if action in {'update', 'update_threshold'}: if action in {'update', 'update_threshold'}:
# Content-Type of Modify request shall be # Content-Type of Modify request shall be
# 'application/mergepatch+json' according to SOL spec. # 'application/merge-patch+json' according to SOL spec.
# But 'application/json' and 'text/plain' is OK for backward # But 'application/json' and 'text/plain' is OK for backward
# compatibility. # compatibility.
return ['application/mergepatch+json', 'application/json', return ['application/merge-patch+json', 'application/json',
'text/plain'] 'text/plain']
return ['application/json', 'text/plain'] return ['application/json', 'text/plain']
def allowed_accept(self, action): def allowed_accept(self, action):
return ['application/json', 'application/mergepatch+json', return ['application/json', 'application/merge-patch+json',
'text/plain'] 'text/plain']
def supported_api_versions(self, action): def supported_api_versions(self, action):
@@ -441,3 +441,7 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
return sol_wsgi.SolResponse(204, None, return sol_wsgi.SolResponse(204, None,
version=api_version.CURRENT_PM_VERSION) version=api_version.CURRENT_PM_VERSION)
def set_default_to_response(self, result, action):
result.headers.setdefault('version', api_version.CURRENT_PM_VERSION)
result.headers.setdefault('accept-ranges', 'none')

2
tacker/tests/functional/sol_kubernetes_v2/base_v2.py
View File

@@ -155,7 +155,7 @@ class BaseVnfLcmKubernetesV2Test(base_v2.BaseTackerTestV2):
path = f"/vnfpm/v2/thresholds/{pm_threshold_id}" path = f"/vnfpm/v2/thresholds/{pm_threshold_id}"
return self.tacker_client.do_request( return self.tacker_client.do_request(
path, "PATCH", body=req_body, version=VNFPM_V2_VERSION, path, "PATCH", body=req_body, version=VNFPM_V2_VERSION,
content_type="application/mergepatch+json") content_type="application/merge-patch+json")
def pm_threshold(self, req_body): def pm_threshold(self, req_body):
path = "/pm_threshold" path = "/pm_threshold"

88
tacker/tests/unit/sol_refactored/api/test_prometheus_plugin.py
View File

@@ -1,88 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
from tacker import context
from tacker.sol_refactored.api import prometheus_plugin_wsgi as pp_wsgi
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored import objects
from tacker.tests.unit import base
from unittest import mock
class TestPrometheusPlugin(base.TestCase):
def setUp(self):
super(TestPrometheusPlugin, self).setUp()
objects.register_all()
self.context = context.get_admin_context()
self.request = mock.Mock()
self.request.context = self.context
@mock.patch.object(pp_wsgi.PrometheusPluginErrorResponse, 'serialize')
def test_response(self, mock_serialize_pp):
class _Test():
def __init__(self, ctx, title):
self.status = 200
self.detail = 'detail'
self.title = title
self.method = 'GET'
self.url = 'url'
self.environ = None
self.body = {}
self.context = ctx
self.status_int = 200
def best_match_content_type(self):
return 'application/json'
def serialize(self, accept):
if self.title == 'error':
raise sol_ex.SolValidationError(
detail='test error')
return self
def test(*args, **kwargs):
return (None, None, None)
def test2(*args, **kwargs):
return _Test(None, None)
def test3(*args, **kwargs):
return _Test(None, 'error')
# make responses
pp_wsgi.PrometheusPluginResponse(
200, {}, content_type='content_type')
pp_wsgi.PrometheusPluginErrorResponse(
_Test(self.context, None), None)
pp_wsgi.PrometheusPluginErrorResponse(
_Test(self.context, 'title'), None)
# no error
p = pp_wsgi.PrometheusPluginResource(
None, 'tacker_prometheus_plugin_api:prometheus_plugin:alert')
p(_Test(self.context, None))
# raise unknown error
p = pp_wsgi.PrometheusPluginResource(
None, 'tacker_prometheus_plugin_api:prometheus_plugin:alert')
p._deserialize_request = test
p._check_policy = test
p._dispatch = test2
p(_Test(self.context, None))
mock_serialize_pp.side_effect = _Test(self.context, 'error')
p._dispatch = test3
p(_Test(self.context, 'error'))

87
tacker/tests/unit/sol_refactored/api/test_server_notification.py
View File

@@ -1,87 +0,0 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
from tacker import context
from tacker.sol_refactored.api import server_notification_wsgi as sn_wsgi
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored import objects
from tacker.tests.unit import base
from unittest import mock
class TestServerNotification(base.TestCase):
def setUp(self):
super(TestServerNotification, self).setUp()
objects.register_all()
self.context = context.get_admin_context()
self.request = mock.Mock()
self.request.context = self.context
@mock.patch.object(sn_wsgi.ServerNotificationErrorResponse, 'serialize')
def test_response(self, mock_serialize_pp):
class _Test():
def __init__(self, ctx, title):
self.status = 200
self.detail = 'detail'
self.title = title
self.method = 'GET'
self.url = 'url'
self.environ = None
self.body = {}
self.context = ctx
self.status_int = 200
def best_match_content_type(self):
return 'application/json'
def serialize(self, _):
if self.title == 'error':
raise sol_ex.SolValidationError(
detail='test error')
return self
def test(*args, **kwargs):
return (None, None, None)
def test2(*args, **kwargs):
return _Test(None, None)
def test3(*args, **kwargs):
return _Test(None, 'error')
# make responses
sn_wsgi.ServerNotificationResponse(
200, {}, content_type='content_type')
sn_wsgi.ServerNotificationErrorResponse(
_Test(self.context, None), None)
sn_wsgi.ServerNotificationErrorResponse(
_Test(self.context, 'title'), None)
# no error
p = sn_wsgi.ServerNotificationResource(
None, 'tacker_server_notification_api:server_notification:notify')
p(_Test(self.context, None))
# raise unknown error
p = sn_wsgi.ServerNotificationResource(
None, 'tacker_server_notification_api:server_notification:notify')
p._deserialize_request = test
p._check_policy = test
p._dispatch = test2
p(_Test(self.context, None))
mock_serialize_pp.side_effect = _Test(self.context, 'error')
p._dispatch = test3
p(_Test(self.context, 'error'))

4
tacker/tests/unit/sol_refactored/api/test_wsgi.py
View File

@@ -46,7 +46,9 @@ class TestWsgi(base.TestCase):
self.assertEqual(problem_details, response.body) self.assertEqual(problem_details, response.body)
def test_check_api_version_no_version(self): def test_check_api_version_no_version(self):
resource = sol_wsgi.SolResource(sol_wsgi.SolAPIController()) controller = sol_wsgi.SolAPIController()
controller.supported_api_versions = mock.Mock(return_value=['1.0'])
resource = sol_wsgi.SolResource(controller)
request = mock.Mock() request = mock.Mock()
request.headers = {} request.headers = {}
self.assertRaises(sol_ex.APIVersionMissing, self.assertRaises(sol_ex.APIVersionMissing,

8
tacker/tests/unit/sol_refactored/common/test_prometheus_plugin.py
View File

@@ -1425,13 +1425,13 @@ class TestPrometheusPluginFm(base.TestCase):
prometheus_plugin.PrometheusPluginFm) prometheus_plugin.PrometheusPluginFm)
self.assertIsInstance(pp._instance, mon_base.MonitoringPluginStub) self.assertIsInstance(pp._instance, mon_base.MonitoringPluginStub)
def test_pm_no_body(self): def test_fm_no_body(self):
self.config_fixture.config( self.config_fixture.config(
group='prometheus_plugin', performance_management=True) group='prometheus_plugin', fault_management=True)
pp = mon_base.MonitoringPlugin.get_instance( pp = mon_base.MonitoringPlugin.get_instance(
prometheus_plugin.PrometheusPluginPm) prometheus_plugin.PrometheusPluginFm)
self.assertRaises( self.assertRaises(
sol_ex.PrometheusPluginValidationError, sol_ex.SolValidationError,
pp._alert, self.request) pp._alert, self.request)

2
tacker/tests/unit/sol_refactored/common/test_server_notification.py
View File

@@ -124,7 +124,7 @@ class TestServerNotification(base.TestCase):
sn = mon_base.MonitoringPlugin.get_instance( sn = mon_base.MonitoringPlugin.get_instance(
server_notification.ServerNotification) server_notification.ServerNotification)
self.assertRaises( self.assertRaises(
sol_ex.ServerNotificationValidationError, sol_ex.SolValidationError,
sn.notify, self.request, 'test_id') sn.notify, self.request, 'test_id')
def test_constructor_error(self): def test_constructor_error(self):

8
tacker/tests/unit/sol_refactored/controller/test_server_notification.py
View File

@@ -152,7 +152,7 @@ class TestServerNotification(base.TestCase):
self.config_fixture.config( self.config_fixture.config(
group='server_notification', server_notification=True) group='server_notification', server_notification=True)
self.assertRaises( self.assertRaises(
sol_ex.ServerNotificationValidationError, sol_ex.SolValidationError,
self.controller.notify, self.controller.notify,
request=self.request, request=self.request,
vnf_instance_id='test_id', vnf_instance_id='test_id',
@@ -169,7 +169,7 @@ class TestServerNotification(base.TestCase):
mock_inst.return_value = objects.VnfInstanceV2.from_dict(_inst) mock_inst.return_value = objects.VnfInstanceV2.from_dict(_inst)
self.assertRaises( self.assertRaises(
sol_ex.ServerNotificationValidationError, sol_ex.SolValidationError,
self.controller.notify, request=self.request, self.controller.notify, request=self.request,
vnf_instance_id='test_id', vnf_instance_id='test_id',
server_id='test_server_id', body=_body) server_id='test_server_id', body=_body)
@@ -185,7 +185,7 @@ class TestServerNotification(base.TestCase):
mock_inst.return_value = objects.VnfInstanceV2.from_dict(_inst) mock_inst.return_value = objects.VnfInstanceV2.from_dict(_inst)
self.assertRaises( self.assertRaises(
sol_ex.ServerNotificationValidationError, sol_ex.SolValidationError,
self.controller.notify, request=self.request, self.controller.notify, request=self.request,
vnf_instance_id='test_id', vnf_instance_id='test_id',
server_id='test_server_id', body=_body) server_id='test_server_id', body=_body)
@@ -196,7 +196,7 @@ class TestServerNotification(base.TestCase):
group='server_notification', server_notification=True) group='server_notification', server_notification=True)
mock_inst.return_value = None mock_inst.return_value = None
self.assertRaises( self.assertRaises(
sol_ex.ServerNotificationValidationError, sol_ex.SolValidationError,
self.controller.notify, request=self.request, self.controller.notify, request=self.request,
vnf_instance_id='test_id', vnf_instance_id='test_id',
server_id='test_server_id', body=_body) server_id='test_server_id', body=_body)

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

@@ -59,7 +59,7 @@ class TestVnffmV1(base.BaseTestCase):
self.assertEqual(['application/json', 'text/plain'], result) self.assertEqual(['application/json', 'text/plain'], result)
result = self.controller.allowed_content_types('update') result = self.controller.allowed_content_types('update')
self.assertEqual(['application/mergepatch+json', 'application/json', self.assertEqual(['application/merge-patch+json', 'application/json',
'text/plain'], result) 'text/plain'], result)
@mock.patch.object(alarm_utils, 'get_alarms_all') @mock.patch.object(alarm_utils, 'get_alarms_all')

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

@@ -388,7 +388,8 @@ class TestVnfpmV2(base.BaseTestCase):
def test_allowed_content_types(self): def test_allowed_content_types(self):
result = self.controller.allowed_content_types('update') result = self.controller.allowed_content_types('update')
top = ['application/mergepatch+json', 'application/json', 'text/plain'] top = ['application/merge-patch+json', 'application/json',
'text/plain']
self.assertEqual(top, result) self.assertEqual(top, result)
result = self.controller.allowed_content_types('create') result = self.controller.allowed_content_types('create')