1 # Copyright (c) 2018 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."""
16 from ipaddress import IPv6Network
17 from robot.api import logger
19 from resources.libraries.python.ssh import SSH
20 from resources.libraries.python.topology import NodeType, Topology
21 from resources.libraries.python.Constants import Constants
22 from resources.libraries.python.VatExecutor import VatTerminal, VatExecutor
25 class IPv6Networks(object):
26 """IPv6 network iterator.
28 TODO: Conform to https://docs.python.org/2/library/stdtypes.html#typeiter
31 def __init__(self, networks):
32 """Initialize internal list of valid networks.
34 :param networks: List of the available IPv6 networks.
36 :raise RuntimeError: If no networks were added.
39 for network in networks:
40 net = IPv6Network(unicode(network))
41 self._networks.append(net)
42 if not self._networks:
43 raise RuntimeError('No IPv6 networks')
45 def next_network(self):
46 """Get the next element of the iterator.
48 :returns: IPv6 network.
49 :rtype: IPv6Network object
50 :raises StopIteration: If there is no more elements.
53 return self._networks.pop()
58 class IPv6Setup(object):
59 """IPv6 setup in topology."""
64 def nodes_set_ipv6_addresses(self, nodes, nodes_addr):
65 """Set IPv6 addresses on all VPP nodes in topology.
67 :param nodes: Nodes of the test topology.
68 :param nodes_addr: Available nodes IPv6 addresses.
70 :type nodes_addr: dict
71 :returns: Affected interfaces as list of (node, interface) tuples.
76 for net in nodes_addr.values():
77 for port in net['ports'].values():
78 host = port.get('node')
82 node = topo.get_node_by_hostname(nodes, host)
85 if node['type'] == NodeType.DUT:
86 port_key = topo.get_interface_by_name(node, port['if'])
87 self.vpp_set_if_ipv6_addr(node, port_key, port['addr'],
90 interfaces.append((node, port['if']))
93 def nodes_clear_ipv6_addresses(self, nodes, nodes_addr):
94 """Remove IPv6 addresses from all VPP nodes in topology.
96 :param nodes: Nodes of the test topology.
97 :param nodes_addr: Available nodes IPv6 addresses.
99 :type nodes_addr: dict
101 for net in nodes_addr.values():
102 for port in net['ports'].values():
103 host = port.get('node')
107 node = topo.get_node_by_hostname(nodes, host)
110 if node['type'] == NodeType.DUT:
111 self.vpp_del_if_ipv6_addr(node, port['if'], port['addr'],
115 def linux_set_if_ipv6_addr(node, interface, addr, prefix):
116 """Set IPv6 address on linux host.
118 :param node: Linux node.
119 :param interface: Node interface.
120 :param addr: IPv6 address.
121 :param prefix: IPv6 address prefix.
130 cmd = "ifconfig {dev} inet6 add {ip}/{p} up".format(dev=interface,
132 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
133 if int(ret_code) != 0:
134 raise Exception('TG ifconfig failed')
137 def linux_del_if_ipv6_addr(node, interface, addr, prefix):
138 """Delete IPv6 address on linux host.
140 :param node: Linux node.
141 :param interface: Node interface.
142 :param addr: IPv6 address.
143 :param prefix: IPv6 address prefix.
152 cmd = "ifconfig {dev} inet6 del {ip}/{p}".format(dev=interface,
155 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
156 if int(ret_code) != 0:
157 raise Exception('TG ifconfig failed')
159 cmd = "ifconfig {dev} down".format(dev=interface)
160 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
161 if int(ret_code) != 0:
162 raise Exception('TG ifconfig failed')
165 def vpp_set_if_ipv6_addr(node, iface_key, addr, prefix):
166 """Set IPv6 address on VPP.
168 :param node: VPP node.
169 :param iface_key: Node interface key.
170 :param addr: IPv6 address.
171 :param prefix: IPv6 address prefix.
177 sw_if_index = Topology.get_interface_sw_index(node, iface_key)
178 with VatTerminal(node) as vat:
179 vat.vat_terminal_exec_cmd_from_template('add_ip_address.vat',
180 sw_if_index=sw_if_index,
182 prefix_length=prefix)
183 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
184 sw_if_index=sw_if_index,
189 cmd_input = 'exec show int'
190 (ret_code, stdout, stderr) = ssh.exec_command_sudo(
191 Constants.VAT_BIN_NAME, cmd_input)
192 logger.debug('ret: {0}'.format(ret_code))
193 logger.debug('stdout: {0}'.format(stdout))
194 logger.debug('stderr: {0}'.format(stderr))
197 def vpp_del_if_ipv6_addr(node, interface, addr, prefix):
198 """Delete IPv6 address on VPP.
200 :param node: VPP node.
201 :param interface: Node interface.
202 :param addr: IPv6 address.
203 :param prefix: IPv6 address prefix.
209 sw_if_index = Topology.get_interface_sw_index(node, interface)
210 with VatTerminal(node) as vat:
211 vat.vat_terminal_exec_cmd_from_template('del_ip_address.vat',
212 sw_if_index=sw_if_index,
214 prefix_length=prefix)
215 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
216 sw_if_index=sw_if_index,
220 def vpp_ra_suppress_link_layer(node, interface):
221 """Suppress ICMPv6 router advertisement message for link scope address.
223 :param node: VPP node.
224 :param interface: Interface name.
228 sw_if_index = Topology.get_interface_sw_index(node, interface)
230 VatExecutor.cmd_from_template(node,
231 'sw_interface_ip6nd_ra_config.vat',
232 sw_if_id=sw_if_index,
236 def vpp_ra_send_after_interval(node, interface, interval=2):
237 """Setup vpp router advertisement(RA) in such way it sends RA packet
238 after and every interval value.
240 :param node: VPP node.
241 :param interface: Interface name.
242 :param interval: Interval for RA resend
247 sw_if_index = Topology.get_interface_sw_index(node, interface)
249 VatExecutor.cmd_from_template(node,
250 'sw_interface_ip6nd_ra_config.vat',
251 sw_if_id=sw_if_index,
252 param='interval {0}'.format(interval))
254 def vpp_all_ra_suppress_link_layer(self, nodes):
255 """Suppress ICMPv6 router advertisement message for link scope address
256 on all VPP nodes in the topology.
258 :param nodes: Nodes of the test topology.
261 for node in nodes.values():
262 if node['type'] == NodeType.TG:
264 for port_k in node['interfaces'].keys():
265 self.vpp_ra_suppress_link_layer(node, port_k)
268 def get_link_address(link, nodes_addr):
269 """Get link IPv6 address.
271 :param link: Link name.
272 :param nodes_addr: Available nodes IPv6 addresses.
274 :type nodes_addr: dict
275 :returns: Link IPv6 address.
278 net = nodes_addr.get(link)
280 raise ValueError('Link "{0}" address not found'.format(link))
281 return net.get('net_addr')
284 def get_link_prefix(link, nodes_addr):
285 """Get link IPv6 address prefix.
287 :param link: Link name.
288 :param nodes_addr: Available nodes IPv6 addresses.
290 :type nodes_addr: dict
291 :returns: Link IPv6 address prefix.
294 net = nodes_addr.get(link)
296 raise ValueError('Link "{0}" address not found'.format(link))
297 return net.get('prefix')