5221034da6cbd8347e8523c8e00d6c8c6515ab2d
[vpp.git] / src / vpp-api / python / vpp_papi / tests / test_vpp_serializer.py
1 #!/usr/bin/env python
2
3 import unittest
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 socket import inet_pton, AF_INET, AF_INET6
8 import logging
9 import sys
10 from ipaddress import *
11
12
13 class TestLimits(unittest.TestCase):
14     def test_string(self):
15         fixed_string = VPPType('fixed_string',
16                                [['string', 'name', 16]])
17
18         b = fixed_string.pack({'name': 'foobar'})
19         self.assertEqual(len(b), 16)
20
21         # Ensure string is nul terminated
22         self.assertEqual(b.decode('ascii')[6], '\x00')
23
24         nt, size = fixed_string.unpack(b)
25         self.assertEqual(size, 16)
26         self.assertEqual(nt.name, 'foobar')
27
28         # Empty string
29         b = fixed_string.pack({'name': ''})
30         self.assertEqual(len(b), 16)
31         nt, size = fixed_string.unpack(b)
32         self.assertEqual(size, 16)
33         self.assertEqual(nt.name, '')
34
35         # String too long
36         with self.assertRaises(VPPSerializerValueError):
37             b = fixed_string.pack({'name': 'foobarfoobar1234'})
38
39         variable_string = VPPType('variable_string',
40                                   [['string', 'name', 0]])
41         b = variable_string.pack({'name': 'foobar'})
42         self.assertEqual(len(b), 4 + len('foobar'))
43
44         nt, size = variable_string.unpack(b)
45         self.assertEqual(size, 4 + len('foobar'))
46         self.assertEqual(nt.name, 'foobar')
47         self.assertEqual(len(nt.name), len('foobar'))
48
49
50     def test_limit(self):
51         limited_type = VPPType('limited_type_t',
52                                [['string', 'name', 0, {'limit': 16}]])
53         unlimited_type = VPPType('limited_type_t',
54                                  [['string', 'name', 0]])
55
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)
60
61         with self.assertRaises(VPPSerializerValueError):
62             b = limited_type.pack({'name': 'foobar'*3})
63
64
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',
70                                        [['u16', 'mtu']])
71
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)
77
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)
84
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)
91
92         # default enum type
93         VPPEnumType('vl_api_enum_t', [["ADDRESS_IP4", 0],
94                                       ["ADDRESS_IP6", 1],
95                                       {"enumtype": "u32"}])
96
97         default_with_enum = VPPType('default_enum_type_t',
98                                     [['u16', 'mtu'], ['vl_api_enum_t',
99                                                       'e', {'default': 1}]])
100
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)
106
107 class TestAddType(unittest.TestCase):
108
109     def test_union(self):
110         un = VPPUnionType('test_union',
111                           [['u8', 'is_bool'],
112                            ['u32', 'is_int']])
113
114         b = un.pack({'is_int': 0x12345678})
115         nt, size = un.unpack(b)
116         self.assertEqual(len(b), size)
117         self.assertEqual(nt.is_bool, 0x12)
118         self.assertEqual(nt.is_int, 0x12345678)
119
120     def test_address(self):
121         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
122                                                      ["ADDRESS_IP6", 1],
123                                                      {"enumtype": "u32"}])
124         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
125                                                     'length': 4})
126         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
127                                                     'length': 16})
128         VPPUnionType('vl_api_address_union_t',
129                      [["vl_api_ip4_address_t", "ip4"],
130                       ["vl_api_ip6_address_t", "ip6"]])
131
132         address = VPPType('vl_api_address_t',
133                           [['vl_api_address_family_t', 'af'],
134                            ['vl_api_address_union_t', 'un']])
135
136         prefix = VPPType('vl_api_prefix_t',
137                          [['vl_api_address_t', 'address'],
138                           ['u8', 'len']])
139
140         va_address_list = VPPType('list_addresses',
141                                   [['u8', 'count'],
142                                    ['vl_api_address_t', 'addresses',
143                                     0, 'count']])
144
145         message_with_va_address_list = VPPType('msg_with_vla',
146                                                [['list_addresses',
147                                                  'vla_address'],
148                                                 ['u8', 'is_cool']])
149
150         b = ip4.pack(inet_pton(AF_INET, '1.1.1.1'))
151         self.assertEqual(len(b), 4)
152         nt, size = ip4.unpack(b)
153         self.assertEqual(str(nt), '1.1.1.1')
154
155         b = ip6.pack(inet_pton(AF_INET6, '1::1'))
156         self.assertEqual(len(b), 16)
157
158         b = address.pack({'af': af.ADDRESS_IP4,
159                           'un':
160                           {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
161         self.assertEqual(len(b), 20)
162
163         nt, size = address.unpack(b)
164         self.assertEqual(str(nt), '2.2.2.2')
165
166         # List of addresses
167         address_list = []
168         for i in range(4):
169             address_list.append({'af': af.ADDRESS_IP4,
170                                  'un':
171                                  {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
172         b = va_address_list.pack({'count': len(address_list),
173                                   'addresses': address_list})
174         self.assertEqual(len(b), 81)
175
176         nt, size = va_address_list.unpack(b)
177         self.assertEqual(str(nt.addresses[0]), '2.2.2.2')
178
179         b = message_with_va_address_list.pack({'vla_address':
180                                                {'count': len(address_list),
181                                                 'addresses': address_list},
182                                                'is_cool': 100})
183         self.assertEqual(len(b), 82)
184         nt, size = message_with_va_address_list.unpack(b)
185         self.assertEqual(nt.is_cool, 100)
186
187     def test_address_with_prefix(self):
188         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
189                                                      ["ADDRESS_IP6", 1],
190                                                      {"enumtype": "u32"}])
191         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
192                                                     'length': 4})
193         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
194                                                     'length': 16})
195         VPPUnionType('vl_api_address_union_t',
196                      [["vl_api_ip4_address_t", "ip4"],
197                       ["vl_api_ip6_address_t", "ip6"]])
198
199         address = VPPType('vl_api_address_t',
200                           [['vl_api_address_family_t', 'af'],
201                            ['vl_api_address_union_t', 'un']])
202
203
204         prefix = VPPType('vl_api_prefix_t',
205                          [['vl_api_address_t', 'address'],
206                           ['u8', 'len']])
207         prefix4 = VPPType('vl_api_ip4_prefix_t',
208                           [['vl_api_ip4_address_t', 'address'],
209                           ['u8', 'len']])
210         prefix6 = VPPType('vl_api_ip6_prefix_t',
211                           [['vl_api_ip6_address_t', 'address'],
212                           ['u8', 'len']])
213
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' })
219
220         awp_type = VPPType('foobar_t',
221                            [['vl_api_address_with_prefix_t', 'address']])
222
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')
229
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')
235
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')
241
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)
249
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)
257
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')
263
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')
269
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')
275
276
277     def test_recursive_address(self):
278         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
279                                                      ["ADDRESS_IP6", 1],
280                                                      {"enumtype": "u32"}])
281         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
282                                                     'length': 4})
283         b = ip4.pack('1.1.1.1')
284         self.assertEqual(len(b), 4)
285         nt, size = ip4.unpack(b)
286
287         self.assertEqual(str(nt), '1.1.1.1')
288
289         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
290                                                     'length': 16})
291         VPPUnionType('vl_api_address_union_t',
292                      [["vl_api_ip4_address_t", "ip4"],
293                       ["vl_api_ip6_address_t", "ip6"]])
294
295         address = VPPType('vl_api_address_t',
296                           [['vl_api_address_family_t', 'af'],
297                            ['vl_api_address_union_t', 'un']])
298
299         prefix = VPPType('vl_api_prefix_t',
300                          [['vl_api_address_t', 'address'],
301                           ['u8', 'len']])
302         message = VPPMessage('svs',
303                              [['vl_api_prefix_t', 'prefix']])
304         message_addr = VPPMessage('svs_address',
305                                   [['vl_api_address_t', 'address']])
306
307         b = message_addr.pack({'address': "1::1"})
308         self.assertEqual(len(b), 20)
309         nt, size = message_addr.unpack(b)
310         self.assertEqual("1::1", str(nt.address))
311         b = message_addr.pack({'address': "1.1.1.1"})
312         self.assertEqual(len(b), 20)
313         nt, size = message_addr.unpack(b)
314         self.assertEqual("1.1.1.1", str(nt.address))
315
316         b = message.pack({'prefix': "1.1.1.0/24"})
317         self.assertEqual(len(b), 21)
318         nt, size = message.unpack(b)
319         self.assertEqual("1.1.1.0/24", str(nt.prefix))
320
321         message_array = VPPMessage('address_array',
322                                    [['vl_api_ip6_address_t',
323                                      'addresses', 2]])
324         b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
325         self.assertEqual(len(b), 32)
326         message_array_vla = VPPMessage('address_array_vla',
327                                        [['u32', 'num'],
328                                         ['vl_api_ip6_address_t',
329                                          'addresses', 0, 'num']])
330         b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
331         self.assertEqual(len(b), 36)
332
333         message_array4 = VPPMessage('address_array4',
334                                     [['vl_api_ip4_address_t',
335                                       'addresses', 2]])
336         b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
337         self.assertEqual(len(b), 8)
338         b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
339                                                "2.2.2.2"]})
340         self.assertEqual(len(b), 8)
341
342         message = VPPMessage('address', [['vl_api_address_t', 'address']])
343         b = message.pack({'address': '1::1'})
344         self.assertEqual(len(b), 20)
345         b = message.pack({'address': '1.1.1.1'})
346         self.assertEqual(len(b), 20)
347         message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
348         b = message.pack({'prefix': '1::1/130'})
349         self.assertEqual(len(b), 21)
350         b = message.pack({'prefix': IPv6Network(u'1::/119')})
351         self.assertEqual(len(b), 21)
352         b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
353         self.assertEqual(len(b), 21)
354
355     def test_zero_vla(self):
356         '''Default zero'ed out for VLAs'''
357         list = VPPType('vl_api_list_t',
358                        [['u8', 'count', 10]])
359
360         # Define an embedded VLA type
361         valist = VPPType('vl_api_valist_t',
362                          [['u8', 'count'],
363                           ['u8', 'string', 0, 'count']])
364         # Define a message
365         vamessage = VPPMessage('vamsg',
366                                [['vl_api_valist_t', 'valist'],
367                                 ['u8', 'is_something']])
368
369         message = VPPMessage('msg',
370                              [['vl_api_list_t', 'list'],
371                               ['u8', 'is_something']])
372
373         # Pack message without VLA specified
374         b = message.pack({'is_something': 1})
375         b = vamessage.pack({'is_something': 1})
376
377     def test_arrays(self):
378         # Test cases
379         # 1. Fixed list
380         # 2. Fixed list of variable length sub type
381         # 3. Variable length type
382         #
383         s = VPPType('str', [['u32', 'length'],
384                             ['u8', 'string', 0, 'length']])
385
386         ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
387         listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
388         valistip4 = VPPType('list_ip4_t',
389                             [['u8', 'count'],
390                              ['ip4_address', 'addresses', 0, 'count']])
391
392         valistip4_legacy = VPPType('list_ip4_t',
393                                    [['u8', 'foo'],
394                                     ['ip4_address', 'addresses', 0]])
395
396         addresses = []
397         for i in range(4):
398             addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
399         b = listip4.pack({'addresses': addresses})
400         self.assertEqual(len(b), 16)
401         nt, size = listip4.unpack(b)
402         self.assertEqual(nt.addresses[0].address,
403                          inet_pton(AF_INET, '2.2.2.2'))
404
405         b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
406         self.assertEqual(len(b), 17)
407
408         nt, size = valistip4.unpack(b)
409         self.assertEqual(nt.count, 4)
410         self.assertEqual(nt.addresses[0].address,
411                          inet_pton(AF_INET, '2.2.2.2'))
412
413         b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
414         self.assertEqual(len(b), 17)
415         nt, size = valistip4_legacy.unpack(b)
416         self.assertEqual(len(nt.addresses), 4)
417         self.assertEqual(nt.addresses[0].address,
418                          inet_pton(AF_INET, '2.2.2.2'))
419
420         string = 'foobar foobar'
421         b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
422         nt, size = s.unpack(b)
423         self.assertEqual(len(b), size)
424
425     def test_string(self):
426         s = VPPType('str', [['u32', 'length'],
427                             ['u8', 'string', 0, 'length']])
428
429         string = ''
430         b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
431         nt, size = s.unpack(b)
432         self.assertEqual(len(b), size)
433
434     def test_message(self):
435         foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
436                                  ['u8', 'client_index'],
437                                  ['u8', 'something'],
438                                  {"crc": "0x559b9f3c"}])
439         b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
440                       'something': 200})
441         nt, size = foo.unpack(b)
442         self.assertEqual(len(b), size)
443         self.assertEqual(nt.something, 200)
444
445     def test_abf(self):
446
447         fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
448                                  [['u8', 'is_uniform'],
449                                   ['u32', 'label'],
450                                   ['u8', 'ttl'],
451                                   ['u8', 'exp']])
452
453         label_stack = {'is_uniform': 0,
454                        'label': 0,
455                        'ttl': 0,
456                        'exp': 0}
457
458         b = fib_mpls_label.pack(label_stack)
459         self.assertEqual(len(b), 7)
460
461         fib_path = VPPType('vl_api_fib_path_t',
462                            [['u32', 'sw_if_index'],
463                             ['u32', 'table_id'],
464                             ['u8', 'weight'],
465                             ['u8', 'preference'],
466                             ['u8', 'is_local'],
467                             ['u8', 'is_drop'],
468                             ['u8', 'is_udp_encap'],
469                             ['u8', 'is_unreach'],
470                             ['u8', 'is_prohibit'],
471                             ['u8', 'is_resolve_host'],
472                             ['u8', 'is_resolve_attached'],
473                             ['u8', 'is_dvr'],
474                             ['u8', 'is_source_lookup'],
475                             ['u8', 'afi'],
476                             ['u8', 'next_hop', 16],
477                             ['u32', 'next_hop_id'],
478                             ['u32', 'rpf_id'],
479                             ['u32', 'via_label'],
480                             ['u8', 'n_labels'],
481                             ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
482         label_stack_list = []
483         for i in range(16):
484             label_stack_list.append(label_stack)
485
486         paths = {'is_udp_encap': 0,
487                  'next_hop': b'\x10\x02\x02\xac',
488                  'table_id': 0,
489                  'afi': 0,
490                  'weight': 1,
491                  'next_hop_id': 4294967295,
492                  'label_stack': label_stack_list,
493                  'n_labels': 0,
494                  'sw_if_index': 4294967295,
495                  'preference': 0}
496
497         b = fib_path.pack(paths)
498         self.assertEqual(len(b), (7*16) + 49)
499
500         abf_policy = VPPType('vl_api_abf_policy_t',
501                              [['u32', 'policy_id'],
502                               ['u32', 'acl_index'],
503                               ['u8', 'n_paths'],
504                               ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
505
506         policy = {
507             'n_paths': 1,
508             'paths': [paths],
509             'acl_index': 0,
510             'policy_id': 10}
511
512         b = abf_policy.pack(policy)
513         self.assertEqual(len(b), (7*16) + 49 + 9)
514
515         abf_policy_add_del = VPPMessage('abf_policy_add_del',
516                                         [['u16', '_vl_msg_id'],
517                                          ['u32', 'client_index'],
518                                          ['u32', 'context'],
519                                          ['u8', 'is_add'],
520                                          ['vl_api_abf_policy_t', 'policy']])
521
522         b = abf_policy_add_del.pack({'is_add': 1,
523                                      'context': 66,
524                                      '_vl_msg_id': 1066,
525                                      'policy': policy})
526
527         nt, size = abf_policy_add_del.unpack(b)
528         self.assertEqual(nt.policy.paths[0].next_hop,
529                          b'\x10\x02\x02\xac\x00\x00\x00\x00'
530                          b'\x00\x00\x00\x00\x00\x00\x00\x00')
531
532     def test_bier(self):
533
534         bier_table_id = VPPType('vl_api_bier_table_id_t',
535                                 [['u8', 'bt_set'],
536                                  ['u8', 'bt_sub_domain'],
537                                  ['u8', 'bt_hdr_len_id']])
538
539         bier_imp_add = VPPMessage('bier_imp_add',
540                                   [['u32', 'client_index'],
541                                    ['u32', 'context'],
542                                    ['vl_api_bier_table_id_t', 'bi_tbl_id'],
543                                    ['u16', 'bi_src'],
544                                    ['u8', 'bi_n_bytes'],
545                                    ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
546
547         table_id = {'bt_set': 0,
548                     'bt_sub_domain': 0,
549                     'bt_hdr_len_id': 0}
550
551         bibytes = b'foobar'
552
553         b = bier_imp_add.pack({'bi_tbl_id': table_id,
554                                'bi_n_bytes': len(bibytes),
555                                'bi_bytes': bibytes})
556
557         self.assertEqual(len(b), 20)
558
559
560 if __name__ == '__main__':
561     unittest.main()