# See the License for the specific language governing permissions and
# limitations under the License.
+import json
import logging
+import traceback
import uuid
+
import requests
-import json
-from rest_framework import status
+from django.db.models import Q
from requests.auth import HTTPBasicAuth
-from catalog.packages import const
-from catalog.pub.database.models import VnfPackageModel, VnfPkgSubscriptionModel, NsdmSubscriptionModel
+from rest_framework import status
+
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
-from django.db.models import Q
+from catalog.packages import const
from catalog.packages.serializers.vnf_pkg_notifications import PkgChangeNotificationSerializer, \
PkgOnboardingNotificationSerializer
+from catalog.pub.database.models import VnfPackageModel, VnfPkgSubscriptionModel, NsdmSubscriptionModel
+from catalog.pub.utils.values import remove_none_key
logger = logging.getLogger(__name__)
class NotificationsUtil(object):
+ """
+ Util for notifications
+ """
+
def __init__(self, notification_type):
self.notification_type = notification_type
self.notifyserializer = None
pass
def send_notification(self):
+ """
+ Send notification
+ :return:
+ """
notification = self.prepare_notification()
subscriptions_filter = {v + "__contains": notification[k] for k, v in self.filter.items()}
# set subscription id
notification["subscriptionId"] = sub.get_subscription_id()
notification['_links']['subscription'] = {
- 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
- pub_config.MSB_SERVICE_PORT,
- self.subscription_root_uri,
- notification["subscriptionId"])
+ 'href': '/%s%s' % (self.subscription_root_uri, notification["subscriptionId"])
}
callbackuri = sub.callback_uri
"""
self.post_notification(callbackuri, notification)
def post_notification(self, callbackuri, notification, auth_info=None):
+ """
+ Post notification
+ :param callbackuri:
+ :param notification:
+ :param auth_info:
+ :return:
+ """
try:
if auth_info:
if const.BASIC in auth_info.get("authType", ''):
params = auth_info.get("paramsBasic", {})
username = params.get("userName")
password = params.get("password")
- resp = requests.post(callbackuri, data=notification, headers={'Connection': 'close'},
- auth=HTTPBasicAuth(username, password))
+ resp = requests.post(callbackuri,
+ data=json.dumps(notification),
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ auth=HTTPBasicAuth(username, password),
+ verify=False)
elif const.OAUTH2_CLIENT_CREDENTIALS in auth_info.get("authType", ''):
# todo
pass
# todo
pass
else:
- resp = requests.post(callbackuri, data=notification, headers={'Connection': 'close'})
- if resp.status_code != status.HTTP_204_NO_CONTENT:
- logger.error("Sending notification to %s failed: %s" % (callbackuri, resp.text))
- else:
+ resp = requests.post(callbackuri,
+ data=json.dumps(notification),
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ verify=False)
+
+ if resp.status_code == status.HTTP_204_NO_CONTENT:
logger.info("Sending notification to %s successfully.", callbackuri)
+ else:
+ logger.error("Sending notification to %s failed: %s" % (callbackuri, resp))
except:
logger.error("Post notification failed.")
logger.error(traceback.format_exc())
class PkgNotifications(NotificationsUtil):
+ """
+ Notification Utils for VNF pckages
+ """
+
def __init__(self, notification_type, vnf_pkg_id, change_type=None, operational_state=None):
super(PkgNotifications, self).__init__(notification_type)
self.filter = {
self.notifyserializer = PkgOnboardingNotificationSerializer
def prepare_notification(self):
+ """
+ Prepare notification
+ :return:
+ """
logger.info('Start to prepare Pkgnotification')
vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=self.vnf_pkg_id)
'vnfdId': vnfd_id,
'_links': {
'vnfPackage': {
- 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
- pub_config.MSB_SERVICE_PORT,
- const.PKG_URL_PREFIX,
- self.vnf_pkg_id)
+ 'href': '/%s/vnf_packages/%s' % (const.PKG_URL_PREFIX, self.vnf_pkg_id)
}
}
}
class NsdNotifications(NotificationsUtil):
+ """
+ Notification Util for NS packages
+ """
+
def __init__(self, notification_type, nsd_info_id, nsd_id, failure_details=None, operational_state=None):
super(NsdNotifications, self).__init__(notification_type)
self.filter = {
# self.notifyserializer = PkgOnboardingNotificationSerializer
def prepare_notification(self):
+ """
+ Prepare notification
+ :return:
+ """
logger.info('Start to prepare Nsdnotification')
notification_content = {
'nsdId': self.nsd_id,
'_links': {
'nsdInfo': {
- 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
- pub_config.MSB_SERVICE_PORT,
- const.NSD_URL_PREFIX,
- self.nsd_info_id)
+ 'href': '/%s/ns_descriptors/%s' % (
+ const.NSD_URL_PREFIX, self.nsd_info_id)
}
}
}
class PnfNotifications(NotificationsUtil):
+ """
+ Notification util for PNF package
+ """
def __init__(self, notification_type, pnfd_info_id, pnfd_id, failure_details=None):
super(PnfNotifications, self).__init__(notification_type)
self.filter = {
# self.notifyserializer = PkgOnboardingNotificationSerializer
def prepare_notification(self, *args, **kwargs):
+ """
+ Prepare notification
+ :param args:
+ :param kwargs:
+ :return:
+ """
logger.info('Start to prepare Pnfnotification')
notification_content = {
'id': str(uuid.uuid4()), # shall be the same if sent multiple times due to multiple subscriptions.
'pnfdId': self.pnfd_id,
'_links': {
'pnfdInfo': {
- 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
- pub_config.MSB_SERVICE_PORT,
- const.NSD_URL_PREFIX,
- self.pnfd_info_id)
+ 'href': '/%s/pnf_descriptors/%s' % (const.NSD_URL_PREFIX,
+ self.pnfd_info_id)
}
}
}
Code Review / modeling / etsicatalog.git / blobdiff