--- /dev/null
+# Copyright 2019 ZTE Corporation.
+#
+# 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 logging
+import uuid
+import requests
+from rest_framework import status
+from catalog.packages import const
+from catalog.pub.database.models import VnfPkgSubscriptionModel, NsdmSubscriptionModel
+from catalog.pub.database.models import VnfPackageModel
+import catalog.pub.utils.timeutil
+from catalog.pub.utils.values import remove_none_key
+from catalog.pub.config import config as pub_config
+import traceback
+
+logger = logging.getLogger(__name__)
+
+
+class NotificationsUtil(object):
+ def __init__(self):
+ pass
+
+ def send_notification(self, notification, filters, isvnfpkg):
+ logger.info("Send Notifications to the callbackUri")
+ subscriptions_filter = {v + "__contains": notification[k] for k, v in filters.items()}
+ logger.debug('send_notification subscriptions_filter = %s' % subscriptions_filter)
+ subscriptions_filter = remove_none_key(subscriptions_filter)
+ if isvnfpkg:
+ subscriptions = VnfPkgSubscriptionModel.objects.filter(**subscriptions_filter)
+ subscription_root_uri = const.VNFPKG_SUBSCRIPTION_ROOT_URI
+ else:
+ subscriptions = NsdmSubscriptionModel.objects.filter(**subscriptions_filter)
+ subscription_root_uri = const.NSDM_SUBSCRIPTION_ROOT_URI
+ if not subscriptions.exists():
+ logger.info("No subscriptions created for the filters %s" % notification)
+ return
+ logger.info("Start sending notifications")
+ for sub in subscriptions:
+ # set subscription id
+ if isvnfpkg:
+ notification["subscriptionId"] = sub.subscription_id
+ else:
+ notification["subscriptionId"] = sub.subscriptionid
+ notification['_links']['subscription'] = {
+ 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ subscription_root_uri,
+ notification["subscriptionId"])
+ }
+ callbackuri = sub.callback_uri
+ """
+ auth_info = json.loads(sub.auth_info)
+ if auth_info["authType"] == const.OAUTH2_CLIENT_CREDENTIALS:
+ pass
+ """
+ self.post_notification(callbackuri, notification)
+
+ def post_notification(self, callbackuri, notification):
+ """
+ params = auth_info.get("paramsBasic", {})
+ username, password = params.get("userName"), params.get("password")
+ logger.info("Sending notification to %s, %s", callbackuri, params)
+ resp = None
+ if username:
+ resp = requests.post(callbackuri,
+ data=notification,
+ auth=HTTPBasicAuth(username, password))
+ else:
+ """
+
+ try:
+ resp = requests.post(callbackuri, data=notification, headers={'Connection': 'close'})
+ if resp.status_code != status.HTTP_204_NO_CONTENT:
+ logger.error("Sendingnotification to %s failed: %s" % (callbackuri, resp.text))
+ else:
+ logger.info("Sending notification to %s successfully.", callbackuri)
+ except:
+ logger.error("Post notification failed.")
+ logger.error(traceback.format_exc())
+
+
+def prepare_vnfpkg_notification(vnf_pkg_id, notification_type, pkg_change_type, operational_state):
+ logger.info('Start to prepare notification')
+ vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
+ vnfd_id = None
+ if vnf_pkg:
+ vnfd_id = vnf_pkg[0].vnfdId
+ notification_content = {
+ 'id': str(uuid.uuid4()), # shall be the same if sent multiple times due to multiple subscriptions.
+ 'notificationType': notification_type,
+ # set 'subscriptionId' after filtering for subscribers
+ 'timeStamp': catalog.pub.utils.timeutil.now_time(),
+ 'vnfPkgId': vnf_pkg_id,
+ 'vnfdId': vnfd_id,
+ 'changeType': pkg_change_type,
+ 'operationalState': operational_state,
+ '_links': {
+ 'vnfPackage': {
+ 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.PKG_URL_PREFIX,
+ vnf_pkg_id)
+ }
+ }
+ }
+ return notification_content
+
+
+def prepare_nsd_notification(nsd_info_id, nsd_id, notification_type, failure_details=None, operational_state=None):
+ logger.info('Start to prepare notification')
+ notification_content = {
+ 'id': str(uuid.uuid4()), # shall be the same if sent multiple times due to multiple subscriptions.
+ 'notificationType': notification_type,
+ # set 'subscriptionId' after filtering for subscribers
+ 'timeStamp': catalog.pub.utils.timeutil.now_time(),
+ 'nsdInfoId': nsd_info_id,
+ 'nsdId': nsd_id,
+ 'onboardingFailureDetails': failure_details,
+ 'nsdOperationalState': operational_state,
+ '_links': {
+ 'nsdInfo': {
+ 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSD_URL_PREFIX,
+ nsd_info_id)
+ }
+ }
+ }
+ return notification_content
+
+
+def prepare_pnfd_notification(pnfd_info_id, pnfd_id, notification_type, failure_details=None):
+ logger.info('Start to prepare notification')
+ notification_content = {
+ 'id': str(uuid.uuid4()), # shall be the same if sent multiple times due to multiple subscriptions.
+ 'notificationType': notification_type,
+ # set 'subscriptionId' after filtering for subscribers
+ 'timeStamp': catalog.pub.utils.timeutil.now_time(),
+ 'pnfdInfoIds': pnfd_info_id,
+ 'pnfdId': pnfd_id,
+ 'onboardingFailureDetails': failure_details,
+ '_links': {
+ 'pnfdInfo': {
+ 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSD_URL_PREFIX,
+ pnfd_info_id)
+ }
+ }
+ }
+ return notification_content
from catalog.pub.exceptions import CatalogException, ResourceNotFoundException
from catalog.pub.utils.values import ignore_case_get
from catalog.pub.utils import fileutil, toscaparser
-from catalog.packages.const import PKG_STATUS
+from catalog.packages import const
+from catalog.packages.biz.notificationsutil import prepare_vnfpkg_notification, NotificationsUtil
logger = logging.getLogger(__name__)
vnf_pkg_id = str(uuid.uuid4())
VnfPackageModel.objects.create(
vnfPackageId=vnf_pkg_id,
- onboardingState=PKG_STATUS.CREATED,
- operationalState=PKG_STATUS.DISABLED,
- usageState=PKG_STATUS.NOT_IN_USE,
+ onboardingState=const.PKG_STATUS.CREATED,
+ operationalState=const.PKG_STATUS.DISABLED,
+ usageState=const.PKG_STATUS.NOT_IN_USE,
userDefinedData=json.dumps(user_defined_data)
)
data = {
"id": vnf_pkg_id,
- "onboardingState": PKG_STATUS.CREATED,
- "operationalState": PKG_STATUS.DISABLED,
- "usageState": PKG_STATUS.NOT_IN_USE,
+ "onboardingState": const.PKG_STATUS.CREATED,
+ "operationalState": const.PKG_STATUS.DISABLED,
+ "usageState": const.PKG_STATUS.NOT_IN_USE,
"userDefinedData": user_defined_data,
"_links": None
}
if del_vnfd_id == vnf["properties"]["descriptor_id"]:
raise CatalogException('VNFD(%s) is referenced.' % del_vnfd_id)
vnf_pkg.delete()
+ send_notification(vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.CHANGE,
+ const.PKG_CHANGE_TYPE.PKG_DELETE)
+
vnf_pkg_path = os.path.join(CATALOG_ROOT_PATH, vnf_pkg_id)
fileutil.delete_dirs(vnf_pkg_path)
logger.info('VNF package(%s) has been deleted.' % vnf_pkg_id)
# if vnf_pkg[0].onboardingState != PKG_STATUS.CREATED:
# logger.error("VNF package(%s) is not CREATED" % vnf_pkg_id)
# raise CatalogException("VNF package(%s) is not CREATED" % vnf_pkg_id)
- vnf_pkg.update(onboardingState=PKG_STATUS.UPLOADING)
+ vnf_pkg.update(onboardingState=const.PKG_STATUS.UPLOADING)
+ send_notification(vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.ONBOARDING,
+ const.PKG_CHANGE_TYPE.OP_STATE_CHANGE)
local_file_name = save(remote_file, vnf_pkg_id)
logger.info('VNF package(%s) has been uploaded.' % vnf_pkg_id)
if not nf_pkg.exists():
logger.error('VNF package(%s) does not exist.' % vnf_pkg_id)
raise ResourceNotFoundException('VNF package(%s) does not exist.' % vnf_pkg_id)
- if nf_pkg[0].onboardingState != PKG_STATUS.ONBOARDED:
+ if nf_pkg[0].onboardingState != const.PKG_STATUS.ONBOARDED:
raise CatalogException("VNF package (%s) is not on-boarded" % vnf_pkg_id)
local_file_path = nf_pkg[0].localFilePath
def upload_vnf_pkg_from_uri(self):
logger.info("Start to upload VNF packge(%s) from URI..." % self.vnf_pkg_id)
vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=self.vnf_pkg_id)
- if vnf_pkg[0].onboardingState != PKG_STATUS.CREATED:
+ if vnf_pkg[0].onboardingState != const.PKG_STATUS.CREATED:
logger.error("VNF package(%s) is not CREATED" % self.vnf_pkg_id)
raise CatalogException("VNF package (%s) is not created" % self.vnf_pkg_id)
- vnf_pkg.update(onboardingState=PKG_STATUS.UPLOADING)
+ vnf_pkg.update(onboardingState=const.PKG_STATUS.UPLOADING)
+ send_notification(self.vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.ONBOARDING,
+ const.PKG_CHANGE_TYPE.OP_STATE_CHANGE)
uri = ignore_case_get(self.data, "addressInformation")
response = urllib.request.urlopen(uri)
def parse_vnfd_and_save(vnf_pkg_id, vnf_pkg_path):
logger.info('Start to process VNF package(%s)...' % vnf_pkg_id)
vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
- vnf_pkg.update(onboardingState=PKG_STATUS.PROCESSING)
+ vnf_pkg.update(onboardingState=const.PKG_STATUS.PROCESSING)
vnfd_json = toscaparser.parse_vnfd(vnf_pkg_path)
vnfd = json.JSONDecoder().decode(vnfd_json)
vnfdVersion=vnfd_ver,
vnfSoftwareVersion=vnf_software_version,
vnfdModel=vnfd_json,
- onboardingState=PKG_STATUS.ONBOARDED,
- operationalState=PKG_STATUS.ENABLED,
- usageState=PKG_STATUS.NOT_IN_USE,
+ onboardingState=const.PKG_STATUS.ONBOARDED,
+ operationalState=const.PKG_STATUS.ENABLED,
+ usageState=const.PKG_STATUS.NOT_IN_USE,
localFilePath=vnf_pkg_path,
vnfPackageUri=os.path.split(vnf_pkg_path)[-1]
)
def handle_upload_failed(vnf_pkg_id):
vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
- vnf_pkg.update(onboardingState=PKG_STATUS.CREATED)
+ vnf_pkg.update(onboardingState=const.PKG_STATUS.CREATED)
+
+
+def send_notification(pkg_id, type, pkg_change_type, operational_state=None):
+ data = prepare_vnfpkg_notification(vnf_pkg_id=pkg_id,
+ notification_type=type,
+ pkg_change_type=pkg_change_type,
+ operational_state=operational_state)
+ filters = {
+ 'vnfdId': 'vnfd_id',
+ 'vnfPkgId': 'vnf_pkg_id'
+ }
+ logger.debug('Notify request data = %s' % data)
+ logger.debug('Notify request filters = %s' % filters)
+ NotificationsUtil().send_notification(data, filters, True)
"VnfPackageOnboardingNotification",
"VnfPackageChangeNotification"
]
+PKG_CHANGE_TYPE = enum(OP_STATE_CHANGE="OP_STATE_CHANGE", PKG_DELETE="PKG_DELETE")
+
+PKG_NOTIFICATION_TYPE = enum(ONBOARDING="VnfPackageOnboardingNotification",
+ CHANGE="VnfPackageChangeNotification")
+
+NSD_NOTIFICATION_TYPE = enum(NSD_ONBOARDING="NsdOnBoardingNotification",
+ NSD_ONBOARDING_FAILURE="NsdOnboardingFailureNotification",
+ NSD_CHANGE="NsdChangeNotification",
+ NSD_DELETION="NsdDeletionNotification",
+ PNFD_ONBOARDING="PnfdOnBoardingNotification",
+ PNFD_ONBOARDING_FAILURE="PnfdOnBoardingFailureNotification",
+ PNFD_DELETION="PnfdDeletionNotification")
+
+PKG_URL_PREFIX = "api/vnfpkgm/v1"
+
+NSD_URL_PREFIX = "api/nsd/v1"
VNFPKG_SUBSCRIPTION_ROOT_URI = "api/vnfpkgm/v1/subscriptions/"
from catalog.packages.biz.nsdm_subscription import NsdmSubscription
from catalog.pub.database.models import NsdmSubscriptionModel
+from catalog.packages.biz.notificationsutil import NotificationsUtil, prepare_nsd_notification, prepare_pnfd_notification
+from catalog.packages import const
+from catalog.pub.config import config as pub_config
+import catalog.pub.utils.timeutil
class TestNsdmSubscription(TestCase):
format='json')
self.assertEqual(response.status_code,
status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+class NotificationTest(TestCase):
+ def setUp(self):
+ NsdmSubscriptionModel(subscriptionid="1",
+ callback_uri="http://127.0.0.1/self",
+ notificationTypes=const.NOTIFICATION_TYPES,
+ nsdId="nsdid1",
+ nsdInfoId="nsdinfoid1",
+ pnfdInfoIds="pnfdInfoIds1",
+ pnfdId="pnfdId1"
+ ).save()
+
+ def tearDown(self):
+ NsdmSubscriptionModel.objects.all().delete()
+
+ @mock.patch("requests.post")
+ @mock.patch("uuid.uuid4")
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
+ mock_nowtime.return_value = "nowtime()"
+ mock_uuid.return_value = "1111"
+ notification_content = prepare_nsd_notification("nsdinfoid1", "nsdid1",
+ const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
+ "NSD(nsdid1) already exists.", operational_state=None)
+ filters = {
+ 'nsdInfoId': 'nsdInfoId',
+ 'nsdId': 'nsdId',
+ }
+ NotificationsUtil().send_notification(notification_content, filters, False)
+ expect_callbackuri = "http://127.0.0.1/self"
+ expect_notification = {
+ 'id': "1111",
+ 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
+ 'timeStamp': "nowtime()",
+ 'nsdInfoId': "nsdinfoid1",
+ 'nsdId': "nsdid1",
+ 'onboardingFailureDetails': "NSD(nsdid1) already exists.",
+ 'nsdOperationalState': None,
+ "subscriptionId": "1",
+ '_links': {
+ 'subscription': {
+ 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSDM_SUBSCRIPTION_ROOT_URI,
+ "1")},
+ 'nsdInfo': {
+ 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSD_URL_PREFIX,
+ "nsdinfoid1")
+ }
+ }
+ }
+ mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})
+
+ @mock.patch("requests.post")
+ @mock.patch("uuid.uuid4")
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
+ mock_nowtime.return_value = "nowtime()"
+ mock_uuid.return_value = "1111"
+ notification_content = prepare_pnfd_notification("pnfdInfoIds1", 'pnfdId1',
+ const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING)
+ filters = {
+ 'pnfdId': 'pnfdId',
+ 'pnfdInfoIds': 'pnfdInfoIds',
+ }
+ NotificationsUtil().send_notification(notification_content, filters, False)
+ expect_callbackuri = "http://127.0.0.1/self"
+ expect_notification = {
+ 'id': "1111",
+ 'notificationType': const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
+ 'timeStamp': "nowtime()",
+ 'pnfdInfoIds': "pnfdInfoIds1",
+ 'pnfdId': "pnfdId1",
+ 'onboardingFailureDetails': None,
+ "subscriptionId": "1",
+ '_links': {
+ 'subscription': {
+ 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSDM_SUBSCRIPTION_ROOT_URI,
+ "1")},
+ 'pnfdInfo': {
+ 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSD_URL_PREFIX,
+ "pnfdInfoIds1")
+ }
+ }
+ }
+ mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
+ headers={'Connection': 'close'})
from rest_framework.test import APIClient
from django.test import TestCase
-from catalog.pub.database.models import VnfPkgSubscriptionModel
+from catalog.pub.database.models import VnfPkgSubscriptionModel, VnfPackageModel
from .const import vnf_subscription_data
+from catalog.packages.biz.notificationsutil import NotificationsUtil, prepare_vnfpkg_notification
+from catalog.packages import const
+from catalog.pub.config import config as pub_config
+import catalog.pub.utils.timeutil
class TestNfPackageSubscription(TestCase):
dummy_uuid = str(uuid.uuid4())
response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid)
self.assertEqual(404, response.status_code)
+
+
+class NotificationTest(TestCase):
+ def setUp(self):
+ VnfPackageModel(vnfPackageId="vnfpkgid1",
+ vnfdId="vnfdid1"
+ ).save()
+
+ VnfPkgSubscriptionModel(subscription_id="1",
+ callback_uri="http://127.0.0.1/self",
+ notification_types=const.NOTIFICATION_TYPES,
+ vnfd_id="vnfdid1",
+ vnf_pkg_id="vnfpkgid1"
+ ).save()
+
+ def tearDown(self):
+ VnfPackageModel.objects.all().delete()
+ VnfPkgSubscriptionModel.objects.all().delete()
+
+ @mock.patch("requests.post")
+ @mock.patch("uuid.uuid4")
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ def test_vnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
+ mock_nowtime.return_value = "nowtime()"
+ mock_uuid.return_value = "1111"
+ notification_content = prepare_vnfpkg_notification("vnfpkgid1", const.PKG_NOTIFICATION_TYPE.CHANGE,
+ const.PKG_CHANGE_TYPE.OP_STATE_CHANGE, operational_state=None)
+ filters = {
+ 'vnfdId': 'vnfd_id',
+ 'vnfPkgId': 'vnf_pkg_id'
+ }
+ NotificationsUtil().send_notification(notification_content, filters, True)
+ expect_callbackuri = "http://127.0.0.1/self"
+ expect_notification = {
+ 'id': "1111",
+ 'notificationType': const.PKG_NOTIFICATION_TYPE.CHANGE,
+ 'timeStamp': "nowtime()",
+ 'vnfPkgId': "vnfpkgid1",
+ 'vnfdId': "vnfdid1",
+ 'changeType': const.PKG_CHANGE_TYPE.OP_STATE_CHANGE,
+ 'operationalState': None,
+ "subscriptionId": "1",
+ '_links': {
+ 'subscription': {
+ 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.VNFPKG_SUBSCRIPTION_ROOT_URI,
+ "1")},
+ 'vnfPackage': {
+ 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.PKG_URL_PREFIX,
+ "vnfpkgid1")
+ }
+ }
+ }
+ mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})
import platform
import unittest
import mock
+
from . import fileutil
import urllib
from . import syscomm
Code Review / modeling / etsicatalog.git / commitdiff