4 object abstractions for representing IP routes in VPP
7 from vpp_object import *
8 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
10 # from vnet/vnet/mpls/mpls_types.h
11 MPLS_IETF_MAX_LABEL = 0xfffff
12 MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
16 MFIB_ITF_FLAG_NONE = 0
17 MFIB_ITF_FLAG_NEGATE_SIGNAL = 1
18 MFIB_ITF_FLAG_ACCEPT = 2
19 MFIB_ITF_FLAG_FORWARD = 4
20 MFIB_ITF_FLAG_SIGNAL_PRESENT = 8
21 MFIB_ITF_FLAG_INTERNAL_COPY = 16
24 class MRouteEntryFlags:
25 MFIB_ENTRY_FLAG_NONE = 0
26 MFIB_ENTRY_FLAG_SIGNAL = 1
27 MFIB_ENTRY_FLAG_DROP = 2
28 MFIB_ENTRY_FLAG_CONNECTED = 4
29 MFIB_ENTRY_FLAG_INHERIT_ACCEPT = 8
32 def find_route(test, ip_addr, len, table_id=0, inet=AF_INET):
35 routes = test.vapi.ip_fib_dump()
38 routes = test.vapi.ip6_fib_dump()
40 route_addr = inet_pton(inet, ip_addr)
42 if route_addr == e.address[:s] \
43 and len == e.address_length \
44 and table_id == e.table_id:
49 class VppRoutePath(object):
57 nh_via_label=MPLS_LABEL_INVALID,
59 self.nh_itf = nh_sw_if_index
60 self.nh_table_id = nh_table_id
61 self.nh_via_label = nh_via_label
62 self.nh_labels = labels
64 self.nh_addr = inet_pton(AF_INET6, nh_addr)
66 self.nh_addr = inet_pton(AF_INET, nh_addr)
69 class VppMRoutePath(VppRoutePath):
71 def __init__(self, nh_sw_if_index, flags):
72 super(VppMRoutePath, self).__init__("0.0.0.0",
74 self.nh_i_flags = flags
77 class VppIpRoute(VppObject):
82 def __init__(self, test, dest_addr,
83 dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0,
84 is_unreach=0, is_prohibit=0):
87 self.dest_addr_len = dest_addr_len
88 self.table_id = table_id
90 self.is_local = is_local
91 self.is_unreach = is_unreach
92 self.is_prohibit = is_prohibit
93 self.dest_addr_p = dest_addr
95 self.dest_addr = inet_pton(AF_INET6, dest_addr)
97 self.dest_addr = inet_pton(AF_INET, dest_addr)
99 def modify(self, paths, is_local=0,
100 is_unreach=0, is_prohibit=0):
102 self.is_local = is_local
103 self.is_unreach = is_unreach
104 self.is_prohibit = is_prohibit
106 def add_vpp_config(self):
107 if self.is_local or self.is_unreach or self.is_prohibit:
108 self._test.vapi.ip_add_del_route(
111 inet_pton(AF_INET6, "::"),
113 is_local=self.is_local,
114 is_unreach=self.is_unreach,
115 is_prohibit=self.is_prohibit,
116 table_id=self.table_id,
119 for path in self.paths:
120 self._test.vapi.ip_add_del_route(
125 table_id=self.table_id,
126 next_hop_out_label_stack=path.nh_labels,
127 next_hop_n_out_labels=len(
129 next_hop_via_label=path.nh_via_label,
131 self._test.registry.register(self, self._test.logger)
133 def remove_vpp_config(self):
134 if self.is_local or self.is_unreach or self.is_prohibit:
135 self._test.vapi.ip_add_del_route(
138 inet_pton(AF_INET6, "::"),
140 is_local=self.is_local,
141 is_unreach=self.is_unreach,
142 is_prohibit=self.is_prohibit,
144 table_id=self.table_id,
147 for path in self.paths:
148 self._test.vapi.ip_add_del_route(self.dest_addr,
152 table_id=self.table_id,
156 def query_vpp_config(self):
157 return find_route(self._test,
161 inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
164 return self.object_id()
173 class VppIpMRoute(VppObject):
178 def __init__(self, test, src_addr, grp_addr,
179 grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
182 self.grp_addr_len = grp_addr_len
183 self.table_id = table_id
184 self.e_flags = e_flags
188 self.grp_addr = inet_pton(AF_INET6, grp_addr)
189 self.src_addr = inet_pton(AF_INET6, src_addr)
191 self.grp_addr = inet_pton(AF_INET, grp_addr)
192 self.src_addr = inet_pton(AF_INET, src_addr)
194 def add_vpp_config(self):
195 for path in self.paths:
196 self._test.vapi.ip_mroute_add_del(self.src_addr,
202 table_id=self.table_id,
204 self._test.registry.register(self, self._test.logger)
206 def remove_vpp_config(self):
207 for path in self.paths:
208 self._test.vapi.ip_mroute_add_del(self.src_addr,
214 table_id=self.table_id,
218 def update_entry_flags(self, flags):
220 self._test.vapi.ip_mroute_add_del(self.src_addr,
226 table_id=self.table_id,
229 def update_path_flags(self, itf, flags):
230 for path in self.paths:
231 if path.nh_itf == itf:
232 path.nh_i_flags = flags
234 self._test.vapi.ip_mroute_add_del(self.src_addr,
240 table_id=self.table_id,
243 def query_vpp_config(self):
244 dump = self._test.vapi.ip_fib_dump()
246 if self.grp_addr == e.address \
247 and self.grp_addr_len == e.address_length \
248 and self.table_id == e.table_id:
253 return self.object_id()
257 return ("%d:(%s,%s/%d)"
259 inet_ntop(AF_INET6, self.src_addr),
260 inet_ntop(AF_INET6, self.grp_addr),
263 return ("%d:(%s,%s/%d)"
265 inet_ntop(AF_INET, self.src_addr),
266 inet_ntop(AF_INET, self.grp_addr),
270 class VppMFibSignal(object):
271 def __init__(self, test, route, interface, packet):
273 self.interface = interface
277 def compare(self, signal):
278 self.test.assertEqual(self.interface, signal.sw_if_index)
279 self.test.assertEqual(self.route.table_id, signal.table_id)
280 self.test.assertEqual(self.route.grp_addr_len,
281 signal.grp_address_len)
282 for i in range(self.route.grp_addr_len / 8):
283 self.test.assertEqual(self.route.grp_addr[i],
284 signal.grp_address[i])
285 if (self.route.grp_addr_len > 32):
287 self.test.assertEqual(self.route.src_addr[i],
288 signal.src_address[i])
291 class VppMplsIpBind(VppObject):
296 def __init__(self, test, local_label, dest_addr, dest_addr_len,
297 table_id=0, ip_table_id=0):
299 self.dest_addr = inet_pton(AF_INET, dest_addr)
300 self.dest_addr_len = dest_addr_len
301 self.local_label = local_label
302 self.table_id = table_id
303 self.ip_table_id = ip_table_id
305 def add_vpp_config(self):
306 self._test.vapi.mpls_ip_bind_unbind(self.local_label,
309 table_id=self.table_id,
310 ip_table_id=self.ip_table_id)
311 self._test.registry.register(self, self._test.logger)
313 def remove_vpp_config(self):
314 self._test.vapi.mpls_ip_bind_unbind(self.local_label,
319 def query_vpp_config(self):
320 dump = self._test.vapi.mpls_fib_dump()
322 if self.local_label == e.label \
323 and self.eos_bit == e.eos_bit \
324 and self.table_id == e.table_id:
329 return self.object_id()
332 return ("%d:%s binds %d:%s/%d"
336 inet_ntop(AF_INET, self.dest_addr),
340 class VppMplsRoute(VppObject):
345 def __init__(self, test, local_label, eos_bit, paths, table_id=0):
348 self.local_label = local_label
349 self.eos_bit = eos_bit
350 self.table_id = table_id
352 def add_vpp_config(self):
353 for path in self.paths:
354 self._test.vapi.mpls_route_add_del(
360 table_id=self.table_id,
361 next_hop_out_label_stack=path.nh_labels,
362 next_hop_n_out_labels=len(
364 next_hop_via_label=path.nh_via_label,
365 next_hop_table_id=path.nh_table_id)
366 self._test.registry.register(self, self._test.logger)
368 def remove_vpp_config(self):
369 for path in self.paths:
370 self._test.vapi.mpls_route_add_del(self.local_label,
375 table_id=self.table_id,
378 def query_vpp_config(self):
379 dump = self._test.vapi.mpls_fib_dump()
381 if self.local_label == e.label \
382 and self.eos_bit == e.eos_bit \
383 and self.table_id == e.table_id:
388 return self.object_id()