Add with-statment support to VatTerminal.
[csit.git] / resources / libraries / python / IPv6Setup.py
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:
5 #
6 #     http://www.apache.org/licenses/LICENSE-2.0
7 #
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.
13
14 """Library to set up IPv6 in topology."""
15
16 from ssh import SSH
17 from ipaddress import IPv6Network
18 from topology import NodeType, Topology
19 from constants import Constants
20 from VatExecutor import VatTerminal, VatExecutor
21 from robot.api import logger
22
23
24 class IPv6Networks(object):
25     """IPv6 network iterator.
26
27        :param networks: List of the available IPv6 networks.
28        :type networks: list
29     """
30     def __init__(self, networks):
31         self._networks = list()
32         for network in networks:
33             net = IPv6Network(unicode(network))
34             self._networks.append(net)
35         num = len(self._networks)
36         if num == 0:
37             raise Exception('No IPv6 networks')
38
39     def next_network(self):
40         """Get the next elemnt of the iterator.
41
42            :return: IPv6 network.
43            :rtype: IPv6Network object
44            :raises: StopIteration if there is no more elements.
45         """
46         if len(self._networks):
47             return self._networks.pop()
48         else:
49             raise StopIteration()
50
51
52 class IPv6Setup(object):
53     """IPv6 setup in topology."""
54
55     def __init__(self):
56         pass
57
58     def nodes_setup_ipv6_addresses(self, nodes, nodes_addr):
59         """Setup IPv6 addresses on all VPP nodes in topology.
60
61            :param nodes: Nodes of the test topology.
62            :param nodes_addr: Available nodes IPv6 adresses.
63            :type nodes: dict
64            :type nodes_addr: dict
65         """
66         for net in nodes_addr.values():
67             for port in net['ports'].values():
68                 host = port.get('node')
69                 if host is None:
70                     continue
71                 topo = Topology()
72                 node = topo.get_node_by_hostname(nodes, host)
73                 if node is None:
74                     continue
75                 if node['type'] == NodeType.DUT:
76                     self.vpp_set_if_ipv6_addr(node, port['if'], port['addr'],
77                                               net['prefix'])
78
79     def nodes_clear_ipv6_addresses(self, nodes, nodes_addr):
80         """Remove IPv6 addresses from all VPP nodes in topology.
81
82            :param nodes: Nodes of the test topology.
83            :param nodes_addr: Available nodes IPv6 adresses.
84            :type nodes: dict
85            :type nodes_addr: dict
86          """
87         for net in nodes_addr.values():
88             for port in net['ports'].values():
89                 host = port.get('node')
90                 if host is None:
91                     continue
92                 topo = Topology()
93                 node = topo.get_node_by_hostname(nodes, host)
94                 if node is None:
95                     continue
96                 if node['type'] == NodeType.DUT:
97                     self.vpp_del_if_ipv6_addr(node, port['if'], port['addr'],
98                                               net['prefix'])
99
100     @staticmethod
101     def linux_set_if_ipv6_addr(node, interface, addr, prefix):
102         """Set IPv6 address on linux host.
103
104            :param node: Linux node.
105            :param interface: Node interface.
106            :param addr: IPv6 address.
107            :param prefix: IPv6 address prefix.
108            :type node: dict
109            :type interface: str
110            :type addr: str
111            :type prefix: str
112         """
113         ssh = SSH()
114         ssh.connect(node)
115
116         cmd = "ifconfig {dev} inet6 add {ip}/{p} up".format(dev=interface,
117                                                             ip=addr, p=prefix)
118         (ret_code, _, _) = ssh.exec_command_sudo(cmd)
119         if int(ret_code) != 0:
120             raise Exception('TG ifconfig failed')
121
122     @staticmethod
123     def linux_del_if_ipv6_addr(node, interface, addr, prefix):
124         """Delete IPv6 address on linux host.
125
126            :param node: Linux node.
127            :param interface: Node interface.
128            :param addr: IPv6 address.
129            :param prefix: IPv6 address prefix.
130            :type node: dict
131            :type interface: str
132            :type addr: str
133            :type prefix: str
134         """
135         ssh = SSH()
136         ssh.connect(node)
137
138         cmd = "ifconfig {dev} inet6 del {ip}/{p}".format(dev=interface,
139                                                          ip=addr,
140                                                          p=prefix)
141         (ret_code, _, _) = ssh.exec_command_sudo(cmd)
142         if int(ret_code) != 0:
143             raise Exception('TG ifconfig failed')
144
145         cmd = "ifconfig {dev} down".format(dev=interface)
146         (ret_code, _, _) = ssh.exec_command_sudo(cmd)
147         if int(ret_code) != 0:
148             raise Exception('TG ifconfig failed')
149
150     @staticmethod
151     def vpp_set_if_ipv6_addr(node, interface, addr, prefix):
152         """Set IPv6 address on VPP.
153
154            :param node: VPP node.
155            :param interface: Node interface.
156            :param addr: IPv6 address.
157            :param prefix: IPv6 address prefix.
158            :type node: dict
159            :type interface: str
160            :type addr: str
161            :type prefix: str
162         """
163         sw_if_index = Topology.get_interface_sw_index(node, interface)
164         with VatTerminal(node) as vat:
165             vat.vat_terminal_exec_cmd_from_template('add_ip_address.vat',
166                                                     sw_if_index=sw_if_index,
167                                                     address=addr,
168                                                     prefix_length=prefix)
169             vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
170                                                     sw_if_index=sw_if_index,
171                                                     state='admin-up')
172
173         ssh = SSH()
174         ssh.connect(node)
175         cmd_input = 'exec show int'
176         (ret_code, stdout, stderr) = ssh.exec_command_sudo(
177             Constants.VAT_BIN_NAME, cmd_input)
178         logger.debug('ret: {0}'.format(ret_code))
179         logger.debug('stdout: {0}'.format(stdout))
180         logger.debug('stderr: {0}'.format(stderr))
181
182     @staticmethod
183     def vpp_del_if_ipv6_addr(node, interface, addr, prefix):
184         """Delete IPv6 address on VPP.
185
186            :param node: VPP node.
187            :param interface: Node interface.
188            :param addr: IPv6 address.
189            :param prefix: IPv6 address prefix.
190            :type node: dict
191            :type interface: str
192            :type addr: str
193            :type prefix: str
194         """
195         sw_if_index = Topology.get_interface_sw_index(node, interface)
196         with VatTerminal(node) as vat:
197             vat.vat_terminal_exec_cmd_from_template('del_ip_address.vat',
198                                                     sw_if_index=sw_if_index,
199                                                     address=addr,
200                                                     prefix_length=prefix)
201             vat.vat_terminal_exec_cmd_from_template('set_if_state.vat',
202                                                     sw_if_index=sw_if_index,
203                                                     state='admin-down')
204
205     @staticmethod
206     def vpp_ra_supress_link_layer(node, interface):
207         """Supress ICMPv6 router advertisement message for link scope address
208
209            :param node: VPP node.
210            :param interface: Interface name.
211            :type node: dict
212            :type interface: str
213         """
214         sw_if_index = Topology.get_interface_sw_index(node, interface)
215         VatExecutor.cmd_from_template(node,
216                                       'sw_interface_ip6nd_ra_config.vat',
217                                       sw_if_id=sw_if_index,
218                                       param='surpress')
219
220     def vpp_all_ra_supress_link_layer(self, nodes):
221         """Supress ICMPv6 router advertisement message for link scope address
222            on all VPP nodes in the topology
223
224            :param nodes: Nodes of the test topology.
225            :type nodes: dict
226         """
227         for node in nodes.values():
228             if node['type'] == NodeType.TG:
229                 continue
230             for port_k, port_v in node['interfaces'].items():
231                 if port_k == 'mgmt':
232                     continue
233                 if_name = port_v.get('name')
234                 if if_name is None:
235                     continue
236                 self.vpp_ra_supress_link_layer(node, if_name)
237
238     @staticmethod
239     def get_link_address(link, nodes_addr):
240         """Get link IPv6 address.
241
242         :param link: Link name.
243         :param nodes_addr: Available nodes IPv6 adresses.
244         :type link: str
245         :type nodes_addr: dict
246         :return: Link IPv6 address.
247         :rtype: str
248         """
249         net = nodes_addr.get(link)
250         if net is None:
251             raise ValueError('Link "{0}" address not found'.format(link))
252         return net.get('net_addr')
253
254     @staticmethod
255     def get_link_prefix(link, nodes_addr):
256         """Get link IPv6 address prefix.
257
258         :param link: Link name.
259         :param nodes_addr: Available nodes IPv6 adresses.
260         :type link: str
261         :type nodes_addr: dict
262         :return: Link IPv6 address prefix.
263         :rtype: int
264         """
265         net = nodes_addr.get(link)
266         if net is None:
267             raise ValueError('Link "{0}" address not found'.format(link))
268         return net.get('prefix')

©2016 FD.io a Linux Foundation Collaborative Project. All Rights Reserved.
Linux Foundation is a registered trademark of The Linux Foundation. Linux is a registered trademark of Linus Torvalds.
Please see our privacy policy and terms of use.