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'))
51 limited_type = VPPType('limited_type_t',
52 [['string', 'name', 0, {'limit': 16}]])
53 unlimited_type = VPPType('limited_type_t',
54 [['string', 'name', 0]])
56 b = limited_type.pack({'name': 'foobar'})
57 self.assertEqual(len(b), 10)
58 b = unlimited_type.pack({'name': 'foobar'})
59 self.assertEqual(len(b), 10)
61 with self.assertRaises(VPPSerializerValueError):
62 b = limited_type.pack({'name': 'foobar'*3})
65 class TestDefaults(unittest.TestCase):
66 def test_defaults(self):
67 default_type = VPPType('default_type_t',
68 [['u16', 'mtu', {'default': 1500, 'limit': 0}]])
69 without_default_type = VPPType('without_default_type_t',
72 b = default_type.pack({})
73 self.assertEqual(len(b), 2)
74 nt, size = default_type.unpack(b)
75 self.assertEqual(len(b), size)
76 self.assertEqual(nt.mtu, 1500)
78 # distinguish between parameter 0 and parameter not passed
79 b = default_type.pack({'mtu': 0})
80 self.assertEqual(len(b), 2)
81 nt, size = default_type.unpack(b)
82 self.assertEqual(len(b), size)
83 self.assertEqual(nt.mtu, 0)
85 # Ensure that basetypes does not inherit default
86 b = without_default_type.pack({})
87 self.assertEqual(len(b), 2)
88 nt, size = default_type.unpack(b)
89 self.assertEqual(len(b), size)
90 self.assertEqual(nt.mtu, 0)
93 VPPEnumType('vl_api_enum_t', [["ADDRESS_IP4", 0],
97 default_with_enum = VPPType('default_enum_type_t',
98 [['u16', 'mtu'], ['vl_api_enum_t',
99 'e', {'default': 1}]])
101 b = default_with_enum.pack({})
102 self.assertEqual(len(b), 6)
103 nt, size = default_with_enum.unpack(b)
104 self.assertEqual(len(b), size)
105 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']])
204 prefix = VPPType('vl_api_prefix_t',
205 [['vl_api_address_t', 'address'],
207 prefix4 = VPPType('vl_api_ip4_prefix_t',
208 [['vl_api_ip4_address_t', 'address'],
210 prefix6 = VPPType('vl_api_ip6_prefix_t',
211 [['vl_api_ip6_address_t', 'address'],
214 address_with_prefix = VPPTypeAlias('vl_api_address_with_prefix_t', {'type': 'vl_api_prefix_t' })
215 address4_with_prefix = VPPTypeAlias('vl_api_ip4_address_with_prefix_t',
216 {'type': 'vl_api_ip4_prefix_t' })
217 address6_with_prefix = VPPTypeAlias('vl_api_ip6_address_with_prefix_t',
218 {'type': 'vl_api_ip6_prefix_t' })
220 awp_type = VPPType('foobar_t',
221 [['vl_api_address_with_prefix_t', 'address']])
223 # address with prefix
224 b = address_with_prefix.pack(IPv4Interface('2.2.2.2/24'))
225 self.assertEqual(len(b), 21)
226 nt, size = address_with_prefix.unpack(b)
227 self.assertTrue(isinstance(nt, IPv4Interface))
228 self.assertEqual(str(nt), '2.2.2.2/24')
230 b = address_with_prefix.pack(IPv6Interface('2::2/64'))
231 self.assertEqual(len(b), 21)
232 nt, size = address_with_prefix.unpack(b)
233 self.assertTrue(isinstance(nt, IPv6Interface))
234 self.assertEqual(str(nt), '2::2/64')
236 b = address_with_prefix.pack(IPv4Network('2.2.2.2/24', strict=False))
237 self.assertEqual(len(b), 21)
238 nt, size = address_with_prefix.unpack(b)
239 self.assertTrue(isinstance(nt, IPv4Interface))
240 self.assertEqual(str(nt), '2.2.2.0/24')
242 b = address4_with_prefix.pack('2.2.2.2/24')
243 self.assertEqual(len(b), 5)
244 nt, size = address4_with_prefix.unpack(b)
245 self.assertTrue(isinstance(nt, IPv4Interface))
246 self.assertEqual(str(nt), '2.2.2.2/24')
247 b = address4_with_prefix.pack(IPv4Interface('2.2.2.2/24'))
248 self.assertEqual(len(b), 5)
250 b = address6_with_prefix.pack('2::2/64')
251 self.assertEqual(len(b), 17)
252 nt, size = address6_with_prefix.unpack(b)
253 self.assertTrue(isinstance(nt, IPv6Interface))
254 self.assertEqual(str(nt), '2::2/64')
255 b = address6_with_prefix.pack(IPv6Interface('2::2/64'))
256 self.assertEqual(len(b), 17)
258 b = prefix.pack('192.168.10.0/24')
259 self.assertEqual(len(b), 21)
260 nt, size = prefix.unpack(b)
261 self.assertTrue(isinstance(nt, IPv4Network))
262 self.assertEqual(str(nt), '192.168.10.0/24')
264 b = awp_type.pack({'address': '1.2.3.4/24'})
265 self.assertEqual(len(b), 21)
266 nt, size = awp_type.unpack(b)
267 self.assertTrue(isinstance(nt.address, IPv4Interface))
268 self.assertEqual(str(nt.address), '1.2.3.4/24')
270 b = awp_type.pack({'address': IPv4Interface('1.2.3.4/24')})
271 self.assertEqual(len(b), 21)
272 nt, size = awp_type.unpack(b)
273 self.assertTrue(isinstance(nt.address, IPv4Interface))
274 self.assertEqual(str(nt.address), '1.2.3.4/24')
276 def test_recursive_address(self):
277 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
279 {"enumtype": "u32"}])
280 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
282 b = ip4.pack('1.1.1.1')
283 self.assertEqual(len(b), 4)
284 nt, size = ip4.unpack(b)
286 self.assertEqual(str(nt), '1.1.1.1')
288 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
290 VPPUnionType('vl_api_address_union_t',
291 [["vl_api_ip4_address_t", "ip4"],
292 ["vl_api_ip6_address_t", "ip6"]])
294 address = VPPType('vl_api_address_t',
295 [['vl_api_address_family_t', 'af'],
296 ['vl_api_address_union_t', 'un']])
298 prefix = VPPType('vl_api_prefix_t',
299 [['vl_api_address_t', 'address'],
301 message = VPPMessage('svs',
302 [['vl_api_prefix_t', 'prefix']])
303 message_addr = VPPMessage('svs_address',
304 [['vl_api_address_t', 'address']])
306 b = message_addr.pack({'address': "1::1"})
307 self.assertEqual(len(b), 20)
308 nt, size = message_addr.unpack(b)
309 self.assertEqual("1::1", str(nt.address))
310 b = message_addr.pack({'address': "1.1.1.1"})
311 self.assertEqual(len(b), 20)
312 nt, size = message_addr.unpack(b)
313 self.assertEqual("1.1.1.1", str(nt.address))
315 b = message.pack({'prefix': "1.1.1.0/24"})
316 self.assertEqual(len(b), 21)
317 nt, size = message.unpack(b)
318 self.assertEqual("1.1.1.0/24", str(nt.prefix))
320 message_array = VPPMessage('address_array',
321 [['vl_api_ip6_address_t',
323 b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
324 self.assertEqual(len(b), 32)
325 message_array_vla = VPPMessage('address_array_vla',
327 ['vl_api_ip6_address_t',
328 'addresses', 0, 'num']])
329 b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
330 self.assertEqual(len(b), 36)
332 message_array4 = VPPMessage('address_array4',
333 [['vl_api_ip4_address_t',
335 b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
336 self.assertEqual(len(b), 8)
337 b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
339 self.assertEqual(len(b), 8)
341 message = VPPMessage('address', [['vl_api_address_t', 'address']])
342 b = message.pack({'address': '1::1'})
343 self.assertEqual(len(b), 20)
344 b = message.pack({'address': '1.1.1.1'})
345 self.assertEqual(len(b), 20)
346 message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
347 b = message.pack({'prefix': '1::1/130'})
348 self.assertEqual(len(b), 21)
349 b = message.pack({'prefix': IPv6Network(u'1::/119')})
350 self.assertEqual(len(b), 21)
351 b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
352 self.assertEqual(len(b), 21)
354 def test_zero_vla(self):
355 '''Default zero'ed out for VLAs'''
356 list = VPPType('vl_api_list_t',
357 [['u8', 'count', 10]])
359 # Define an embedded VLA type
360 valist = VPPType('vl_api_valist_t',
362 ['u8', 'string', 0, 'count']])
364 vamessage = VPPMessage('vamsg',
365 [['vl_api_valist_t', 'valist'],
366 ['u8', 'is_something']])
368 message = VPPMessage('msg',
369 [['vl_api_list_t', 'list'],
370 ['u8', 'is_something']])
372 # Pack message without VLA specified
373 b = message.pack({'is_something': 1})
374 b = vamessage.pack({'is_something': 1})
376 def test_arrays(self):
379 # 2. Fixed list of variable length sub type
380 # 3. Variable length type
382 s = VPPType('str', [['u32', 'length'],
383 ['u8', 'string', 0, 'length']])
385 ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
386 listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
387 valistip4 = VPPType('list_ip4_t',
389 ['ip4_address', 'addresses', 0, 'count']])
391 valistip4_legacy = VPPType('list_ip4_t',
393 ['ip4_address', 'addresses', 0]])
397 addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
398 b = listip4.pack({'addresses': addresses})
399 self.assertEqual(len(b), 16)
400 nt, size = listip4.unpack(b)
401 self.assertEqual(nt.addresses[0].address,
402 inet_pton(AF_INET, '2.2.2.2'))
404 b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
405 self.assertEqual(len(b), 17)
407 nt, size = valistip4.unpack(b)
408 self.assertEqual(nt.count, 4)
409 self.assertEqual(nt.addresses[0].address,
410 inet_pton(AF_INET, '2.2.2.2'))
412 b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
413 self.assertEqual(len(b), 17)
414 nt, size = valistip4_legacy.unpack(b)
415 self.assertEqual(len(nt.addresses), 4)
416 self.assertEqual(nt.addresses[0].address,
417 inet_pton(AF_INET, '2.2.2.2'))
419 string = 'foobar foobar'
420 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
421 nt, size = s.unpack(b)
422 self.assertEqual(len(b), size)
424 def test_string(self):
425 s = VPPType('str', [['u32', 'length'],
426 ['u8', 'string', 0, 'length']])
429 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
430 nt, size = s.unpack(b)
431 self.assertEqual(len(b), size)
433 def test_message(self):
434 foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
435 ['u8', 'client_index'],
437 {"crc": "0x559b9f3c"}])
438 b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
440 nt, size = foo.unpack(b)
441 self.assertEqual(len(b), size)
442 self.assertEqual(nt.something, 200)
446 fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
447 [['u8', 'is_uniform'],
452 label_stack = {'is_uniform': 0,
457 b = fib_mpls_label.pack(label_stack)
458 self.assertEqual(len(b), 7)
460 fib_path = VPPType('vl_api_fib_path_t',
461 [['u32', 'sw_if_index'],
464 ['u8', 'preference'],
467 ['u8', 'is_udp_encap'],
468 ['u8', 'is_unreach'],
469 ['u8', 'is_prohibit'],
470 ['u8', 'is_resolve_host'],
471 ['u8', 'is_resolve_attached'],
473 ['u8', 'is_source_lookup'],
475 ['u8', 'next_hop', 16],
476 ['u32', 'next_hop_id'],
478 ['u32', 'via_label'],
480 ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
481 label_stack_list = []
483 label_stack_list.append(label_stack)
485 paths = {'is_udp_encap': 0,
486 'next_hop': b'\x10\x02\x02\xac',
490 'next_hop_id': 4294967295,
491 'label_stack': label_stack_list,
493 'sw_if_index': 4294967295,
496 b = fib_path.pack(paths)
497 self.assertEqual(len(b), (7*16) + 49)
499 abf_policy = VPPType('vl_api_abf_policy_t',
500 [['u32', 'policy_id'],
501 ['u32', 'acl_index'],
503 ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
511 b = abf_policy.pack(policy)
512 self.assertEqual(len(b), (7*16) + 49 + 9)
514 abf_policy_add_del = VPPMessage('abf_policy_add_del',
515 [['u16', '_vl_msg_id'],
516 ['u32', 'client_index'],
519 ['vl_api_abf_policy_t', 'policy']])
521 b = abf_policy_add_del.pack({'is_add': 1,
526 nt, size = abf_policy_add_del.unpack(b)
527 self.assertEqual(nt.policy.paths[0].next_hop,
528 b'\x10\x02\x02\xac\x00\x00\x00\x00'
529 b'\x00\x00\x00\x00\x00\x00\x00\x00')
533 bier_table_id = VPPType('vl_api_bier_table_id_t',
535 ['u8', 'bt_sub_domain'],
536 ['u8', 'bt_hdr_len_id']])
538 bier_imp_add = VPPMessage('bier_imp_add',
539 [['u32', 'client_index'],
541 ['vl_api_bier_table_id_t', 'bi_tbl_id'],
543 ['u8', 'bi_n_bytes'],
544 ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
546 table_id = {'bt_set': 0,
552 b = bier_imp_add.pack({'bi_tbl_id': table_id,
553 'bi_n_bytes': len(bibytes),
554 'bi_bytes': bibytes})
556 self.assertEqual(len(b), 20)
559 VPPEnumType('vl_api_eid_type_t',
560 [["EID_TYPE_API_PREFIX", 0],
561 ["EID_TYPE_API_MAC", 1],
562 ["EID_TYPE_API_NSH", 2],
563 {"enumtype": "u32"}])
565 VPPTypeAlias('vl_api_mac_address_t', {'type': 'u8',
568 VPPType('vl_api_nsh_t',
572 VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
574 {"enumtype": "u32"}])
575 VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
577 VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
579 VPPUnionType('vl_api_address_union_t',
580 [["vl_api_ip4_address_t", "ip4"],
581 ["vl_api_ip6_address_t", "ip6"]])
583 VPPType('vl_api_address_t',
584 [['vl_api_address_family_t', 'af'],
585 ['vl_api_address_union_t', 'un']])
587 VPPType('vl_api_prefix_t',
588 [['vl_api_address_t', 'address'],
591 VPPUnionType('vl_api_eid_address_t',
592 [["vl_api_prefix_t", "prefix"],
593 ["vl_api_mac_address_t", "mac"],
594 ["vl_api_nsh_t", "nsh"]])
596 eid = VPPType('vl_api_eid_t',
597 [["vl_api_eid_type_t", "type"],
598 ["vl_api_eid_address_t", "address"]])
600 b = eid.pack({'type':1,
602 'mac': MACAddress('aa:bb:cc:dd:ee:ff')}})
603 self.assertEqual(len(b), 25)
604 nt, size = eid.unpack(b)
605 self.assertEqual(str(nt.address.mac), 'aa:bb:cc:dd:ee:ff')
606 self.assertIsNone(nt.address.prefix)
609 class TestVppSerializerLogging(unittest.TestCase):
611 def test_logger(self):
612 # test logger name 'vpp_papi.serializer'
613 with self.assertRaises(VPPSerializerValueError) as ctx:
614 with self.assertLogs('vpp_papi.serializer', level='DEBUG') as cm:
615 u = VPPUnionType('vl_api_eid_address_t',
616 [["vl_api_prefix_t", "prefix"],
617 ["vl_api_mac_address_t", "mac"],
618 ["vl_api_nsh_t", "nsh"]])
619 self.assertEqual(cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"])
621 # test parent logger name 'vpp_papi'
622 with self.assertRaises(VPPSerializerValueError) as ctx:
623 with self.assertLogs('vpp_papi', level='DEBUG') as cm:
624 u = VPPUnionType('vl_api_eid_address_t',
625 [["vl_api_prefix_t", "prefix"],
626 ["vl_api_mac_address_t", "mac"],
627 ["vl_api_nsh_t", "nsh"]])
628 self.assertEqual(cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"])
631 if __name__ == '__main__':