osm_nbi/osm_vnfm/vnf_subscription.py

   1 # Copyright 2021 Selvi Jayaraman (Tata Elxsi)
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 __author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
  17
  18 from osm_nbi.subscription_topics import CommonSubscriptions
  19 from osm_nbi.validation import vnf_subscription
  20
  21
  22 class VnflcmSubscriptionsTopic(CommonSubscriptions):
  23     schema_new = vnf_subscription
  24
  25     def _subscription_mapper(self, _id, data, table):
  26         """
  27         Performs data transformation on subscription request
  28         :param _id: subscription reference id
  29         :param data: data to be transformed
  30         :param table: table in which transformed data are inserted
  31         """
  32         formatted_data = []
  33         formed_data = {
  34             "reference": data.get("_id"),
  35             "CallbackUri": data.get("CallbackUri"),
  36         }
  37         if data.get("authentication"):
  38             formed_data.update({"authentication": data.get("authentication")})
  39         if data.get("filter"):
  40             if data["filter"].get("VnfInstanceSubscriptionFilter"):
  41                 key = list(data["filter"]["VnfInstanceSubscriptionFilter"].keys())[0]
  42                 identifier = data["filter"]["VnfInstanceSubscriptionFilter"][key]
  43                 formed_data.update({"identifier": identifier})
  44             if data["filter"].get("notificationTypes"):
  45                 for elem in data["filter"].get("notificationTypes"):
  46                     update_dict = formed_data.copy()
  47                     update_dict["notificationType"] = elem
  48                     if elem == "VnfIdentifierCreationNotification":
  49                         update_dict["operationTypes"] = "CREATE"
  50                         update_dict["operationStates"] = "ANY"
  51                         formatted_data.append(update_dict)
  52                     elif elem == "VnfIdentifierDeletionNotification":
  53                         update_dict["operationTypes"] = "DELETE"
  54                         update_dict["operationStates"] = "ANY"
  55                         formatted_data.append(update_dict)
  56                     elif elem == "VnfLcmOperationOccurrenceNotification":
  57                         if "operationTypes" in data["filter"].keys():
  58                             update_dict["operationTypes"] = data["filter"][
  59                                 "operationTypes"
  60                             ]
  61                         else:
  62                             update_dict["operationTypes"] = "ANY"
  63                         if "operationStates" in data["filter"].keys():
  64                             update_dict["operationStates"] = data["filter"][
  65                                 "operationStates"
  66                             ]
  67                         else:
  68                             update_dict["operationStates"] = "ANY"
  69                         formatted_data.append(update_dict)
  70         self.db.create_list(table, formatted_data)
  71         return None