4 from vpp_papi.vpp_serializer import VPPType, VPPEnumType
5 from vpp_papi.vpp_serializer import VPPUnionType, VPPMessage
6 from vpp_papi.vpp_serializer import VPPTypeAlias, VPPSerializerValueError
7 from vpp_papi import MACAddress
8 from socket import inet_pton, AF_INET, AF_INET6
11 from ipaddress import *
14 class TestLimits(unittest.TestCase):
15 def test_string(self):
16 fixed_string = VPPType('fixed_string',
17 [['string', 'name', 16]])
19 b = fixed_string.pack({'name': 'foobar'})
20 self.assertEqual(len(b), 16)
22 # Ensure string is nul terminated
23 self.assertEqual(b.decode('ascii')[6], '\x00')
25 nt, size = fixed_string.unpack(b)
26 self.assertEqual(size, 16)
27 self.assertEqual(nt.name, 'foobar')
30 b = fixed_string.pack({'name': ''})
31 self.assertEqual(len(b), 16)
32 nt, size = fixed_string.unpack(b)
33 self.assertEqual(size, 16)
34 self.assertEqual(nt.name, '')
37 with self.assertRaises(VPPSerializerValueError):
38 b = fixed_string.pack({'name': 'foobarfoobar1234'})
40 variable_string = VPPType('variable_string',
41 [['string', 'name', 0]])
42 b = variable_string.pack({'name': 'foobar'})
43 self.assertEqual(len(b), 4 + len('foobar'))
45 nt, size = variable_string.unpack(b)
46 self.assertEqual(size, 4 + len('foobar'))
47 self.assertEqual(nt.name, 'foobar')
48 self.assertEqual(len(nt.name), len('foobar'))
52 limited_type = VPPType('limited_type_t',
53 [['string', 'name', 0, {'limit': 16}]])
54 unlimited_type = VPPType('limited_type_t',
55 [['string', 'name', 0]])
57 b = limited_type.pack({'name': 'foobar'})
58 self.assertEqual(len(b), 10)
59 b = unlimited_type.pack({'name': 'foobar'})
60 self.assertEqual(len(b), 10)
62 with self.assertRaises(VPPSerializerValueError):
63 b = limited_type.pack({'name': 'foobar'*3})
66 class TestDefaults(unittest.TestCase):
67 def test_defaults(self):
68 default_type = VPPType('default_type_t',
69 [['u16', 'mtu', {'default': 1500, 'limit': 0}]])
70 without_default_type = VPPType('without_default_type_t',
73 b = default_type.pack({})
74 self.assertEqual(len(b), 2)
75 nt, size = default_type.unpack(b)
76 self.assertEqual(len(b), size)
77 self.assertEqual(nt.mtu, 1500)
79 # distinguish between parameter 0 and parameter not passed
80 b = default_type.pack({'mtu': 0})
81 self.assertEqual(len(b), 2)
82 nt, size = default_type.unpack(b)
83 self.assertEqual(len(b), size)
84 self.assertEqual(nt.mtu, 0)
86 # Ensure that basetypes does not inherit default
87 b = without_default_type.pack({})
88 self.assertEqual(len(b), 2)
89 nt, size = default_type.unpack(b)
90 self.assertEqual(len(b), size)
91 self.assertEqual(nt.mtu, 0)
94 VPPEnumType('vl_api_enum_t', [["ADDRESS_IP4", 0],
98 default_with_enum = VPPType('default_enum_type_t',
99 [['u16', 'mtu'], ['vl_api_enum_t',
100 'e', {'default': 1}]])
102 b = default_with_enum.pack({})
103 self.assertEqual(len(b), 6)
104 nt, size = default_with_enum.unpack(b)
105 self.assertEqual(len(b), size)
106 self.assertEqual(nt.e, 1)
108 class TestAddType(unittest.TestCase):
110 def test_union(self):
111 un = VPPUnionType('test_union',
115 b = un.pack({'is_int': 0x12345678})
116 nt, size = un.unpack(b)
117 self.assertEqual(len(b), size)
118 self.assertEqual(nt.is_bool, 0x12)
119 self.assertEqual(nt.is_int, 0x12345678)
121 def test_address(self):
122 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
124 {"enumtype": "u32"}])
125 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
127 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
129 VPPUnionType('vl_api_address_union_t',
130 [["vl_api_ip4_address_t", "ip4"],
131 ["vl_api_ip6_address_t", "ip6"]])
133 address = VPPType('vl_api_address_t',
134 [['vl_api_address_family_t', 'af'],
135 ['vl_api_address_union_t', 'un']])
137 prefix = VPPType('vl_api_prefix_t',
138 [['vl_api_address_t', 'address'],
141 va_address_list = VPPType('list_addresses',
143 ['vl_api_address_t', 'addresses',
146 message_with_va_address_list = VPPType('msg_with_vla',
151 b = ip4.pack(inet_pton(AF_INET, '1.1.1.1'))
152 self.assertEqual(len(b), 4)
153 nt, size = ip4.unpack(b)
154 self.assertEqual(str(nt), '1.1.1.1')
156 b = ip6.pack(inet_pton(AF_INET6, '1::1'))
157 self.assertEqual(len(b), 16)
159 b = address.pack({'af': af.ADDRESS_IP4,
161 {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
162 self.assertEqual(len(b), 20)
164 nt, size = address.unpack(b)
165 self.assertEqual(str(nt), '2.2.2.2')
170 address_list.append({'af': af.ADDRESS_IP4,
172 {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
173 b = va_address_list.pack({'count': len(address_list),
174 'addresses': address_list})
175 self.assertEqual(len(b), 81)
177 nt, size = va_address_list.unpack(b)
178 self.assertEqual(str(nt.addresses[0]), '2.2.2.2')
180 b = message_with_va_address_list.pack({'vla_address':
181 {'count': len(address_list),
182 'addresses': address_list},
184 self.assertEqual(len(b), 82)
185 nt, size = message_with_va_address_list.unpack(b)
186 self.assertEqual(nt.is_cool, 100)
188 def test_address_with_prefix(self):
189 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
191 {"enumtype": "u32"}])
192 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
194 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
196 VPPUnionType('vl_api_address_union_t',
197 [["vl_api_ip4_address_t", "ip4"],
198 ["vl_api_ip6_address_t", "ip6"]])
200 address = VPPType('vl_api_address_t',
201 [['vl_api_address_family_t', 'af'],
202 ['vl_api_address_union_t', 'un']])
205 prefix = VPPType('vl_api_prefix_t',
206 [['vl_api_address_t', 'address'],
208 prefix4 = VPPType('vl_api_ip4_prefix_t',
209 [['vl_api_ip4_address_t', 'address'],
211 prefix6 = VPPType('vl_api_ip6_prefix_t',
212 [['vl_api_ip6_address_t', 'address'],
215 address_with_prefix = VPPTypeAlias('vl_api_address_with_prefix_t', {'type': 'vl_api_prefix_t' })
216 address4_with_prefix = VPPTypeAlias('vl_api_ip4_address_with_prefix_t',
217 {'type': 'vl_api_ip4_prefix_t' })
218 address6_with_prefix = VPPTypeAlias('vl_api_ip6_address_with_prefix_t',
219 {'type': 'vl_api_ip6_prefix_t' })
221 awp_type = VPPType('foobar_t',
222 [['vl_api_address_with_prefix_t', 'address']])
224 # address with prefix
225 b = address_with_prefix.pack(IPv4Interface('2.2.2.2/24'))
226 self.assertEqual(len(b), 21)
227 nt, size = address_with_prefix.unpack(b)
228 self.assertTrue(isinstance(nt, IPv4Interface))
229 self.assertEqual(str(nt), '2.2.2.2/24')
231 b = address_with_prefix.pack(IPv6Interface('2::2/64'))
232 self.assertEqual(len(b), 21)
233 nt, size = address_with_prefix.unpack(b)
234 self.assertTrue(isinstance(nt, IPv6Interface))
235 self.assertEqual(str(nt), '2::2/64')
237 b = address_with_prefix.pack(IPv4Network('2.2.2.2/24', strict=False))
238 self.assertEqual(len(b), 21)
239 nt, size = address_with_prefix.unpack(b)
240 self.assertTrue(isinstance(nt, IPv4Interface))
241 self.assertEqual(str(nt), '2.2.2.0/24')
243 b = address4_with_prefix.pack('2.2.2.2/24')
244 self.assertEqual(len(b), 5)
245 nt, size = address4_with_prefix.unpack(b)
246 self.assertTrue(isinstance(nt, IPv4Interface))
247 self.assertEqual(str(nt), '2.2.2.2/24')
248 b = address4_with_prefix.pack(IPv4Interface('2.2.2.2/24'))
249 self.assertEqual(len(b), 5)
251 b = address6_with_prefix.pack('2::2/64')
252 self.assertEqual(len(b), 17)
253 nt, size = address6_with_prefix.unpack(b)
254 self.assertTrue(isinstance(nt, IPv6Interface))
255 self.assertEqual(str(nt), '2::2/64')
256 b = address6_with_prefix.pack(IPv6Interface('2::2/64'))
257 self.assertEqual(len(b), 17)
259 b = prefix.pack('192.168.10.0/24')
260 self.assertEqual(len(b), 21)
261 nt, size = prefix.unpack(b)
262 self.assertTrue(isinstance(nt, IPv4Network))
263 self.assertEqual(str(nt), '192.168.10.0/24')
265 b = awp_type.pack({'address': '1.2.3.4/24'})
266 self.assertEqual(len(b), 21)
267 nt, size = awp_type.unpack(b)
268 self.assertTrue(isinstance(nt.address, IPv4Interface))
269 self.assertEqual(str(nt.address), '1.2.3.4/24')
271 b = awp_type.pack({'address': IPv4Interface('1.2.3.4/24')})
272 self.assertEqual(len(b), 21)
273 nt, size = awp_type.unpack(b)
274 self.assertTrue(isinstance(nt.address, IPv4Interface))
275 self.assertEqual(str(nt.address), '1.2.3.4/24')
278 def test_recursive_address(self):
279 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
281 {"enumtype": "u32"}])
282 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
284 b = ip4.pack('1.1.1.1')
285 self.assertEqual(len(b), 4)
286 nt, size = ip4.unpack(b)
288 self.assertEqual(str(nt), '1.1.1.1')
290 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
292 VPPUnionType('vl_api_address_union_t',
293 [["vl_api_ip4_address_t", "ip4"],
294 ["vl_api_ip6_address_t", "ip6"]])
296 address = VPPType('vl_api_address_t',
297 [['vl_api_address_family_t', 'af'],
298 ['vl_api_address_union_t', 'un']])
300 prefix = VPPType('vl_api_prefix_t',
301 [['vl_api_address_t', 'address'],
303 message = VPPMessage('svs',
304 [['vl_api_prefix_t', 'prefix']])
305 message_addr = VPPMessage('svs_address',
306 [['vl_api_address_t', 'address']])
308 b = message_addr.pack({'address': "1::1"})
309 self.assertEqual(len(b), 20)
310 nt, size = message_addr.unpack(b)
311 self.assertEqual("1::1", str(nt.address))
312 b = message_addr.pack({'address': "1.1.1.1"})
313 self.assertEqual(len(b), 20)
314 nt, size = message_addr.unpack(b)
315 self.assertEqual("1.1.1.1", str(nt.address))
317 b = message.pack({'prefix': "1.1.1.0/24"})
318 self.assertEqual(len(b), 21)
319 nt, size = message.unpack(b)
320 self.assertEqual("1.1.1.0/24", str(nt.prefix))
322 message_array = VPPMessage('address_array',
323 [['vl_api_ip6_address_t',
325 b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
326 self.assertEqual(len(b), 32)
327 message_array_vla = VPPMessage('address_array_vla',
329 ['vl_api_ip6_address_t',
330 'addresses', 0, 'num']])
331 b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
332 self.assertEqual(len(b), 36)
334 message_array4 = VPPMessage('address_array4',
335 [['vl_api_ip4_address_t',
337 b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
338 self.assertEqual(len(b), 8)
339 b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
341 self.assertEqual(len(b), 8)
343 message = VPPMessage('address', [['vl_api_address_t', 'address']])
344 b = message.pack({'address': '1::1'})
345 self.assertEqual(len(b), 20)
346 b = message.pack({'address': '1.1.1.1'})
347 self.assertEqual(len(b), 20)
348 message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
349 b = message.pack({'prefix': '1::1/130'})
350 self.assertEqual(len(b), 21)
351 b = message.pack({'prefix': IPv6Network(u'1::/119')})
352 self.assertEqual(len(b), 21)
353 b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
354 self.assertEqual(len(b), 21)
356 def test_zero_vla(self):
357 '''Default zero'ed out for VLAs'''
358 list = VPPType('vl_api_list_t',
359 [['u8', 'count', 10]])
361 # Define an embedded VLA type
362 valist = VPPType('vl_api_valist_t',
364 ['u8', 'string', 0, 'count']])
366 vamessage = VPPMessage('vamsg',
367 [['vl_api_valist_t', 'valist'],
368 ['u8', 'is_something']])
370 message = VPPMessage('msg',
371 [['vl_api_list_t', 'list'],
372 ['u8', 'is_something']])
374 # Pack message without VLA specified
375 b = message.pack({'is_something': 1})
376 b = vamessage.pack({'is_something': 1})
378 def test_arrays(self):
381 # 2. Fixed list of variable length sub type
382 # 3. Variable length type
384 s = VPPType('str', [['u32', 'length'],
385 ['u8', 'string', 0, 'length']])
387 ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
388 listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
389 valistip4 = VPPType('list_ip4_t',
391 ['ip4_address', 'addresses', 0, 'count']])
393 valistip4_legacy = VPPType('list_ip4_t',
395 ['ip4_address', 'addresses', 0]])
399 addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
400 b = listip4.pack({'addresses': addresses})
401 self.assertEqual(len(b), 16)
402 nt, size = listip4.unpack(b)
403 self.assertEqual(nt.addresses[0].address,
404 inet_pton(AF_INET, '2.2.2.2'))
406 b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
407 self.assertEqual(len(b), 17)
409 nt, size = valistip4.unpack(b)
410 self.assertEqual(nt.count, 4)
411 self.assertEqual(nt.addresses[0].address,
412 inet_pton(AF_INET, '2.2.2.2'))
414 b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
415 self.assertEqual(len(b), 17)
416 nt, size = valistip4_legacy.unpack(b)
417 self.assertEqual(len(nt.addresses), 4)
418 self.assertEqual(nt.addresses[0].address,
419 inet_pton(AF_INET, '2.2.2.2'))
421 string = 'foobar foobar'
422 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
423 nt, size = s.unpack(b)
424 self.assertEqual(len(b), size)
426 def test_string(self):
427 s = VPPType('str', [['u32', 'length'],
428 ['u8', 'string', 0, 'length']])
431 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
432 nt, size = s.unpack(b)
433 self.assertEqual(len(b), size)
435 def test_message(self):
436 foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
437 ['u8', 'client_index'],
439 {"crc": "0x559b9f3c"}])
440 b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
442 nt, size = foo.unpack(b)
443 self.assertEqual(len(b), size)
444 self.assertEqual(nt.something, 200)
448 fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
449 [['u8', 'is_uniform'],
454 label_stack = {'is_uniform': 0,
459 b = fib_mpls_label.pack(label_stack)
460 self.assertEqual(len(b), 7)
462 fib_path = VPPType('vl_api_fib_path_t',
463 [['u32', 'sw_if_index'],
466 ['u8', 'preference'],
469 ['u8', 'is_udp_encap'],
470 ['u8', 'is_unreach'],
471 ['u8', 'is_prohibit'],
472 ['u8', 'is_resolve_host'],
473 ['u8', 'is_resolve_attached'],
475 ['u8', 'is_source_lookup'],
477 ['u8', 'next_hop', 16],
478 ['u32', 'next_hop_id'],
480 ['u32', 'via_label'],
482 ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
483 label_stack_list = []
485 label_stack_list.append(label_stack)
487 paths = {'is_udp_encap': 0,
488 'next_hop': b'\x10\x02\x02\xac',
492 'next_hop_id': 4294967295,
493 'label_stack': label_stack_list,
495 'sw_if_index': 4294967295,
498 b = fib_path.pack(paths)
499 self.assertEqual(len(b), (7*16) + 49)
501 abf_policy = VPPType('vl_api_abf_policy_t',
502 [['u32', 'policy_id'],
503 ['u32', 'acl_index'],
505 ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
513 b = abf_policy.pack(policy)
514 self.assertEqual(len(b), (7*16) + 49 + 9)
516 abf_policy_add_del = VPPMessage('abf_policy_add_del',
517 [['u16', '_vl_msg_id'],
518 ['u32', 'client_index'],
521 ['vl_api_abf_policy_t', 'policy']])
523 b = abf_policy_add_del.pack({'is_add': 1,
528 nt, size = abf_policy_add_del.unpack(b)
529 self.assertEqual(nt.policy.paths[0].next_hop,
530 b'\x10\x02\x02\xac\x00\x00\x00\x00'
531 b'\x00\x00\x00\x00\x00\x00\x00\x00')
535 bier_table_id = VPPType('vl_api_bier_table_id_t',
537 ['u8', 'bt_sub_domain'],
538 ['u8', 'bt_hdr_len_id']])
540 bier_imp_add = VPPMessage('bier_imp_add',
541 [['u32', 'client_index'],
543 ['vl_api_bier_table_id_t', 'bi_tbl_id'],
545 ['u8', 'bi_n_bytes'],
546 ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
548 table_id = {'bt_set': 0,
554 b = bier_imp_add.pack({'bi_tbl_id': table_id,
555 'bi_n_bytes': len(bibytes),
556 'bi_bytes': bibytes})
558 self.assertEqual(len(b), 20)
562 VPPEnumType('vl_api_eid_type_t',
563 [["EID_TYPE_API_PREFIX", 0],
564 ["EID_TYPE_API_MAC", 1],
565 ["EID_TYPE_API_NSH", 2],
566 {"enumtype": "u32"}])
568 VPPTypeAlias('vl_api_mac_address_t', {'type': 'u8',
571 VPPType('vl_api_nsh_t',
575 VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
577 {"enumtype": "u32"}])
578 VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
580 VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
582 VPPUnionType('vl_api_address_union_t',
583 [["vl_api_ip4_address_t", "ip4"],
584 ["vl_api_ip6_address_t", "ip6"]])
586 VPPType('vl_api_address_t',
587 [['vl_api_address_family_t', 'af'],
588 ['vl_api_address_union_t', 'un']])
590 VPPType('vl_api_prefix_t',
591 [['vl_api_address_t', 'address'],
594 VPPUnionType('vl_api_eid_address_t',
595 [["vl_api_prefix_t", "prefix"],
596 ["vl_api_mac_address_t", "mac"],
597 ["vl_api_nsh_t", "nsh"]])
599 eid = VPPType('vl_api_eid_t',
600 [["vl_api_eid_type_t", "type"],
601 ["vl_api_eid_address_t", "address"]])
603 b = eid.pack({'type':1,
605 'mac': MACAddress('aa:bb:cc:dd:ee:ff')}})
606 self.assertEqual(len(b), 25)
607 nt, size = eid.unpack(b)
608 self.assertEqual(str(nt.address.mac), 'aa:bb:cc:dd:ee:ff')
609 self.assertIsNone(nt.address.prefix)
612 if __name__ == '__main__':