[Jenkins]Jenkins

Code Coverage

Cobertura Coverage Report > osm_nbi.osm_vnfm >

vnf_subscription.py

Trend

Classes100%
 
Lines16%
   
Conditionals100%
 

File Coverage summary

NameClassesLinesConditionals
vnf_subscription.py
100%
1/1
16%
6/37
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
vnf_subscription.py
16%
6/37
N/A

Source

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
1
__author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"




17






18
1
from osm_nbi.subscription_topics import CommonSubscriptions




19
1
from osm_nbi.validation import vnf_subscription




20






21
1
class VnflcmSubscriptionsTopic(CommonSubscriptions):




22
1
    schema_new = vnf_subscription




23
1
    def _subscription_mapper(self, _id, data, table):




24

        """




25

        Performs data transformation on subscription request




26

        :param _id: subscription reference id




27

        :param data: data to be transformed




28

        :param table: table in which transformed data are inserted




29

        """




30
0
        formatted_data = []




31
0
        formed_data = {




32

            "reference": data.get("_id"),




33

            "CallbackUri": data.get("CallbackUri")




34

        }




35
0
        if data.get("authentication"):




36
0
            formed_data.update({"authentication": data.get("authentication")})




37
0
        if data.get("filter"):




38
0
            if data["filter"].get("VnfInstanceSubscriptionFilter"):




39
0
                key = list(data["filter"]["VnfInstanceSubscriptionFilter"].keys())[0]




40
0
                identifier = data["filter"]["VnfInstanceSubscriptionFilter"][key]




41
0
                formed_data.update({"identifier": identifier})




42
0
            if data["filter"].get("notificationTypes"):




43
0
                for elem in data["filter"].get("notificationTypes"):




44
0
                    update_dict = formed_data.copy()




45
0
                    update_dict["notificationType"] = elem




46
0
                    if elem == "VnfIdentifierCreationNotification":




47
0
                        update_dict["operationTypes"] = "CREATE"




48
0
                        update_dict["operationStates"] = "ANY"




49
0
                        formatted_data.append(update_dict)




50
0
                    elif elem == "VnfIdentifierDeletionNotification":




51
0
                        update_dict["operationTypes"] = "DELETE"




52
0
                        update_dict["operationStates"] = "ANY"




53
0
                        formatted_data.append(update_dict)




54
0
                    elif elem == "VnfLcmOperationOccurrenceNotification":




55
0
                        if "operationTypes" in data["filter"].keys():




56
0
                            update_dict["operationTypes"] = data["filter"]["operationTypes"]




57

                        else:




58
0
                            update_dict["operationTypes"] = "ANY"




59
0
                        if "operationStates" in data["filter"].keys():




60
0
                            update_dict["operationStates"] = data["filter"]["operationStates"]




61

                        else:




62
0
                            update_dict["operationStates"] = "ANY"




63
0
                        formatted_data.append(update_dict)




64
0
        self.db.create_list(table, formatted_data)




65
0
        return None