f758c067f31011c0d79434874f7bf50c5bdd1308
[vpp.git] / test / vpp_ip_route.py
1 """
2   IP Routes
3
4   object abstractions for representing IP routes in VPP
5 """
6
7 import socket
8 from vpp_object import *
9
10 # from vnet/vnet/mpls/mpls_types.h
11 MPLS_IETF_MAX_LABEL = 0xfffff
12 MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
13
14
15 class VppRoutePath(object):
16
17     def __init__(
18             self,
19             nh_addr,
20             nh_sw_if_index,
21             nh_table_id=0,
22             labels=[],
23             nh_via_label=MPLS_LABEL_INVALID,
24             is_ip6=0):
25         self.nh_itf = nh_sw_if_index
26         self.nh_table_id = nh_table_id
27         self.nh_via_label = nh_via_label
28         self.nh_labels = labels
29         if is_ip6:
30             self.nh_addr = socket.inet_pton(socket.AF_INET6, nh_addr)
31         else:
32             self.nh_addr = socket.inet_pton(socket.AF_INET, nh_addr)
33
34
35 class VppMRoutePath(VppRoutePath):
36
37     def __init__(self, nh_sw_if_index, flags):
38         super(VppMRoutePath, self).__init__("0.0.0.0",
39                                             nh_sw_if_index)
40         self.nh_i_flags = flags
41
42
43 class VppIpRoute(VppObject):
44     """
45     IP Route
46     """
47
48     def __init__(self, test, dest_addr,
49                  dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0):
50         self._test = test
51         self.paths = paths
52         self.dest_addr_len = dest_addr_len
53         self.table_id = table_id
54         self.is_ip6 = is_ip6
55         self.is_local = is_local
56         if is_ip6:
57             self.dest_addr = socket.inet_pton(socket.AF_INET6, dest_addr)
58         else:
59             self.dest_addr = socket.inet_pton(socket.AF_INET, dest_addr)
60
61     def add_vpp_config(self):
62         if self.is_local:
63             self._test.vapi.ip_add_del_route(
64                 self.dest_addr,
65                 self.dest_addr_len,
66                 socket.inet_pton(socket.AF_INET6, "::"),
67                 0xffffffff,
68                 is_local=1,
69                 table_id=self.table_id,
70                 is_ipv6=self.is_ip6)
71         else:
72             for path in self.paths:
73                 self._test.vapi.ip_add_del_route(
74                     self.dest_addr,
75                     self.dest_addr_len,
76                     path.nh_addr,
77                     path.nh_itf,
78                     table_id=self.table_id,
79                     next_hop_out_label_stack=path.nh_labels,
80                     next_hop_n_out_labels=len(
81                         path.nh_labels),
82                     next_hop_via_label=path.nh_via_label,
83                     is_ipv6=self.is_ip6)
84         self._test.registry.register(self, self._test.logger)
85
86     def remove_vpp_config(self):
87         if self.is_local:
88             self._test.vapi.ip_add_del_route(
89                 self.dest_addr,
90                 self.dest_addr_len,
91                 socket.inet_pton(socket.AF_INET6, "::"),
92                 0xffffffff,
93                 is_local=1,
94                 is_add=0,
95                 table_id=self.table_id,
96                 is_ipv6=self.is_ip6)
97         else:
98             for path in self.paths:
99                 self._test.vapi.ip_add_del_route(self.dest_addr,
100                                                  self.dest_addr_len,
101                                                  path.nh_addr,
102                                                  path.nh_itf,
103                                                  table_id=self.table_id,
104                                                  is_add=0)
105
106     def query_vpp_config(self):
107         dump = self._test.vapi.ip_fib_dump()
108         for e in dump:
109             if self.dest_addr == e.address \
110                and self.dest_addr_len == e.address_length \
111                and self.table_id == e.table_id:
112                 return True
113         return False
114
115     def __str__(self):
116         return self.object_id()
117
118     def object_id(self):
119         return ("%d:%s/%d"
120                 % (self.table_id,
121                    socket.inet_ntop(socket.AF_INET, self.dest_addr),
122                    self.dest_addr_len))
123
124
125 class VppIpMRoute(VppObject):
126     """
127     IP Multicast Route
128     """
129
130     def __init__(self, test, src_addr, grp_addr,
131                  grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
132         self._test = test
133         self.paths = paths
134         self.grp_addr_len = grp_addr_len
135         self.table_id = table_id
136         self.e_flags = e_flags
137         self.is_ip6 = is_ip6
138
139         if is_ip6:
140             self.grp_addr = socket.inet_pton(socket.AF_INET6, grp_addr)
141             self.src_addr = socket.inet_pton(socket.AF_INET6, src_addr)
142         else:
143             self.grp_addr = socket.inet_pton(socket.AF_INET, grp_addr)
144             self.src_addr = socket.inet_pton(socket.AF_INET, src_addr)
145
146     def add_vpp_config(self):
147         for path in self.paths:
148             self._test.vapi.ip_mroute_add_del(self.src_addr,
149                                               self.grp_addr,
150                                               self.grp_addr_len,
151                                               self.e_flags,
152                                               path.nh_itf,
153                                               path.nh_i_flags,
154                                               table_id=self.table_id,
155                                               is_ipv6=self.is_ip6)
156         self._test.registry.register(self, self._test.logger)
157
158     def remove_vpp_config(self):
159         for path in self.paths:
160             self._test.vapi.ip_mroute_add_del(self.src_addr,
161                                               self.grp_addr,
162                                               self.grp_addr_len,
163                                               self.e_flags,
164                                               path.nh_itf,
165                                               path.nh_i_flags,
166                                               table_id=self.table_id,
167                                               is_add=0,
168                                               is_ipv6=self.is_ip6)
169
170     def update_entry_flags(self, flags):
171         self.e_flags = flags
172         self._test.vapi.ip_mroute_add_del(self.src_addr,
173                                           self.grp_addr,
174                                           self.grp_addr_len,
175                                           self.e_flags,
176                                           0xffffffff,
177                                           0,
178                                           table_id=self.table_id,
179                                           is_ipv6=self.is_ip6)
180
181     def update_path_flags(self, itf, flags):
182         for path in self.paths:
183             if path.nh_itf == itf:
184                 path.nh_i_flags = flags
185                 break
186         self._test.vapi.ip_mroute_add_del(self.src_addr,
187                                           self.grp_addr,
188                                           self.grp_addr_len,
189                                           self.e_flags,
190                                           path.nh_itf,
191                                           path.nh_i_flags,
192                                           table_id=self.table_id,
193                                           is_ipv6=self.is_ip6)
194
195     def query_vpp_config(self):
196         dump = self._test.vapi.ip_fib_dump()
197         for e in dump:
198             if self.grp_addr == e.address \
199                and self.grp_addr_len == e.address_length \
200                and self.table_id == e.table_id:
201                 return True
202         return False
203
204     def __str__(self):
205         return self.object_id()
206
207     def object_id(self):
208         if self.is_ip6:
209             return ("%d:(%s,%s/%d)"
210                     % (self.table_id,
211                        socket.inet_ntop(socket.AF_INET6, self.src_addr),
212                        socket.inet_ntop(socket.AF_INET6, self.grp_addr),
213                        self.grp_addr_len))
214         else:
215             return ("%d:(%s,%s/%d)"
216                     % (self.table_id,
217                        socket.inet_ntop(socket.AF_INET, self.src_addr),
218                        socket.inet_ntop(socket.AF_INET, self.grp_addr),
219                        self.grp_addr_len))
220
221
222 class VppMFibSignal(object):
223     def __init__(self, test, route, interface, packet):
224         self.route = route
225         self.interface = interface
226         self.packet = packet
227         self.test = test
228
229     def compare(self, signal):
230         self.test.assertEqual(self.interface, signal.sw_if_index)
231         self.test.assertEqual(self.route.table_id, signal.table_id)
232         self.test.assertEqual(self.route.grp_addr_len,
233                               signal.grp_address_len)
234         for i in range(self.route.grp_addr_len / 8):
235             self.test.assertEqual(self.route.grp_addr[i],
236                                   signal.grp_address[i])
237         if (self.route.grp_addr_len > 32):
238             for i in range(4):
239                 self.test.assertEqual(self.route.src_addr[i],
240                                       signal.src_address[i])
241
242
243 class VppMplsIpBind(VppObject):
244     """
245     MPLS to IP Binding
246     """
247
248     def __init__(self, test, local_label, dest_addr, dest_addr_len,
249                  table_id=0, ip_table_id=0):
250         self._test = test
251         self.dest_addr = socket.inet_pton(socket.AF_INET, dest_addr)
252         self.dest_addr_len = dest_addr_len
253         self.local_label = local_label
254         self.table_id = table_id
255         self.ip_table_id = ip_table_id
256
257     def add_vpp_config(self):
258         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
259                                             self.dest_addr,
260                                             self.dest_addr_len,
261                                             table_id=self.table_id,
262                                             ip_table_id=self.ip_table_id)
263         self._test.registry.register(self, self._test.logger)
264
265     def remove_vpp_config(self):
266         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
267                                             self.dest_addr,
268                                             self.dest_addr_len,
269                                             is_bind=0)
270
271     def query_vpp_config(self):
272         dump = self._test.vapi.mpls_fib_dump()
273         for e in dump:
274             if self.local_label == e.label \
275                and self.eos_bit == e.eos_bit \
276                and self.table_id == e.table_id:
277                 return True
278         return False
279
280     def __str__(self):
281         return self.object_id()
282
283     def object_id(self):
284         return ("%d:%s binds %d:%s/%d"
285                 % (self.table_id,
286                    self.local_label,
287                    self.ip_table_id,
288                    socket.inet_ntop(socket.AF_INET, self.dest_addr),
289                    self.dest_addr_len))
290
291
292 class VppMplsRoute(VppObject):
293     """
294     MPLS Route/LSP
295     """
296
297     def __init__(self, test, local_label, eos_bit, paths, table_id=0):
298         self._test = test
299         self.paths = paths
300         self.local_label = local_label
301         self.eos_bit = eos_bit
302         self.table_id = table_id
303
304     def add_vpp_config(self):
305         for path in self.paths:
306             self._test.vapi.mpls_route_add_del(
307                 self.local_label,
308                 self.eos_bit,
309                 1,
310                 path.nh_addr,
311                 path.nh_itf,
312                 table_id=self.table_id,
313                 next_hop_out_label_stack=path.nh_labels,
314                 next_hop_n_out_labels=len(
315                     path.nh_labels),
316                 next_hop_via_label=path.nh_via_label,
317                 next_hop_table_id=path.nh_table_id)
318         self._test.registry.register(self, self._test.logger)
319
320     def remove_vpp_config(self):
321         for path in self.paths:
322             self._test.vapi.mpls_route_add_del(self.local_label,
323                                                self.eos_bit,
324                                                1,
325                                                path.nh_addr,
326                                                path.nh_itf,
327                                                table_id=self.table_id,
328                                                is_add=0)
329
330     def query_vpp_config(self):
331         dump = self._test.vapi.mpls_fib_dump()
332         for e in dump:
333             if self.local_label == e.label \
334                and self.eos_bit == e.eos_bit \
335                and self.table_id == e.table_id:
336                 return True
337         return False
338
339     def __str__(self):
340         return self.object_id()
341
342     def object_id(self):
343         return ("%d:%s/%d"
344                 % (self.table_id,
345                    self.local_label,
346                    20+self.eos_bit))