X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FIPv6Setup.py;h=72aeb305a2462dd1f26d8d8ed70fce7364b31c97;hp=598905735a5ad0894a8a02af6d165b1cb30d7377;hb=cc0c2870a18fb74a56410eca2d1870bddc945397;hpb=8120dcdc84da7ff1dee097240bc1ecf18914397c diff --git a/resources/libraries/python/IPv6Setup.py b/resources/libraries/python/IPv6Setup.py index 598905735a..72aeb305a2 100644 --- a/resources/libraries/python/IPv6Setup.py +++ b/resources/libraries/python/IPv6Setup.py @@ -1,4 +1,4 @@ -# Copyright (c) 2016 Cisco and/or its affiliates. +# Copyright (c) 2018 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,37 +13,43 @@ """Library to set up IPv6 in topology.""" -from ssh import SSH from ipaddress import IPv6Network -from topology import NodeType, Topology -from constants import Constants -from VatExecutor import VatTerminal, VatExecutor from robot.api import logger +from resources.libraries.python.ssh import SSH +from resources.libraries.python.topology import NodeType, Topology +from resources.libraries.python.Constants import Constants +from resources.libraries.python.VatExecutor import VatTerminal, VatExecutor + class IPv6Networks(object): """IPv6 network iterator. - :param networks: List of the available IPv6 networks. - :type networks: list + TODO: Conform to https://docs.python.org/2/library/stdtypes.html#typeiter """ + def __init__(self, networks): - self._networks = list() + """Initialize internal list of valid networks. + + :param networks: List of the available IPv6 networks. + :type networks: list + :raise RuntimeError: If no networks were added. + """ + self._networks = [] for network in networks: net = IPv6Network(unicode(network)) self._networks.append(net) - num = len(self._networks) - if num == 0: - raise Exception('No IPv6 networks') + if not self._networks: + raise RuntimeError('No IPv6 networks') def next_network(self): - """Get the next elemnt of the iterator. + """Get the next element of the iterator. - :return: IPv6 network. - :rtype: IPv6Network object - :raises: StopIteration if there is no more elements. + :returns: IPv6 network. + :rtype: IPv6Network object + :raises StopIteration: If there is no more elements. """ - if len(self._networks): + if self._networks: return self._networks.pop() else: raise StopIteration() @@ -58,14 +64,13 @@ class IPv6Setup(object): def nodes_set_ipv6_addresses(self, nodes, nodes_addr): """Set IPv6 addresses on all VPP nodes in topology. - :param nodes: Nodes of the test topology. - :param nodes_addr: Available nodes IPv6 adresses. - :type nodes: dict - :type nodes_addr: dict - :return: affected interfaces as list of (node, interface) tuples - :rtype: list + :param nodes: Nodes of the test topology. + :param nodes_addr: Available nodes IPv6 addresses. + :type nodes: dict + :type nodes_addr: dict + :returns: Affected interfaces as list of (node, interface) tuples. + :rtype: list """ - interfaces = [] for net in nodes_addr.values(): @@ -78,20 +83,21 @@ class IPv6Setup(object): if node is None: continue if node['type'] == NodeType.DUT: - self.vpp_set_if_ipv6_addr(node, port['if'], port['addr'], + port_key = topo.get_interface_by_name(node, port['if']) + self.vpp_set_if_ipv6_addr(node, port_key, port['addr'], net['prefix']) interfaces.append((node, port['if'])) - return interfaces + def nodes_clear_ipv6_addresses(self, nodes, nodes_addr): """Remove IPv6 addresses from all VPP nodes in topology. - :param nodes: Nodes of the test topology. - :param nodes_addr: Available nodes IPv6 adresses. - :type nodes: dict - :type nodes_addr: dict - """ + :param nodes: Nodes of the test topology. + :param nodes_addr: Available nodes IPv6 addresses. + :type nodes: dict + :type nodes_addr: dict + """ for net in nodes_addr.values(): for port in net['ports'].values(): host = port.get('node') @@ -109,14 +115,14 @@ class IPv6Setup(object): def linux_set_if_ipv6_addr(node, interface, addr, prefix): """Set IPv6 address on linux host. - :param node: Linux node. - :param interface: Node interface. - :param addr: IPv6 address. - :param prefix: IPv6 address prefix. - :type node: dict - :type interface: str - :type addr: str - :type prefix: str + :param node: Linux node. + :param interface: Node interface. + :param addr: IPv6 address. + :param prefix: IPv6 address prefix. + :type node: dict + :type interface: str + :type addr: str + :type prefix: str """ ssh = SSH() ssh.connect(node) @@ -131,14 +137,14 @@ class IPv6Setup(object): def linux_del_if_ipv6_addr(node, interface, addr, prefix): """Delete IPv6 address on linux host. - :param node: Linux node. - :param interface: Node interface. - :param addr: IPv6 address. - :param prefix: IPv6 address prefix. - :type node: dict - :type interface: str - :type addr: str - :type prefix: str + :param node: Linux node. + :param interface: Node interface. + :param addr: IPv6 address. + :param prefix: IPv6 address prefix. + :type node: dict + :type interface: str + :type addr: str + :type prefix: str """ ssh = SSH() ssh.connect(node) @@ -156,19 +162,19 @@ class IPv6Setup(object): raise Exception('TG ifconfig failed') @staticmethod - def vpp_set_if_ipv6_addr(node, interface, addr, prefix): + def vpp_set_if_ipv6_addr(node, iface_key, addr, prefix): """Set IPv6 address on VPP. - :param node: VPP node. - :param interface: Node interface. - :param addr: IPv6 address. - :param prefix: IPv6 address prefix. - :type node: dict - :type interface: str - :type addr: str - :type prefix: str + :param node: VPP node. + :param iface_key: Node interface key. + :param addr: IPv6 address. + :param prefix: IPv6 address prefix. + :type node: dict + :type iface_key: str + :type addr: str + :type prefix: str """ - sw_if_index = Topology.get_interface_sw_index(node, interface) + sw_if_index = Topology.get_interface_sw_index(node, iface_key) with VatTerminal(node) as vat: vat.vat_terminal_exec_cmd_from_template('add_ip_address.vat', sw_if_index=sw_if_index, @@ -191,14 +197,14 @@ class IPv6Setup(object): def vpp_del_if_ipv6_addr(node, interface, addr, prefix): """Delete IPv6 address on VPP. - :param node: VPP node. - :param interface: Node interface. - :param addr: IPv6 address. - :param prefix: IPv6 address prefix. - :type node: dict - :type interface: str - :type addr: str - :type prefix: str + :param node: VPP node. + :param interface: Node interface. + :param addr: IPv6 address. + :param prefix: IPv6 address prefix. + :type node: dict + :type interface: str + :type addr: str + :type prefix: str """ sw_if_index = Topology.get_interface_sw_index(node, interface) with VatTerminal(node) as vat: @@ -211,47 +217,62 @@ class IPv6Setup(object): state='admin-down') @staticmethod - def vpp_ra_supress_link_layer(node, interface): - """Supress ICMPv6 router advertisement message for link scope address + def vpp_ra_suppress_link_layer(node, interface): + """Suppress ICMPv6 router advertisement message for link scope address. - :param node: VPP node. - :param interface: Interface name. - :type node: dict - :type interface: str + :param node: VPP node. + :param interface: Interface name. + :type node: dict + :type interface: str """ sw_if_index = Topology.get_interface_sw_index(node, interface) - VatExecutor.cmd_from_template(node, - 'sw_interface_ip6nd_ra_config.vat', - sw_if_id=sw_if_index, - param='surpress') + if sw_if_index: + VatExecutor.cmd_from_template(node, + 'sw_interface_ip6nd_ra_config.vat', + sw_if_id=sw_if_index, + param='surpress') - def vpp_all_ra_supress_link_layer(self, nodes): - """Supress ICMPv6 router advertisement message for link scope address - on all VPP nodes in the topology - - :param nodes: Nodes of the test topology. - :type nodes: dict + @staticmethod + def vpp_ra_send_after_interval(node, interface, interval=2): + """Setup vpp router advertisement(RA) in such way it sends RA packet + after and every interval value. + + :param node: VPP node. + :param interface: Interface name. + :param interval: Interval for RA resend + :type node: dict + :type interface: str + :type interval: int + """ + sw_if_index = Topology.get_interface_sw_index(node, interface) + if sw_if_index: + VatExecutor.cmd_from_template(node, + 'sw_interface_ip6nd_ra_config.vat', + sw_if_id=sw_if_index, + param='interval {0}'.format(interval)) + + def vpp_all_ra_suppress_link_layer(self, nodes): + """Suppress ICMPv6 router advertisement message for link scope address + on all VPP nodes in the topology. + + :param nodes: Nodes of the test topology. + :type nodes: dict """ for node in nodes.values(): if node['type'] == NodeType.TG: continue - for port_k, port_v in node['interfaces'].items(): - if port_k == 'mgmt': - continue - if_name = port_v.get('name') - if if_name is None: - continue - self.vpp_ra_supress_link_layer(node, if_name) + for port_k in node['interfaces'].keys(): + self.vpp_ra_suppress_link_layer(node, port_k) @staticmethod def get_link_address(link, nodes_addr): """Get link IPv6 address. :param link: Link name. - :param nodes_addr: Available nodes IPv6 adresses. + :param nodes_addr: Available nodes IPv6 addresses. :type link: str :type nodes_addr: dict - :return: Link IPv6 address. + :returns: Link IPv6 address. :rtype: str """ net = nodes_addr.get(link) @@ -264,10 +285,10 @@ class IPv6Setup(object): """Get link IPv6 address prefix. :param link: Link name. - :param nodes_addr: Available nodes IPv6 adresses. + :param nodes_addr: Available nodes IPv6 addresses. :type link: str :type nodes_addr: dict - :return: Link IPv6 address prefix. + :returns: Link IPv6 address prefix. :rtype: int """ net = nodes_addr.get(link)