-# Copyright (c) 2016 Cisco and/or its affiliates.
+# Copyright (c) 2017 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:
import HoneycombUtil as HcUtil
-# pylint: disable=too-many-public-methods
-# pylint: disable=too-many-lines
class InterfaceKeywords(object):
"""Keywords for Interface manipulation.
"temporary-valid-lifetime",
"temporary-preferred-lifetime")
ETH_PARAMS = ("mtu", )
- ROUTING_PARAMS = ("vrf-id", )
+ ROUTING_PARAMS = ("ipv4-vrf-id", "ipv6-vrf-id")
VXLAN_PARAMS = ("src", "dst", "vni", "encap-vrf-id")
L2_PARAMS = ("bridge-domain", "split-horizon-group",
"bridged-virtual-interface")
:rtype: dict
"""
+ try:
+ interface = Topology.convert_interface_reference(
+ node, interface, "name")
+ except RuntimeError:
+ if isinstance(interface, basestring):
+ # Probably name of a custom interface (TAP, VxLAN, Vhost, ...)
+ pass
+ else:
+ raise
+
intfs = InterfaceKeywords.get_all_interfaces_oper_data(node)
for intf in intfs:
if intf["name"] == interface:
path = ("interfaces", ("interface", "name", interface), "ietf-ip:ipv4",
"address")
if isinstance(network, basestring):
- address = {"address": [{"ip": ip_addr, "netmask": network}, ]}
+ address = [{"ip": ip_addr, "netmask": network}]
elif isinstance(network, int) and (0 < network < 33):
- address = {"address": [{"ip": ip_addr, "prefix-length": network}, ]}
+ address = [{"ip": ip_addr, "prefix-length": network}]
else:
raise HoneycombError("Value {0} is not a valid netmask or network "
"prefix length.".format(network))
:rtype: bytearray
"""
+ interface = Topology.convert_interface_reference(
+ node, interface, "name")
+
path = ("interfaces", ("interface", "name", interface), "ietf-ip:ipv4",
"neighbor")
neighbor = [{"ip": ip_addr, "link-layer-address": link_layer_address}, ]
:rtype: bytearray
"""
+ interface = Topology.convert_interface_reference(
+ node, interface, "name")
+
path = ("interfaces", ("interface", "name", interface), "ietf-ip:ipv6")
address = {"address": [{"ip": ip_addr, "prefix-length": prefix_len}, ]}
return InterfaceKeywords._set_interface_properties(
:rtype: bytearray
"""
+ interface = Topology.convert_interface_reference(
+ node, interface, "name")
+
path = ("interfaces", ("interface", "name", interface), "ietf-ip:ipv6",
"address")
address = [{"ip": ip_addr, "prefix-length": prefix_len}, ]
:rtype: bytearray
"""
+ interface = Topology.convert_interface_reference(
+ node, interface, "name")
+
path = ("interfaces", ("interface", "name", interface), "ietf-ip:ipv6",
"neighbor")
neighbor = [{"ip": ip_addr, "link-layer-address": link_layer_address}, ]
:raises HoneycombError: If the parameter is not valid.
"""
+ interface = Topology.convert_interface_reference(
+ node, interface, "name")
+
if param not in InterfaceKeywords.ROUTING_PARAMS:
raise HoneycombError("The parameter {0} is invalid.".format(param))
path = ("interfaces",
("interface", "name", super_interface),
"vpp-vlan:sub-interfaces",
- ("sub-interface", "identifier", identifier),
+ ("sub-interface", "identifier", int(identifier)),
"enabled")
return InterfaceKeywords._set_interface_properties(
"information about the tag-rewrite.")
@staticmethod
- def add_ipv4_address_to_sub_interface(node, super_interface, identifier,
- ip_addr, network):
+ def add_ip_address_to_sub_interface(node, super_interface, identifier,
+ ip_addr, network, ip_version):
"""Add an ipv4 address to the specified sub-interface, with the provided
netmask or network prefix length. Any existing ipv4 addresses on the
sub-interface will be replaced.
:param identifier: The ID of sub-interface.
:param ip_addr: IPv4 address to be set.
:param network: Network mask or network prefix length.
+ :param ip_version: ipv4 or ipv6
:type node: dict
:type super_interface: str
:type identifier: int
:type ip_addr: str
:type network: str or int
+ :type ip_version: string
:returns: Content of response.
:rtype: bytearray
:raises HoneycombError: If the provided netmask or prefix is not valid.
("interface", "name", super_interface),
"vpp-vlan:sub-interfaces",
("sub-interface", "identifier", int(identifier)),
- "ipv4")
+ ip_version.lower())
- if isinstance(network, basestring):
+ if isinstance(network, basestring) and ip_version.lower() == "ipv4":
address = {"address": [{"ip": ip_addr, "netmask": network}, ]}
elif isinstance(network, int) and 0 < network < 33:
node, super_interface, path, address)
@staticmethod
- def remove_all_ipv4_addresses_from_sub_interface(node, super_interface, # pylint: disable=invalid-name
- identifier):
+ def remove_all_ip_addresses_from_sub_interface(node, super_interface,
+ identifier, ip_version):
"""Remove all ipv4 addresses from the specified sub-interface.
:param node: Honeycomb node.
:param super_interface: Super interface.
:param identifier: The ID of sub-interface.
+ :param ip_version: ipv4 or ipv6
:type node: dict
:type super_interface: str
:type identifier: int
+ :type ip_version: string
:returns: Content of response.
:rtype: bytearray
"""
("interface", "name", super_interface),
"vpp-vlan:sub-interfaces",
("sub-interface", "identifier", int(identifier)),
- "ipv4", "address")
+ str(ip_version), "address")
return InterfaceKeywords._set_interface_properties(
node, super_interface, path, None)
@staticmethod
- def compare_data_structures(data, ref, ignore=(), list_order=True):
- """Checks if data obtained from UUT is as expected.
+ def compare_data_structures(data, ref, _path=''):
+ """Checks if data obtained from UUT is as expected. If it is not,
+ proceeds down the list/dictionary tree and finds the point of mismatch.
:param data: Data to be checked.
:param ref: Referential data used for comparison.
- :param ignore: Dictionary keys to be ignored.
- :param list_order: Whether to consider the order of list items\
- in comparison.
+ :param _path: Used in recursive calls, stores the path taken down
+ the JSON tree.
:type data: dict
:type ref: dict
- :type ignore: iterable
- :type list_order: bool
+ :type _path: str
- :raises HoneycombError: If a parameter from referential data is not
- present in operational data or if it has different value.
+ :raises HoneycombError: If the data structures do not match in some way,
+ or if they are not in deserialized JSON format.
"""
- errors = ""
-
- for key, item in ref.items():
- if key in ignore:
- continue
- try:
- if data[key] != item:
- if not list_order and sorted(data[key]) == sorted(item):
- pass
+ if data == ref:
+ return True
+
+ elif isinstance(data, dict) and isinstance(ref, dict):
+ for key in ref:
+ if key not in data:
+ raise HoneycombError(
+ "Key {key} is not present in path {path}. Keys in path:"
+ "{data_keys}".format(
+ key=key,
+ path=_path,
+ data_keys=data.keys()))
+
+ if data[key] != ref[key]:
+ if isinstance(data[key], list) \
+ or isinstance(data[key], dict):
+ InterfaceKeywords.compare_data_structures(
+ data[key], ref[key],
+ _path + '[{0}]'.format(key))
else:
- errors += ("\nThe value of parameter '{0}' is "
- "incorrect. It should be "
- "'{1}' but it is '{2}'".
- format(key, item, data[key]))
- except KeyError:
- errors += ("\nThe parameter '{0}' is not present in "
- "operational data".format(key))
+ raise HoneycombError(
+ "Data mismatch, key {key} in path {path} has value"
+ " {data}, but should be {ref}".format(
+ key=key,
+ path=_path,
+ data=data[key],
+ ref=ref[key]))
+
+ elif isinstance(data, list) and isinstance(ref, list):
+ for item in ref:
+ if item not in data:
+ if isinstance(item, dict):
+ InterfaceKeywords.compare_data_structures(
+ data[0], item,
+ _path + '[{0}]'.format(ref.index(item)))
+ else:
+ raise HoneycombError(
+ "Data mismatch, list item {index} in path {path}"
+ " has value {data}, but should be {ref}".format(
+ index=ref.index(item),
+ path=_path,
+ data=data[0],
+ ref=item))
- if errors:
- raise HoneycombError(errors)
+ else:
+ raise HoneycombError(
+ "Unexpected data type {data_type} in path {path}, reference"
+ " type is {ref_type}. Must be list or dictionary.".format(
+ data_type=type(data),
+ ref_type=type(ref),
+ path=_path))
@staticmethod
def compare_interface_lists(list1, list2):
interface = interface.replace("/", "%2F")
data = {
- "v3po:acl": {
+ "vpp-interface-acl:acl": {
"ingress": {
"ip4-acl": {
"classify-table": table_name
}
}
- path = "/interface/" + interface + "/v3po:acl"
+ path = "/interface/" + interface + "/vpp-interface-acl:acl"
status_code, resp = HcUtil.\
put_honeycomb_data(node, "config_vpp_interfaces", data, path,
data_representation=DataRepresentation.JSON)
interface = interface.replace("/", "%2F")
- path = "/interface/" + interface + "/v3po:acl"
+ path = "/interface/" + interface + "/vpp-interface-acl:acl"
status_code, resp = HcUtil.\
delete_honeycomb_data(node, "config_vpp_interfaces", path)
"""
interface = intf.replace("/", "%2F")
- path = "/interface/{0}/pbb-rewrite/".format(interface)
+ path = "/interface/{0}/pbb-rewrite".format(interface)
status_code, resp = HcUtil. \
put_honeycomb_data(node, "config_vpp_interfaces", params, path,
data_representation=DataRepresentation.JSON)
return True
raise HoneycombError("Interface index {0} not present in list"
" of disabled interfaces.".format(interface))
+
+ @staticmethod
+ def configure_interface_span(node, dst_interface, src_interfaces=None):
+ """Configure SPAN port mirroring on the specified interfaces. If no
+ source interface is provided, SPAN will be disabled.
+
+ :param node: Honeycomb node.
+ :param dst_interface: Interface to mirror packets to.
+ :param src_interfaces: List of interfaces to mirror packets from.
+ :type node: dict
+ :type dst_interface: str
+ :type src_interfaces: list of dict
+ :returns: Content of response.
+ :rtype: bytearray
+ :raises HoneycombError: If SPAN could not be configured.
+ """
+
+ interface = Topology.convert_interface_reference(
+ node, dst_interface, "name")
+ interface = interface.replace("/", "%2F")
+ path = "/interface/" + interface + "/span"
+
+ if not src_interfaces:
+ status_code, _ = HcUtil.delete_honeycomb_data(
+ node, "config_vpp_interfaces", path)
+ else:
+ for src_interface in src_interfaces:
+ src_interface["iface-ref"] = Topology.\
+ convert_interface_reference(
+ node, src_interface["iface-ref"], "name")
+ data = {
+ "span": {
+ "mirrored-interfaces": {
+ "mirrored-interface": src_interfaces
+ }
+ }
+ }
+
+ status_code, _ = HcUtil.put_honeycomb_data(
+ node, "config_vpp_interfaces", data, path)
+
+ if status_code not in (HTTPCodes.OK, HTTPCodes.ACCEPTED):
+ raise HoneycombError(
+ "Configuring SPAN failed. Status code:{0}".format(status_code))
+
+ @staticmethod
+ def add_interface_local0_to_topology(node):
+ """Use Topology methods to add interface "local0" to working topology,
+ if not already present.
+
+ :param node: DUT node.
+ :type node: dict
+ """
+
+ if Topology.get_interface_by_sw_index(node, 0) is None:
+ local0_key = Topology.add_new_port(node, "localzero")
+ Topology.update_interface_sw_if_index(node, local0_key, 0)
+ Topology.update_interface_name(node, local0_key, "local0")
Code Review / csit.git / blobdiff