[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






22
1
class VnflcmSubscriptionsTopic(CommonSubscriptions):




23
1
    schema_new = vnf_subscription




24






25
1
    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
0
        formatted_data = []




33
0
        formed_data = {




34

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




35

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




36

        }




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




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




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




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




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




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




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




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




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




46
0
                    update_dict = formed_data.copy()




47
0
                    update_dict["notificationType"] = elem




48
0
                    if elem == "VnfIdentifierCreationNotification":




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




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




51
0
                        formatted_data.append(update_dict)




52
0
                    elif elem == "VnfIdentifierDeletionNotification":




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




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




55
0
                        formatted_data.append(update_dict)




56
0
                    elif elem == "VnfLcmOperationOccurrenceNotification":




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




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




59

                                "operationTypes"




60

                            ]




61

                        else:




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




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




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




65

                                "operationStates"




66

                            ]




67

                        else:




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




69
0
                        formatted_data.append(update_dict)




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




71
0
        return None