Coverage for osm_nbi/osm_vnfm/vnf_subscription.py: 16%
37 statements
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-30 10:14 +0000
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-30 10:14 +0000
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.
16__author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
18from osm_nbi.subscription_topics import CommonSubscriptions
19from osm_nbi.validation import vnf_subscription
22class VnflcmSubscriptionsTopic(CommonSubscriptions):
23 schema_new = vnf_subscription
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