1 # Copyright (c) 2016 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
14 """Library to set up IPv6 in topology."""
17 from ipaddress import IPv6Network
18 from topology import NodeType
19 from topology import Topology
20 from constants import Constants
21 from robot.api import logger
24 class IPv6Networks(object):
25 """IPv6 network iterator.
27 :param networks: List of the available IPv6 networks.
30 def __init__(self, networks):
31 self._networks = list()
32 for network in networks:
33 net = IPv6Network(unicode(network))
34 self._networks.append(net)
35 num = len(self._networks)
37 raise Exception('No IPv6 networks')
39 def next_network(self):
40 """Get the next elemnt of the iterator.
42 :return: IPv6 network.
43 :rtype: IPv6Network object
44 :raises: StopIteration if there is no more elements.
46 if len(self._networks):
47 return self._networks.pop()
52 class IPv6Setup(object):
53 """IPv6 setup in topology."""
58 def nodes_setup_ipv6_addresses(self, nodes, nodes_addr):
59 """Setup IPv6 addresses on all VPP nodes in topology.
61 :param nodes: Nodes of the test topology.
62 :param nodes_addr: Available nodes IPv6 adresses.
64 :type nodes_addr: dict
66 for net in nodes_addr.values():
67 for port in net['ports'].values():
68 host = port.get('node')
72 node = topo.get_node_by_hostname(nodes, host)
75 if node['type'] == NodeType.DUT:
76 self.vpp_set_if_ipv6_addr(node, port['if'], port['addr'],
79 def nodes_clear_ipv6_addresses(self, nodes, nodes_addr):
80 """Remove IPv6 addresses from all VPP nodes in topology.
82 :param nodes: Nodes of the test topology.
83 :param nodes_addr: Available nodes IPv6 adresses.
85 :type nodes_addr: dict
87 for net in nodes_addr.values():
88 for port in net['ports'].values():
89 host = port.get('node')
93 node = topo.get_node_by_hostname(nodes, host)
96 if node['type'] == NodeType.DUT:
97 self.vpp_del_if_ipv6_addr(node, port['if'], port['addr'],
101 def linux_set_if_ipv6_addr(node, interface, addr, prefix):
102 """Set IPv6 address on linux host.
104 :param node: Linux node.
105 :param interface: Node interface.
106 :param addr: IPv6 address.
107 :param prefix: IPv6 address prefix.
116 cmd = "ifconfig {dev} inet6 add {ip}/{p} up".format(dev=interface,
118 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
119 if int(ret_code) != 0:
120 raise Exception('TG ifconfig failed')
123 def linux_del_if_ipv6_addr(node, interface, addr, prefix):
124 """Delete IPv6 address on linux host.
126 :param node: Linux node.
127 :param interface: Node interface.
128 :param addr: IPv6 address.
129 :param prefix: IPv6 address prefix.
138 cmd = "ifconfig {dev} inet6 del {ip}/{p}".format(dev=interface,
141 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
142 if int(ret_code) != 0:
143 raise Exception('TG ifconfig failed')
145 cmd = "ifconfig {dev} down".format(dev=interface)
146 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
147 if int(ret_code) != 0:
148 raise Exception('TG ifconfig failed')
151 def vpp_set_if_ipv6_addr(node, interface, addr, prefix):
152 """Set IPv6 address on VPP.
154 :param node: VPP node.
155 :param interface: Node interface.
156 :param addr: IPv6 address.
157 :param prefix: IPv6 address prefix.
166 cmd = '{c}'.format(c=Constants.VAT_BIN_NAME)
167 cmd_input = 'sw_interface_add_del_address {dev} {ip}/{p}'.format(
168 dev=interface, ip=addr, p=prefix)
169 (ret_code, _, _) = ssh.exec_command_sudo(cmd, cmd_input)
170 if int(ret_code) != 0:
171 raise Exception('VPP sw_interface_add_del_address failed on {h}'
172 .format(h=node['host']))
174 cmd_input = 'sw_interface_set_flags {dev} admin-up'.format(
176 (ret_code, _, _) = ssh.exec_command_sudo(cmd, cmd_input)
177 if int(ret_code) != 0:
178 raise Exception('VPP sw_interface_set_flags failed on {h}'.format(
181 cmd_input = 'exec show int'
182 (ret_code, stdout, stderr) = ssh.exec_command_sudo(cmd, cmd_input)
183 logger.debug('ret: {0}'.format(ret_code))
184 logger.debug('stdout: {0}'.format(stdout))
185 logger.debug('stderr: {0}'.format(stderr))
188 def vpp_del_if_ipv6_addr(node, interface, addr, prefix):
189 """Delete IPv6 address on VPP.
191 :param node: VPP node.
192 :param interface: Node interface.
193 :param addr: IPv6 address.
194 :param prefix: IPv6 address prefix.
203 cmd = '{c}'.format(c=Constants.VAT_BIN_NAME)
204 cmd_input = 'sw_interface_add_del_address {dev} {ip}/{p} del'.format(
205 dev=interface, ip=addr, p=prefix)
206 (ret_code, _, _) = ssh.exec_command_sudo(cmd, cmd_input)
207 if int(ret_code) != 0:
209 'sw_interface_add_del_address failed on {h}'.
210 format(h=node['host']))
212 cmd_input = 'sw_interface_set_flags {dev} admin-down'.format(
214 (ret_code, _, _) = ssh.exec_command_sudo(cmd, cmd_input)
215 if int(ret_code) != 0:
216 raise Exception('VPP sw_interface_set_flags failed on {h}'.format(
220 def vpp_ra_supress_link_layer(node, interface):
221 """Supress ICMPv6 router advertisement message for link scope address
223 :param node: VPP node.
224 :param interface: Interface name.
231 cmd = '{c}'.format(c=Constants.VAT_BIN_NAME)
232 cmd_input = 'exec ip6 nd {0} ra-surpress-link-layer'.format(
234 (ret_code, _, _) = ssh.exec_command_sudo(cmd, cmd_input)
235 if int(ret_code) != 0:
236 raise Exception("'{0}' failed on {1}".format(cmd_input,
239 def vpp_all_ra_supress_link_layer(self, nodes):
240 """Supress ICMPv6 router advertisement message for link scope address
241 on all VPP nodes in the topology
243 :param nodes: Nodes of the test topology.
246 for node in nodes.values():
247 if node['type'] == NodeType.TG:
249 for port_k, port_v in node['interfaces'].items():
252 if_name = port_v.get('name')
255 self.vpp_ra_supress_link_layer(node, if_name)
258 def get_link_address(link, nodes_addr):
259 """Get link IPv6 address.
261 :param link: Link name.
262 :param nodes_addr: Available nodes IPv6 adresses.
264 :type nodes_addr: dict
265 :return: Link IPv6 address.
268 net = nodes_addr.get(link)
270 raise ValueError('Link "{0}" address not found'.format(link))
271 return net.get('net_addr')
274 def get_link_prefix(link, nodes_addr):
275 """Get link IPv6 address prefix.
277 :param link: Link name.
278 :param nodes_addr: Available nodes IPv6 adresses.
280 :type nodes_addr: dict
281 :return: Link IPv6 address prefix.
284 net = nodes_addr.get(link)
286 raise ValueError('Link "{0}" address not found'.format(link))
287 return net.get('prefix')