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
15 def find_route(test, ip_addr, len, table_id=0, inet=AF_INET):
18 routes = test.vapi.ip_fib_dump()
21 routes = test.vapi.ip6_fib_dump()
23 route_addr = inet_pton(inet, ip_addr)
25 if route_addr == e.address[:s] \
26 and len == e.address_length \
27 and table_id == e.table_id:
32 class VppRoutePath(object):
40 nh_via_label=MPLS_LABEL_INVALID,
42 self.nh_itf = nh_sw_if_index
43 self.nh_table_id = nh_table_id
44 self.nh_via_label = nh_via_label
45 self.nh_labels = labels
47 self.nh_addr = inet_pton(AF_INET6, nh_addr)
49 self.nh_addr = inet_pton(AF_INET, nh_addr)
52 class VppMRoutePath(VppRoutePath):
54 def __init__(self, nh_sw_if_index, flags):
55 super(VppMRoutePath, self).__init__("0.0.0.0",
57 self.nh_i_flags = flags
60 class VppIpRoute(VppObject):
65 def __init__(self, test, dest_addr,
66 dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0,
67 is_unreach=0, is_prohibit=0):
70 self.dest_addr_len = dest_addr_len
71 self.table_id = table_id
73 self.is_local = is_local
74 self.is_unreach = is_unreach
75 self.is_prohibit = is_prohibit
76 self.dest_addr_p = dest_addr
78 self.dest_addr = inet_pton(AF_INET6, dest_addr)
80 self.dest_addr = inet_pton(AF_INET, dest_addr)
82 def modify(self, paths, is_local=0,
83 is_unreach=0, is_prohibit=0):
85 self.is_local = is_local
86 self.is_unreach = is_unreach
87 self.is_prohibit = is_prohibit
89 def add_vpp_config(self):
90 if self.is_local or self.is_unreach or self.is_prohibit:
91 self._test.vapi.ip_add_del_route(
94 inet_pton(AF_INET6, "::"),
96 is_local=self.is_local,
97 is_unreach=self.is_unreach,
98 is_prohibit=self.is_prohibit,
99 table_id=self.table_id,
102 for path in self.paths:
103 self._test.vapi.ip_add_del_route(
108 table_id=self.table_id,
109 next_hop_out_label_stack=path.nh_labels,
110 next_hop_n_out_labels=len(
112 next_hop_via_label=path.nh_via_label,
114 self._test.registry.register(self, self._test.logger)
116 def remove_vpp_config(self):
117 if self.is_local or self.is_unreach or self.is_prohibit:
118 self._test.vapi.ip_add_del_route(
121 inet_pton(AF_INET6, "::"),
123 is_local=self.is_local,
124 is_unreach=self.is_unreach,
125 is_prohibit=self.is_prohibit,
127 table_id=self.table_id,
130 for path in self.paths:
131 self._test.vapi.ip_add_del_route(self.dest_addr,
135 table_id=self.table_id,
139 def query_vpp_config(self):
140 return find_route(self._test,
144 inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
147 return self.object_id()
156 class VppIpMRoute(VppObject):
161 def __init__(self, test, src_addr, grp_addr,
162 grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
165 self.grp_addr_len = grp_addr_len
166 self.table_id = table_id
167 self.e_flags = e_flags
171 self.grp_addr = inet_pton(AF_INET6, grp_addr)
172 self.src_addr = inet_pton(AF_INET6, src_addr)
174 self.grp_addr = inet_pton(AF_INET, grp_addr)
175 self.src_addr = inet_pton(AF_INET, src_addr)
177 def add_vpp_config(self):
178 for path in self.paths:
179 self._test.vapi.ip_mroute_add_del(self.src_addr,
185 table_id=self.table_id,
187 self._test.registry.register(self, self._test.logger)
189 def remove_vpp_config(self):
190 for path in self.paths:
191 self._test.vapi.ip_mroute_add_del(self.src_addr,
197 table_id=self.table_id,
201 def update_entry_flags(self, flags):
203 self._test.vapi.ip_mroute_add_del(self.src_addr,
209 table_id=self.table_id,
212 def update_path_flags(self, itf, flags):
213 for path in self.paths:
214 if path.nh_itf == itf:
215 path.nh_i_flags = flags
217 self._test.vapi.ip_mroute_add_del(self.src_addr,
223 table_id=self.table_id,
226 def query_vpp_config(self):
227 dump = self._test.vapi.ip_fib_dump()
229 if self.grp_addr == e.address \
230 and self.grp_addr_len == e.address_length \
231 and self.table_id == e.table_id:
236 return self.object_id()
240 return ("%d:(%s,%s/%d)"
242 inet_ntop(AF_INET6, self.src_addr),
243 inet_ntop(AF_INET6, self.grp_addr),
246 return ("%d:(%s,%s/%d)"
248 inet_ntop(AF_INET, self.src_addr),
249 inet_ntop(AF_INET, self.grp_addr),
253 class VppMFibSignal(object):
254 def __init__(self, test, route, interface, packet):
256 self.interface = interface
260 def compare(self, signal):
261 self.test.assertEqual(self.interface, signal.sw_if_index)
262 self.test.assertEqual(self.route.table_id, signal.table_id)
263 self.test.assertEqual(self.route.grp_addr_len,
264 signal.grp_address_len)
265 for i in range(self.route.grp_addr_len / 8):
266 self.test.assertEqual(self.route.grp_addr[i],
267 signal.grp_address[i])
268 if (self.route.grp_addr_len > 32):
270 self.test.assertEqual(self.route.src_addr[i],
271 signal.src_address[i])
274 class VppMplsIpBind(VppObject):
279 def __init__(self, test, local_label, dest_addr, dest_addr_len,
280 table_id=0, ip_table_id=0):
282 self.dest_addr = inet_pton(AF_INET, dest_addr)
283 self.dest_addr_len = dest_addr_len
284 self.local_label = local_label
285 self.table_id = table_id
286 self.ip_table_id = ip_table_id
288 def add_vpp_config(self):
289 self._test.vapi.mpls_ip_bind_unbind(self.local_label,
292 table_id=self.table_id,
293 ip_table_id=self.ip_table_id)
294 self._test.registry.register(self, self._test.logger)
296 def remove_vpp_config(self):
297 self._test.vapi.mpls_ip_bind_unbind(self.local_label,
302 def query_vpp_config(self):
303 dump = self._test.vapi.mpls_fib_dump()
305 if self.local_label == e.label \
306 and self.eos_bit == e.eos_bit \
307 and self.table_id == e.table_id:
312 return self.object_id()
315 return ("%d:%s binds %d:%s/%d"
319 inet_ntop(AF_INET, self.dest_addr),
323 class VppMplsRoute(VppObject):
328 def __init__(self, test, local_label, eos_bit, paths, table_id=0):
331 self.local_label = local_label
332 self.eos_bit = eos_bit
333 self.table_id = table_id
335 def add_vpp_config(self):
336 for path in self.paths:
337 self._test.vapi.mpls_route_add_del(
343 table_id=self.table_id,
344 next_hop_out_label_stack=path.nh_labels,
345 next_hop_n_out_labels=len(
347 next_hop_via_label=path.nh_via_label,
348 next_hop_table_id=path.nh_table_id)
349 self._test.registry.register(self, self._test.logger)
351 def remove_vpp_config(self):
352 for path in self.paths:
353 self._test.vapi.mpls_route_add_del(self.local_label,
358 table_id=self.table_id,
361 def query_vpp_config(self):
362 dump = self._test.vapi.mpls_fib_dump()
364 if self.local_label == e.label \
365 and self.eos_bit == e.eos_bit \
366 and self.table_id == e.table_id:
371 return self.object_id()