diff options
| -rw-r--r-- | robotframework-onap/ONAPLibrary/JSON.py | 27 | ||||
| -rw-r--r-- | robotframework-onap/ONAPLibrary/JSONKeywords.py | 65 | ||||
| -rw-r--r-- | robotframework-onap/ONAPLibrary/Protobuf.py | 26 | ||||
| -rw-r--r-- | robotframework-onap/ONAPLibrary/ProtobufKeywords.py | 21 | ||||
| -rw-r--r-- | robotframework-onap/ONAPLibrary/VESProtobuf.py | 120 |
5 files changed, 259 insertions, 0 deletions
diff --git a/robotframework-onap/ONAPLibrary/JSON.py b/robotframework-onap/ONAPLibrary/JSON.py new file mode 100644 index 0000000..6eb1e65 --- /dev/null +++ b/robotframework-onap/ONAPLibrary/JSON.py @@ -0,0 +1,27 @@ +# Copyright 2019 AT&T Intellectual Property. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from ONAPLibrary.robotlibcore import HybridCore +from ONAPLibrary.JSONKeywords import JSONKeywords + + +class JSON(HybridCore): + """JSON is common resource for simple json helper keywords. + """ + + def __init__(self): + self.keyword_implementors = [ + JSONKeywords() + ] + HybridCore.__init__(self, self.keyword_implementors) diff --git a/robotframework-onap/ONAPLibrary/JSONKeywords.py b/robotframework-onap/ONAPLibrary/JSONKeywords.py new file mode 100644 index 0000000..bd1a59d --- /dev/null +++ b/robotframework-onap/ONAPLibrary/JSONKeywords.py @@ -0,0 +1,65 @@ +# Copyright 2019 AT&T Intellectual Property. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import json + +from robot.api.deco import keyword +from deepdiff import DeepDiff + + +class JSONKeywords(object): + """JSON is common resource for simple json helper keywords. + """ + + def __init__(self): + pass + + @keyword + def json_equals(self, left, right): + """JSON Equals takes in two strings or json objects, converts them into json if needed and then compares them, + returning if they are equal or not.""" + if isinstance(left, str) or isinstance(left, unicode): + left_json = json.loads(left) + else: + left_json = left + if isinstance(right, str) or isinstance(right, unicode): + right_json = json.loads(right) + else: + right_json = right + + ddiff = DeepDiff(left_json, right_json, ignore_order=True) + if ddiff == {}: + return True + else: + return False + + @keyword + def make_list_into_dict(self, dict_list, key): + """ Converts a list of dicts that contains a field that has a unique key into a dict of dicts """ + d = {} + if isinstance(dict_list, list): + for thisDict in dict_list: + v = thisDict[key] + d[v] = thisDict + return d + + @keyword + def find_element_in_array(self, searched_array, key, value): + """ Takes in an array and a key value, it will return the items in the array that has a key and value that + matches what you pass in """ + elements = [] + for item in searched_array: + if key in item: + if item[key] == value: + elements.append(item) + return elements diff --git a/robotframework-onap/ONAPLibrary/Protobuf.py b/robotframework-onap/ONAPLibrary/Protobuf.py new file mode 100644 index 0000000..c81a0a2 --- /dev/null +++ b/robotframework-onap/ONAPLibrary/Protobuf.py @@ -0,0 +1,26 @@ +# Copyright 2019 AT&T Intellectual Property. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from ONAPLibrary.robotlibcore import HybridCore +from ONAPLibrary.ProtobufKeywords import ProtobufKeywords + + +class Protobuf(HybridCore): + """ Utilities useful for Protobuf manipulation """ + + def __init__(self): + self.keyword_implementors = [ + ProtobufKeywords() + ] + HybridCore.__init__(self, self.keyword_implementors) diff --git a/robotframework-onap/ONAPLibrary/ProtobufKeywords.py b/robotframework-onap/ONAPLibrary/ProtobufKeywords.py new file mode 100644 index 0000000..9fded9b --- /dev/null +++ b/robotframework-onap/ONAPLibrary/ProtobufKeywords.py @@ -0,0 +1,21 @@ +from ONAPLibrary.VESProtobuf import * +from ONAPLibrary.JSONKeywords import JSONKeywords +from robot.api.deco import keyword + + +class ProtobufKeywords(object): + """ Utilities useful for Protobuf manipulation """ + + def __init__(self): + super(ProtobufKeywords, self).__init__() + + @keyword + def compare_file_to_message(self, file_name, message): + with open(file_name, "rb") as file_to_do: + return self.compare_two_messages(file_to_do.read(), message) + + @staticmethod + def compare_two_messages(left, right): + left_json = VESProtobuf.binary_to_json(left) + right_json = VESProtobuf.binary_to_json(right) + return JSONKeywords().json_equals(left_json, right_json) diff --git a/robotframework-onap/ONAPLibrary/VESProtobuf.py b/robotframework-onap/ONAPLibrary/VESProtobuf.py new file mode 100644 index 0000000..d747a0d --- /dev/null +++ b/robotframework-onap/ONAPLibrary/VESProtobuf.py @@ -0,0 +1,120 @@ +# Copyright 2019 AT&T Intellectual Property. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# noinspection PyPackageRequirements +from google.protobuf import descriptor +# noinspection PyPackageRequirements +from google.protobuf import descriptor_pb2 +# noinspection PyPackageRequirements +from google.protobuf import message_factory +# noinspection PyPackageRequirements +from google.protobuf.json_format import MessageToJson + + +class VESProtobuf(object): + """ non keywords methods related to VES """ + + def __init__(self): + super(VESProtobuf, self).__init__() + + @staticmethod + def create_ves_event(): + file_descriptor_proto = descriptor_pb2.FileDescriptorProto() + file_descriptor_proto.name = 'VesEvent' + VESProtobuf.create_commoneventheader(file_descriptor_proto) + VESProtobuf.create_vesevent(file_descriptor_proto) + return file_descriptor_proto + + @staticmethod + def create_vesevent(file_descriptor_proto): + message_type = file_descriptor_proto.message_type.add() + message_type.name = "VesEvent" + VESProtobuf.create_message_field(message_type, 1, "commonEventHeader", "CommonEventHeader") + VESProtobuf.create_field(message_type, 2, "eventFields", descriptor.FieldDescriptor.TYPE_BYTES) + return message_type + + @staticmethod + def create_commoneventheader(file_descriptor_proto): + message_type = file_descriptor_proto.message_type.add() + message_type.name = "CommonEventHeader" + enum_type = VESProtobuf.create_enum_type(message_type, 'Priority') + VESProtobuf.create_enum_type_value(enum_type, ["PRIORITY_NOT_PROVIDED", "HIGH", "MEDIUM", "NORMAL", "LOW"]) + VESProtobuf.create_field(message_type, 1, "version", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 2, "domain", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 3, "sequence", descriptor.FieldDescriptor.TYPE_UINT32) + VESProtobuf.create_enum_field(message_type, 4, "priority", "Priority") + VESProtobuf.create_field(message_type, 5, "eventId", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 6, "eventName", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 7, "eventType", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 8, "lastEpochMicrosec", descriptor.FieldDescriptor.TYPE_UINT64) + VESProtobuf.create_field(message_type, 9, "startEpochMicrosec", descriptor.FieldDescriptor.TYPE_UINT64) + VESProtobuf.create_field(message_type, 10, "nfNamingCode", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 11, "nfcNamingCode", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 12, "nfVendorName", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 13, "reportingEntityId", descriptor.FieldDescriptor.TYPE_BYTES) + VESProtobuf.create_field(message_type, 14, "reportingEntityName", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 15, "sourceId", descriptor.FieldDescriptor.TYPE_BYTES) + VESProtobuf.create_field(message_type, 16, "sourceName", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 17, "timeZoneOffset", descriptor.FieldDescriptor.TYPE_STRING) + VESProtobuf.create_field(message_type, 18, "vesEventListenerVersion", + descriptor.FieldDescriptor.TYPE_STRING) + return message_type + + @staticmethod + def create_enum_type(desc_proto, name): + enum_type = desc_proto.enum_type.add() + enum_type.name = name + return enum_type + + @staticmethod + def create_enum_type_value(enum_type, value_list): + for i in range(len(value_list)): + enum_type_val = enum_type.value.add() + enum_type_val.name = value_list[i] + enum_type_val.number = i + + @staticmethod + def create_field(desc_proto, number, name, field_type): + field = desc_proto.field.add() + field.number = number + field.name = name + field.type = field_type + + @staticmethod + def create_enum_field(desc_proto, number, name, type_name): + field = desc_proto.field.add() + field.number = number + field.name = name + field.type = descriptor.FieldDescriptor.TYPE_ENUM + field.type_name = type_name + + @staticmethod + def create_message_field(desc_proto, number, name, type_name): + field = desc_proto.field.add() + field.number = number + field.name = name + field.type = descriptor.FieldDescriptor.TYPE_MESSAGE + field.type_name = type_name + + @staticmethod + def get_message_definitions(): + return message_factory.GetMessages((VESProtobuf.create_ves_event(),)) + + @staticmethod + def binary_to_json(binary_message): + defs = VESProtobuf.get_message_definitions() + ves = defs['VesEvent']() + ves.MergeFromString(binary_message) + json = MessageToJson(ves) + return json |