X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FIPv6Util.py;h=683f892f621a66efc069060b8cee669131b3244f;hp=437a7c0cbb085496c22294889ce5dd713d4957fd;hb=d68951ac245150eeefa6e0f4156e4c1b5c9e9325;hpb=cbd47fbe97945e9dc6584d08cd2266e3a7536a68 diff --git a/resources/libraries/python/IPv6Util.py b/resources/libraries/python/IPv6Util.py index 437a7c0cbb..683f892f62 100644 --- a/resources/libraries/python/IPv6Util.py +++ b/resources/libraries/python/IPv6Util.py @@ -1,4 +1,4 @@ -# Copyright (c) 2016 Cisco and/or its affiliates. +# Copyright (c) 2019 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: @@ -13,123 +13,72 @@ """IPv6 utilities library.""" -import re +from resources.libraries.python.InterfaceUtil import InterfaceUtil +from resources.libraries.python.IPUtil import IPUtil +from resources.libraries.python.PapiExecutor import PapiSocketExecutor +from resources.libraries.python.topology import NodeType -from resources.libraries.python.ssh import SSH -from resources.libraries.python.VatExecutor import VatTerminal -from resources.libraries.python.topology import Topology - -class IPv6Util(object): +class IPv6Util: """IPv6 utilities""" @staticmethod - def ipv6_ping(src_node, dst_addr, count=3, data_size=56, timeout=1): - """IPv6 ping. + def vpp_ra_suppress_link_layer(node, interface): + """Suppress ICMPv6 router advertisement message for link scope address. - :param src_node: Node where ping run. - :param dst_addr: Destination IPv6 address. - :param count: Number of echo requests. (Optional) - :param data_size: Number of the data bytes. (Optional) - :param timeout: Time to wait for a response, in seconds. (Optional) - :type src_node: dict - :type dst_addr: str - :type count: int - :type data_size: int - :type timeout: int - :return: Number of lost packets. - :rtype: int + :param node: VPP node. + :param interface: Interface name. + :type node: dict + :type interface: str """ - ssh = SSH() - ssh.connect(src_node) - - cmd = "ping6 -c {c} -s {s} -W {W} {dst}".format(c=count, s=data_size, - W=timeout, - dst=dst_addr) - (ret_code, stdout, _) = ssh.exec_command(cmd) - - regex = re.compile(r'(\d+) packets transmitted, (\d+) received') - match = regex.search(stdout) - sent, received = match.groups() - packet_lost = int(sent) - int(received) - - return packet_lost + cmd = u"sw_interface_ip6nd_ra_config" + args = dict( + sw_if_index=InterfaceUtil.get_interface_index(node, interface), + suppress=1 + ) + err_msg = f"Failed to suppress ICMPv6 router advertisement message " \ + f"on interface {interface}" - @staticmethod - def ipv6_ping_port(nodes_ip, src_node, dst_node, port, cnt=3, - size=56, timeout=1): - """Send IPv6 ping to the node port. - - :param nodes_ip: Nodes IPv6 addresses. - :param src_node: Node where ping run. - :param dst_node: Destination node. - :param port: Port on the destination node. - :param cnt: Number of echo requests. (Optional) - :param size: Number of the data bytes. (Optional) - :param timeout: Time to wait for a response, in seconds. (Optional) - :type nodes_ip: dict - :type src_node: dict - :type dst_node: dict - :type port: str - :type cnt: int - :type size: int - :type timeout: int - :return: Number of lost packets. - :rtype: int - """ - dst_ip = IPv6Util.get_node_port_ipv6_address(dst_node, port, nodes_ip) - return IPv6Util.ipv6_ping(src_node, dst_ip, cnt, size, timeout) + with PapiSocketExecutor(node) as papi_exec: + papi_exec.add(cmd, **args).get_reply(err_msg) @staticmethod - def get_node_port_ipv6_address(node, iface_key, nodes_addr): - """Return IPv6 address of the node port. + def vpp_ra_send_after_interval(node, interface, interval=2): + """Setup vpp router advertisement(RA) in such way it sends RA packet + after every interval value. - :param node: Node in the topology. - :param iface_key: Interface key of the node. - :param nodes_addr: Nodes IPv6 addresses. + :param node: VPP node. + :param interface: Interface name. + :param interval: Interval in seconds for RA resend. :type node: dict - :type iface_key: str - :type nodes_addr: dict - :return: IPv6 address string. - :rtype: str + :type interface: str + :type interval: int """ - interface = Topology.get_interface_name(node, iface_key) - for net in nodes_addr.values(): - for port in net['ports'].values(): - host = port.get('node') - dev = port.get('if') - if host == node['host'] and dev == interface: - ip = port.get('addr') - if ip is not None: - return ip - else: - raise Exception( - 'Node {n} port {p} IPv6 address is not set'.format( - n=node['host'], p=interface)) + cmd = u"sw_interface_ip6nd_ra_config" + args = dict( + sw_if_index=InterfaceUtil.get_interface_index(node, interface), + initial_interval=int(interval) + ) + err_msg = f"Failed to set router advertisement interval " \ + f"on interface {interface}" - raise Exception('Node {n} port {p} IPv6 address not found.'.format( - n=node['host'], p=interface)) + with PapiSocketExecutor(node) as papi_exec: + papi_exec.add(cmd, **args).get_reply(err_msg) @staticmethod - def add_ip_neighbor(node, interface, ip_address, mac_address): - """Add IP neighbor. + def vpp_all_ra_suppress_link_layer(nodes): + """Suppress ICMPv6 router advertisement message for link scope address + on all VPP nodes in the topology. - :param node: VPP node to add ip neighbor. - :param interface: Interface name or sw_if_index. - :param ip_address: IP address. - :param mac_address: MAC address. - :type node: dict - :type interface: str or int - :type ip_address: str - :type mac_address: str + :param nodes: Nodes of the test topology. + :type nodes: dict """ - if isinstance(interface, basestring): - sw_if_index = Topology.get_interface_sw_index(node, interface) - else: - sw_if_index = interface - - with VatTerminal(node) as vat: - vat.vat_terminal_exec_cmd_from_template("add_ip_neighbor.vat", - sw_if_index=sw_if_index, - ip_address=ip_address, - mac_address=mac_address) + for node in nodes.values(): + if node[u"type"] == NodeType.TG: + continue + for port_k in node[u"interfaces"].keys(): + ip6_addr_list = IPUtil.vpp_get_interface_ip_addresses( + node, port_k, u"ipv6" + ) + if ip6_addr_list: + IPv6Util.vpp_ra_suppress_link_layer(node, port_k)