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,
216 def vpp_ra_suppress_link_layer(node, interface):
217 """Suppress ICMPv6 router advertisement message for link scope address.
219 :param node: VPP node.
220 :param interface: Interface name.
224 sw_if_index = Topology.get_interface_sw_index(node, interface)
225 VatExecutor.cmd_from_template(node,
226 'sw_interface_ip6nd_ra_config.vat',
227 sw_if_id=sw_if_index,
231 def vpp_ra_send_after_interval(node, interface, interval=2):
232 """Setup vpp router advertisement(RA) in such way it sends RA packet
233 after and every interval value.
235 :param node: VPP node.
236 :param interface: Interface name.
237 :param interval: Interval for RA resend
242 sw_if_index = Topology.get_interface_sw_index(node, interface)
243 VatExecutor.cmd_from_template(node,
244 'sw_interface_ip6nd_ra_config.vat',
245 sw_if_id=sw_if_index,
246 param='interval {0}'.format(interval))
248 def vpp_all_ra_suppress_link_layer(self, nodes):
249 """Suppress ICMPv6 router advertisement message for link scope address
250 on all VPP nodes in the topology.
252 :param nodes: Nodes of the test topology.
255 for node in nodes.values():
256 if node['type'] == NodeType.TG:
258 for port_k in node['interfaces'].keys():
259 self.vpp_ra_suppress_link_layer(node, port_k)
262 def get_link_address(link, nodes_addr):
263 """Get link IPv6 address.
265 :param link: Link name.
266 :param nodes_addr: Available nodes IPv6 addresses.
268 :type nodes_addr: dict
269 :return: Link IPv6 address.
272 net = nodes_addr.get(link)
274 raise ValueError('Link "{0}" address not found'.format(link))
275 return net.get('net_addr')
278 def get_link_prefix(link, nodes_addr):
279 """Get link IPv6 address prefix.
281 :param link: Link name.
282 :param nodes_addr: Available nodes IPv6 addresses.
284 :type nodes_addr: dict
285 :return: Link IPv6 address prefix.
288 net = nodes_addr.get(link)
290 raise ValueError('Link "{0}" address not found'.format(link))
291 return net.get('prefix')