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 :return: 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.
173 sw_if_index = Topology.get_interface_sw_index(node, iface_key)
174 with VatTerminal(node) as vat:
175 vat.vat_terminal_exec_cmd_from_template('add_ip_address.vat',
176 sw_if_index=sw_if_index,
178 prefix_length=prefix)
179 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
180 sw_if_index=sw_if_index,
185 cmd_input = 'exec show int'
186 (ret_code, stdout, stderr) = ssh.exec_command_sudo(
187 Constants.VAT_BIN_NAME, cmd_input)
188 logger.debug('ret: {0}'.format(ret_code))
189 logger.debug('stdout: {0}'.format(stdout))
190 logger.debug('stderr: {0}'.format(stderr))
193 def vpp_del_if_ipv6_addr(node, interface, addr, prefix):
194 """Delete IPv6 address on VPP.
196 :param node: VPP node.
197 :param interface: Node interface.
198 :param addr: IPv6 address.
199 :param prefix: IPv6 address prefix.
205 sw_if_index = Topology.get_interface_sw_index(node, interface)
206 with VatTerminal(node) as vat:
207 vat.vat_terminal_exec_cmd_from_template('del_ip_address.vat',
208 sw_if_index=sw_if_index,
210 prefix_length=prefix)
211 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
212 sw_if_index=sw_if_index,
217 def vpp_set_ipv6_neighbor(node, interface, ipaddr, macaddr):
218 """Set IPv6 neighbor.
220 :param node: VPP node.
221 :param interface: Interface name.
222 :param ip_address: IPv6 address.
223 :param mac_address: mac address.
226 :type ip_address: str
227 :type mac_address: str
229 sw_if_index = Topology.get_interface_sw_index(node, interface)
230 VatExecutor.cmd_from_template(node,
231 'add_ip_neighbor.vat',
232 sw_if_index=sw_if_index,
237 def vpp_ra_suppress_link_layer(node, interface):
238 """Suppress ICMPv6 router advertisement message for link scope address.
240 :param node: VPP node.
241 :param interface: Interface name.
245 sw_if_index = Topology.get_interface_sw_index(node, interface)
246 VatExecutor.cmd_from_template(node,
247 'sw_interface_ip6nd_ra_config.vat',
248 sw_if_id=sw_if_index,
252 def vpp_ra_send_after_interval(node, interface, interval=2):
253 """Setup vpp router advertisement(RA) in such way it sends RA packet
254 after and every interval value.
256 :param node: VPP node.
257 :param interface: Interface name.
258 :param interval: Interval for RA resend
263 sw_if_index = Topology.get_interface_sw_index(node, interface)
264 VatExecutor.cmd_from_template(node,
265 'sw_interface_ip6nd_ra_config.vat',
266 sw_if_id=sw_if_index,
267 param='interval {0}'.format(interval))
269 def vpp_all_ra_suppress_link_layer(self, nodes):
270 """Suppress ICMPv6 router advertisement message for link scope address
271 on all VPP nodes in the topology.
273 :param nodes: Nodes of the test topology.
276 for node in nodes.values():
277 if node['type'] == NodeType.TG:
279 for port_k in node['interfaces'].keys():
280 self.vpp_ra_suppress_link_layer(node, port_k)
283 def get_link_address(link, nodes_addr):
284 """Get link IPv6 address.
286 :param link: Link name.
287 :param nodes_addr: Available nodes IPv6 addresses.
289 :type nodes_addr: dict
290 :return: Link IPv6 address.
293 net = nodes_addr.get(link)
295 raise ValueError('Link "{0}" address not found'.format(link))
296 return net.get('net_addr')
299 def get_link_prefix(link, nodes_addr):
300 """Get link IPv6 address prefix.
302 :param link: Link name.
303 :param nodes_addr: Available nodes IPv6 addresses.
305 :type nodes_addr: dict
306 :return: Link IPv6 address prefix.
309 net = nodes_addr.get(link)
311 raise ValueError('Link "{0}" address not found'.format(link))
312 return net.get('prefix')