Coverage for osm_nbi/osm_vnfm/vnf_subscription.py: 16%

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 

18from osm_nbi.subscription_topics import CommonSubscriptions 

19from osm_nbi.validation import vnf_subscription 

20 

21 

22class 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