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."""
16 from robot.api import logger
17 from ipaddress import IPv6Network
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 :param networks: List of the available IPv6 networks.
31 def __init__(self, networks):
32 self._networks = list()
33 for network in networks:
34 net = IPv6Network(unicode(network))
35 self._networks.append(net)
36 num = len(self._networks)
38 raise Exception('No IPv6 networks')
40 def next_network(self):
41 """Get the next element of the iterator.
43 :return: IPv6 network.
44 :rtype: IPv6Network object
45 :raises: StopIteration if there is no more elements.
47 if len(self._networks):
48 return self._networks.pop()
53 class IPv6Setup(object):
54 """IPv6 setup in topology."""
59 def nodes_set_ipv6_addresses(self, nodes, nodes_addr):
60 """Set IPv6 addresses on all VPP nodes in topology.
62 :param nodes: Nodes of the test topology.
63 :param nodes_addr: Available nodes IPv6 addresses.
65 :type nodes_addr: dict
66 :returns: Affected interfaces as list of (node, interface) tuples.
71 for net in nodes_addr.values():
72 for port in net['ports'].values():
73 host = port.get('node')
77 node = topo.get_node_by_hostname(nodes, host)
80 if node['type'] == NodeType.DUT:
81 port_key = topo.get_interface_by_name(node, port['if'])
82 self.vpp_set_if_ipv6_addr(node, port_key, port['addr'],
85 interfaces.append((node, port['if']))
88 def nodes_clear_ipv6_addresses(self, nodes, nodes_addr):
89 """Remove IPv6 addresses from all VPP nodes in topology.
91 :param nodes: Nodes of the test topology.
92 :param nodes_addr: Available nodes IPv6 addresses.
94 :type nodes_addr: dict
96 for net in nodes_addr.values():
97 for port in net['ports'].values():
98 host = port.get('node')
102 node = topo.get_node_by_hostname(nodes, host)
105 if node['type'] == NodeType.DUT:
106 self.vpp_del_if_ipv6_addr(node, port['if'], port['addr'],
110 def linux_set_if_ipv6_addr(node, interface, addr, prefix):
111 """Set IPv6 address on linux host.
113 :param node: Linux node.
114 :param interface: Node interface.
115 :param addr: IPv6 address.
116 :param prefix: IPv6 address prefix.
125 cmd = "ifconfig {dev} inet6 add {ip}/{p} up".format(dev=interface,
127 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
128 if int(ret_code) != 0:
129 raise Exception('TG ifconfig failed')
132 def linux_del_if_ipv6_addr(node, interface, addr, prefix):
133 """Delete IPv6 address on linux host.
135 :param node: Linux node.
136 :param interface: Node interface.
137 :param addr: IPv6 address.
138 :param prefix: IPv6 address prefix.
147 cmd = "ifconfig {dev} inet6 del {ip}/{p}".format(dev=interface,
150 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
151 if int(ret_code) != 0:
152 raise Exception('TG ifconfig failed')
154 cmd = "ifconfig {dev} down".format(dev=interface)
155 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
156 if int(ret_code) != 0:
157 raise Exception('TG ifconfig failed')
160 def vpp_set_if_ipv6_addr(node, iface_key, addr, prefix):
161 """Set IPv6 address on VPP.
163 :param node: VPP node.
164 :param iface_key: Node interface key.
165 :param addr: IPv6 address.
166 :param prefix: IPv6 address prefix.
172 sw_if_index = Topology.get_interface_sw_index(node, iface_key)
173 with VatTerminal(node) as vat:
174 vat.vat_terminal_exec_cmd_from_template('add_ip_address.vat',
175 sw_if_index=sw_if_index,
177 prefix_length=prefix)
178 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
179 sw_if_index=sw_if_index,
184 cmd_input = 'exec show int'
185 (ret_code, stdout, stderr) = ssh.exec_command_sudo(
186 Constants.VAT_BIN_NAME, cmd_input)
187 logger.debug('ret: {0}'.format(ret_code))
188 logger.debug('stdout: {0}'.format(stdout))
189 logger.debug('stderr: {0}'.format(stderr))
192 def vpp_del_if_ipv6_addr(node, interface, addr, prefix):
193 """Delete IPv6 address on VPP.
195 :param node: VPP node.
196 :param interface: Node interface.
197 :param addr: IPv6 address.
198 :param prefix: IPv6 address prefix.
204 sw_if_index = Topology.get_interface_sw_index(node, interface)
205 with VatTerminal(node) as vat:
206 vat.vat_terminal_exec_cmd_from_template('del_ip_address.vat',
207 sw_if_index=sw_if_index,
209 prefix_length=prefix)
210 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
211 sw_if_index=sw_if_index,
215 def vpp_ra_suppress_link_layer(node, interface):
216 """Suppress ICMPv6 router advertisement message for link scope address.
218 :param node: VPP node.
219 :param interface: Interface name.
223 sw_if_index = Topology.get_interface_sw_index(node, interface)
224 VatExecutor.cmd_from_template(node,
225 'sw_interface_ip6nd_ra_config.vat',
226 sw_if_id=sw_if_index,
230 def vpp_ra_send_after_interval(node, interface, interval=2):
231 """Setup vpp router advertisement(RA) in such way it sends RA packet
232 after and every interval value.
234 :param node: VPP node.
235 :param interface: Interface name.
236 :param interval: Interval for RA resend
241 sw_if_index = Topology.get_interface_sw_index(node, interface)
242 VatExecutor.cmd_from_template(node,
243 'sw_interface_ip6nd_ra_config.vat',
244 sw_if_id=sw_if_index,
245 param='interval {0}'.format(interval))
247 def vpp_all_ra_suppress_link_layer(self, nodes):
248 """Suppress ICMPv6 router advertisement message for link scope address
249 on all VPP nodes in the topology.
251 :param nodes: Nodes of the test topology.
254 for node in nodes.values():
255 if node['type'] == NodeType.TG:
257 for port_k in node['interfaces'].keys():
258 self.vpp_ra_suppress_link_layer(node, port_k)
261 def get_link_address(link, nodes_addr):
262 """Get link IPv6 address.
264 :param link: Link name.
265 :param nodes_addr: Available nodes IPv6 addresses.
267 :type nodes_addr: dict
268 :returns: Link IPv6 address.
271 net = nodes_addr.get(link)
273 raise ValueError('Link "{0}" address not found'.format(link))
274 return net.get('net_addr')
277 def get_link_prefix(link, nodes_addr):
278 """Get link IPv6 address prefix.
280 :param link: Link name.
281 :param nodes_addr: Available nodes IPv6 addresses.
283 :type nodes_addr: dict
284 :returns: Link IPv6 address prefix.
287 net = nodes_addr.get(link)
289 raise ValueError('Link "{0}" address not found'.format(link))
290 return net.get('prefix')