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 self.vpp_set_if_ipv6_addr(node, port['if'], port['addr'],
84 interfaces.append((node, port['if']))
87 def nodes_clear_ipv6_addresses(self, nodes, nodes_addr):
88 """Remove IPv6 addresses from all VPP nodes in topology.
90 :param nodes: Nodes of the test topology.
91 :param nodes_addr: Available nodes IPv6 addresses.
93 :type nodes_addr: dict
95 for net in nodes_addr.values():
96 for port in net['ports'].values():
97 host = port.get('node')
101 node = topo.get_node_by_hostname(nodes, host)
104 if node['type'] == NodeType.DUT:
105 self.vpp_del_if_ipv6_addr(node, port['if'], port['addr'],
109 def linux_set_if_ipv6_addr(node, interface, addr, prefix):
110 """Set IPv6 address on linux host.
112 :param node: Linux node.
113 :param interface: Node interface.
114 :param addr: IPv6 address.
115 :param prefix: IPv6 address prefix.
124 cmd = "ifconfig {dev} inet6 add {ip}/{p} up".format(dev=interface,
126 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
127 if int(ret_code) != 0:
128 raise Exception('TG ifconfig failed')
131 def linux_del_if_ipv6_addr(node, interface, addr, prefix):
132 """Delete IPv6 address on linux host.
134 :param node: Linux node.
135 :param interface: Node interface.
136 :param addr: IPv6 address.
137 :param prefix: IPv6 address prefix.
146 cmd = "ifconfig {dev} inet6 del {ip}/{p}".format(dev=interface,
149 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
150 if int(ret_code) != 0:
151 raise Exception('TG ifconfig failed')
153 cmd = "ifconfig {dev} down".format(dev=interface)
154 (ret_code, _, _) = ssh.exec_command_sudo(cmd)
155 if int(ret_code) != 0:
156 raise Exception('TG ifconfig failed')
159 def vpp_set_if_ipv6_addr(node, interface, addr, prefix):
160 """Set IPv6 address on VPP.
162 :param node: VPP node.
163 :param interface: Node interface.
164 :param addr: IPv6 address.
165 :param prefix: IPv6 address prefix.
171 sw_if_index = Topology.get_interface_sw_index(node, interface)
172 with VatTerminal(node) as vat:
173 vat.vat_terminal_exec_cmd_from_template('add_ip_address.vat',
174 sw_if_index=sw_if_index,
176 prefix_length=prefix)
177 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
178 sw_if_index=sw_if_index,
183 cmd_input = 'exec show int'
184 (ret_code, stdout, stderr) = ssh.exec_command_sudo(
185 Constants.VAT_BIN_NAME, cmd_input)
186 logger.debug('ret: {0}'.format(ret_code))
187 logger.debug('stdout: {0}'.format(stdout))
188 logger.debug('stderr: {0}'.format(stderr))
191 def vpp_del_if_ipv6_addr(node, interface, addr, prefix):
192 """Delete IPv6 address on VPP.
194 :param node: VPP node.
195 :param interface: Node interface.
196 :param addr: IPv6 address.
197 :param prefix: IPv6 address prefix.
203 sw_if_index = Topology.get_interface_sw_index(node, interface)
204 with VatTerminal(node) as vat:
205 vat.vat_terminal_exec_cmd_from_template('del_ip_address.vat',
206 sw_if_index=sw_if_index,
208 prefix_length=prefix)
209 vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
210 sw_if_index=sw_if_index,
215 def vpp_set_ipv6_neighbor(node, interface, ipaddr, macaddr):
216 """Set IPv6 neighbor.
218 :param node: VPP node.
219 :param interface: Interface name.
220 :param ip_address: IPv6 address.
221 :param mac_address: mac address.
224 :type ip_address: str
225 :type mac_address: str
227 sw_if_index = Topology.get_interface_sw_index(node, interface)
228 VatExecutor.cmd_from_template(node,
229 'add_ip_neighbor.vat',
230 sw_if_index=sw_if_index,
235 def vpp_ra_suppress_link_layer(node, interface):
236 """Suppress ICMPv6 router advertisement message for link scope address.
238 :param node: VPP node.
239 :param interface: Interface name.
243 sw_if_index = Topology.get_interface_sw_index(node, interface)
244 VatExecutor.cmd_from_template(node,
245 'sw_interface_ip6nd_ra_config.vat',
246 sw_if_id=sw_if_index,
250 def vpp_ra_send_after_interval(node, interface, interval=2):
251 """Setup vpp router advertisement(RA) in such way it sends RA packet
252 after and every interval value.
254 :param node: VPP node.
255 :param interface: Interface name.
256 :param interval: Interval for RA resend
261 sw_if_index = Topology.get_interface_sw_index(node, interface)
262 VatExecutor.cmd_from_template(node,
263 'sw_interface_ip6nd_ra_config.vat',
264 sw_if_id=sw_if_index,
265 param='interval {0}'.format(interval))
267 def vpp_all_ra_suppress_link_layer(self, nodes):
268 """Suppress ICMPv6 router advertisement message for link scope address
269 on all VPP nodes in the topology.
271 :param nodes: Nodes of the test topology.
274 for node in nodes.values():
275 if node['type'] == NodeType.TG:
277 for port_k, port_v in node['interfaces'].items():
278 if_name = port_v.get('name')
281 self.vpp_ra_suppress_link_layer(node, if_name)
284 def get_link_address(link, nodes_addr):
285 """Get link IPv6 address.
287 :param link: Link name.
288 :param nodes_addr: Available nodes IPv6 addresses.
290 :type nodes_addr: dict
291 :return: Link IPv6 address.
294 net = nodes_addr.get(link)
296 raise ValueError('Link "{0}" address not found'.format(link))
297 return net.get('net_addr')
300 def get_link_prefix(link, nodes_addr):
301 """Get link IPv6 address prefix.
303 :param link: Link name.
304 :param nodes_addr: Available nodes IPv6 addresses.
306 :type nodes_addr: dict
307 :return: Link IPv6 address prefix.
310 net = nodes_addr.get(link)
312 raise ValueError('Link "{0}" address not found'.format(link))
313 return net.get('prefix')