X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FVatJsonUtil.py;h=3e956e790ddba3679e1d4c9c8c0caaf83501c7c2;hb=d07f6cae7f18c1513650d4cb690115d60201e704;hp=05c93a1d064fcb46738323e11f8698df47a1affd;hpb=8c12ff59f1a5e750151f5eb0e806dcc80e91c3c2;p=csit.git diff --git a/resources/libraries/python/VatJsonUtil.py b/resources/libraries/python/VatJsonUtil.py index 05c93a1d06..3e956e790d 100644 --- a/resources/libraries/python/VatJsonUtil.py +++ b/resources/libraries/python/VatJsonUtil.py @@ -1,4 +1,4 @@ -# Copyright (c) 2016 Cisco and/or its affiliates. +# Copyright (c) 2021 Cisco and/or its affiliates. # 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: @@ -18,7 +18,7 @@ from robot.api import logger from resources.libraries.python.parsers.JsonParser import JsonParser -class VatJsonUtil(object): +class VatJsonUtil: """Utilities to work with JSON data format from VAT.""" @staticmethod @@ -30,11 +30,11 @@ class VatJsonUtil(object): :param mac_address: MAC address. :type mac_address: str - :return: List representation of MAC address. + :returns: List representation of MAC address. :rtype: list """ - list_mac = [] - for num in mac_address.split(":"): + list_mac = list() + for num in mac_address.split(u":"): list_mac.append(int(num, 16)) return list_mac @@ -49,29 +49,32 @@ class VatJsonUtil(object): :param mac_address: MAC address of interface we are looking for. :type interfaces_list: dict :type mac_address: str - :return: Interface from JSON. + :returns: Interface from JSON. :rtype: dict """ - interface_dict = {} + interface_dict = dict() list_mac_address = VatJsonUtil._convert_mac_to_number_list(mac_address) - logger.trace("MAC address {0} converted to list {1}." - .format(mac_address, list_mac_address)) + logger.trace( + f"MAC address {mac_address} converted to list {list_mac_address}." + ) for interface in interfaces_list: # TODO: create vat json integrity checking and move there - if "l2_address" not in interface: + if u"l2_address" not in interface: raise KeyError( - "key l2_address not found in interface dict." - "Probably input list is not parsed from correct VAT " - "json output.") - if "l2_address_length" not in interface: + u"key l2_address not found in interface dict." + u"Probably input list is not parsed from correct VAT " + u"json output." + ) + if u"l2_address_length" not in interface: raise KeyError( - "key l2_address_length not found in interface " - "dict. Probably input list is not parsed from correct " - "VAT json output.") - mac_from_json = interface["l2_address"][:6] + u"key l2_address_length not found in interface " + u"dict. Probably input list is not parsed from correct " + u"VAT json output." + ) + mac_from_json = interface[u"l2_address"][:6] if mac_from_json == list_mac_address: - if interface["l2_address_length"] != 6: - raise ValueError("l2_address_length value is not 6.") + if interface[u"l2_address_length"] != 6: + raise ValueError(u"l2_address_length value is not 6.") interface_dict = interface break return interface_dict @@ -85,24 +88,114 @@ class VatJsonUtil(object): :param node: Node dictionary. :param interface_dump_json: JSON output from dump_interface_list VAT - command. + command. :type node: dict :type interface_dump_json: str """ interface_list = JsonParser().parse_data(interface_dump_json) - for ifc in node['interfaces'].values(): - if_mac = ifc['mac_address'] + for ifc in node[u"interfaces"].values(): + if_mac = ifc[u"mac_address"] interface_dict = VatJsonUtil.get_vpp_interface_by_mac( - interface_list, if_mac) + interface_list, if_mac + ) if not interface_dict: - raise Exception('Interface {0} not found by MAC {1}' - .format(ifc, if_mac)) - ifc['name'] = interface_dict["interface_name"] - ifc['vpp_sw_index'] = interface_dict["sw_if_index"] - ifc['mtu'] = interface_dict["mtu"] + logger.trace(f"Interface {ifc} not found by MAC {if_mac}") + ifc[u"vpp_sw_index"] = None + continue + ifc[u"name"] = interface_dict[u"interface_name"] + ifc[u"vpp_sw_index"] = interface_dict[u"sw_if_index"] + ifc[u"mtu"] = interface_dict[u"mtu"] @staticmethod - def verify_vat_retval(vat_out, exp_retval=0, err_msg='VAT cmd failed'): + def get_interface_sw_index_from_json(interface_dump_json, interface_name): + """Get sw_if_index from given JSON output by interface name. + + :param interface_dump_json: JSON output from dump_interface_list VAT + command. + :param interface_name: Interface name. + :type interface_dump_json: str + :type interface_name: str + :returns: SW interface index. + :rtype: int + :raises ValueError: If interface not found in interface_dump_json. + """ + logger.trace(interface_dump_json) + interface_list = JsonParser().parse_data(interface_dump_json) + for interface in interface_list: + try: + if interface[u"interface_name"] == interface_name: + index = interface[u"sw_if_index"] + logger.debug( + f"Interface with name {interface_name} " + f"has sw_if_index {index}." + ) + return index + except KeyError: + pass + raise ValueError(f"Interface with name {interface_name} not found.") + + @staticmethod + def get_interface_name_from_json(interface_dump_json, sw_if_index): + """Get interface name from given JSON output by sw_if_index. + + :param interface_dump_json: JSON output from dump_interface_list VAT + command. + :param sw_if_index: SW interface index. + :type interface_dump_json: str + :type sw_if_index: int + :returns: Interface name. + :rtype: str + :raises ValueError: If interface not found in interface_dump_json. + """ + logger.trace(interface_dump_json) + interface_list = JsonParser().parse_data(interface_dump_json) + for interface in interface_list: + try: + if interface[u"sw_if_index"] == sw_if_index: + interface_name = interface[u"interface_name"] + logger.debug( + f"Interface with sw_if_index {sw_if_index} " + f"has name {interface_name}." + ) + return interface_name + except KeyError: + pass + raise ValueError(f"Interface with sw_if_index {sw_if_index} not found.") + + @staticmethod + def get_interface_mac_from_json(interface_dump_json, sw_if_index): + """Get interface MAC address from given JSON output by sw_if_index. + + :param interface_dump_json: JSON output from dump_interface_list VAT + command. + :param sw_if_index: SW interface index. + :type interface_dump_json: str + :type sw_if_index: int + :returns: Interface MAC address. + :rtype: str + :raises ValueError: If interface not found in interface_dump_json. + """ + logger.trace(interface_dump_json) + interface_list = JsonParser().parse_data(interface_dump_json) + for interface in interface_list: + try: + if interface[u"sw_if_index"] == sw_if_index: + mac_from_json = interface[u"l2_address"][:6] \ + if u"l2_address" in list(interface.keys()) else u"" + mac_address = u":".join( + f"{item:02x}" for item in mac_from_json + ) + logger.debug( + f"Interface with sw_if_index {sw_if_index} " + f"has MAC address {mac_address}." + ) + return mac_address + except KeyError: + pass + raise ValueError(f"Interface with sw_if_index {sw_if_index} not found.") + + @staticmethod + def verify_vat_retval(vat_out, exp_retval=0, err_msg=u"VAT cmd failed"): """Verify return value of VAT command. VAT command JSON output should be object (dict in python) or array. We @@ -118,8 +211,8 @@ class VatJsonUtil(object): :type err_msg: str :raises RuntimeError: If VAT command return value is incorrect. """ - if type(vat_out) is dict: - retval = vat_out.get('retval') + if isinstance(vat_out, dict): + retval = vat_out.get(u"retval") if retval is not None: if retval != exp_retval: raise RuntimeError(err_msg)