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
   7 from socket import inet_pton, AF_INET, AF_INET6
   8 import logging
   9 import sys
  10 from ipaddress import *
  11
  12
  13 class TestAddType(unittest.TestCase):
  14
  15     def test_union(self):
  16         un = VPPUnionType('test_union',
  17                           [['u8', 'is_bool'],
  18                            ['u32', 'is_int']])
  19
  20         b = un.pack({'is_int': 0x12345678})
  21         nt, size = un.unpack(b)
  22         self.assertEqual(len(b), size)
  23         self.assertEqual(nt.is_bool, 0x12)
  24         self.assertEqual(nt.is_int, 0x12345678)
  25
  26     def test_address(self):
  27         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
  28                                                      ["ADDRESS_IP6", 1],
  29                                                      {"enumtype": "u32"}])
  30         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
  31                                                     'length': 4})
  32         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
  33                                                     'length': 16})
  34         VPPUnionType('vl_api_address_union_t',
  35                      [["vl_api_ip4_address_t", "ip4"],
  36                       ["vl_api_ip6_address_t", "ip6"]])
  37
  38         address = VPPType('vl_api_address_t',
  39                           [['vl_api_address_family_t', 'af'],
  40                            ['vl_api_address_union_t', 'un']])
  41
  42         va_address_list = VPPType('list_addresses',
  43                                   [['u8', 'count'],
  44                                    ['vl_api_address_t', 'addresses',
  45                                     0, 'count']])
  46
  47         message_with_va_address_list = VPPType('msg_with_vla',
  48                                                [['list_addresses',
  49                                                  'vla_address'],
  50                                                 ['u8', 'is_cool']])
  51
  52         b = ip4.pack(inet_pton(AF_INET, '1.1.1.1'))
  53         self.assertEqual(len(b), 4)
  54         nt, size = ip4.unpack(b)
  55         self.assertEqual(str(nt), '1.1.1.1')
  56
  57         b = ip6.pack(inet_pton(AF_INET6, '1::1'))
  58         self.assertEqual(len(b), 16)
  59
  60         b = address.pack({'af': af.ADDRESS_IP4,
  61                           'un':
  62                           {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
  63         self.assertEqual(len(b), 20)
  64
  65         nt, size = address.unpack(b)
  66         self.assertEqual(str(nt), '2.2.2.2')
  67
  68         # List of addresses
  69         address_list = []
  70         for i in range(4):
  71             address_list.append({'af': af.ADDRESS_IP4,
  72                                  'un':
  73                                  {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
  74         b = va_address_list.pack({'count': len(address_list),
  75                                   'addresses': address_list})
  76         self.assertEqual(len(b), 81)
  77
  78         nt, size = va_address_list.unpack(b)
  79         self.assertEqual(str(nt.addresses[0]), '2.2.2.2')
  80
  81         b = message_with_va_address_list.pack({'vla_address':
  82                                                {'count': len(address_list),
  83                                                 'addresses': address_list},
  84                                                'is_cool': 100})
  85         self.assertEqual(len(b), 82)
  86         nt, size = message_with_va_address_list.unpack(b)
  87         self.assertEqual(nt.is_cool, 100)
  88
  89     def test_recursive_address(self):
  90         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
  91                                                      ["ADDRESS_IP6", 1],
  92                                                      {"enumtype": "u32"}])
  93         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
  94                                                     'length': 4})
  95         b = ip4.pack('1.1.1.1')
  96         self.assertEqual(len(b), 4)
  97         nt, size = ip4.unpack(b)
  98
  99         self.assertEqual(str(nt), '1.1.1.1')
 100
 101         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
 102                                                     'length': 16})
 103         VPPUnionType('vl_api_address_union_t',
 104                      [["vl_api_ip4_address_t", "ip4"],
 105                       ["vl_api_ip6_address_t", "ip6"]])
 106
 107         address = VPPType('vl_api_address_t',
 108                           [['vl_api_address_family_t', 'af'],
 109                            ['vl_api_address_union_t', 'un']])
 110
 111         prefix = VPPType('vl_api_prefix_t',
 112                          [['vl_api_address_t', 'address'],
 113                           ['u8', 'address_length']])
 114         message = VPPMessage('svs',
 115                              [['vl_api_prefix_t', 'prefix']])
 116         message_addr = VPPMessage('svs_address',
 117                                   [['vl_api_address_t', 'address']])
 118
 119         b = message_addr.pack({'address': "1::1"})
 120         self.assertEqual(len(b), 20)
 121         nt, size = message_addr.unpack(b)
 122         self.assertEqual("1::1", str(nt.address))
 123         b = message_addr.pack({'address': "1.1.1.1"})
 124         self.assertEqual(len(b), 20)
 125         nt, size = message_addr.unpack(b)
 126         self.assertEqual("1.1.1.1", str(nt.address))
 127
 128         b = message.pack({'prefix': "1.1.1.0/24"})
 129         self.assertEqual(len(b), 21)
 130         nt, size = message.unpack(b)
 131         self.assertEqual("1.1.1.0/24", str(nt.prefix))
 132
 133         message_array = VPPMessage('address_array',
 134                                    [['vl_api_ip6_address_t',
 135                                      'addresses', 2]])
 136         b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
 137         self.assertEqual(len(b), 32)
 138         message_array_vla = VPPMessage('address_array_vla',
 139                                        [['u32', 'num'],
 140                                         ['vl_api_ip6_address_t',
 141                                          'addresses', 0, 'num']])
 142         b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
 143         self.assertEqual(len(b), 36)
 144
 145         message_array4 = VPPMessage('address_array4',
 146                                     [['vl_api_ip4_address_t',
 147                                       'addresses', 2]])
 148         b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
 149         self.assertEqual(len(b), 8)
 150         b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
 151                                                "2.2.2.2"]})
 152         self.assertEqual(len(b), 8)
 153
 154         message = VPPMessage('address', [['vl_api_address_t', 'address']])
 155         b = message.pack({'address': '1::1'})
 156         self.assertEqual(len(b), 20)
 157         b = message.pack({'address': '1.1.1.1'})
 158         self.assertEqual(len(b), 20)
 159         message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
 160         b = message.pack({'prefix': '1::1/130'})
 161         self.assertEqual(len(b), 21)
 162         b = message.pack({'prefix': IPv6Network(u'1::/119')})
 163         self.assertEqual(len(b), 21)
 164         b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
 165         self.assertEqual(len(b), 21)
 166
 167     def test_zero_vla(self):
 168         '''Default zero'ed out for VLAs'''
 169         list = VPPType('vl_api_list_t',
 170                        [['u8', 'count', 10]])
 171
 172         # Define an embedded VLA type
 173         valist = VPPType('vl_api_valist_t',
 174                          [['u8', 'count'],
 175                           ['u8', 'string', 0, 'count']])
 176         # Define a message
 177         vamessage = VPPMessage('vamsg',
 178                                [['vl_api_valist_t', 'valist'],
 179                                 ['u8', 'is_something']])
 180
 181         message = VPPMessage('msg',
 182                              [['vl_api_list_t', 'list'],
 183                               ['u8', 'is_something']])
 184
 185         # Pack message without VLA specified
 186         b = message.pack({'is_something': 1})
 187         b = vamessage.pack({'is_something': 1})
 188
 189     def test_arrays(self):
 190         # Test cases
 191         # 1. Fixed list
 192         # 2. Fixed list of variable length sub type
 193         # 3. Variable length type
 194         #
 195         s = VPPType('str', [['u32', 'length'],
 196                             ['u8', 'string', 0, 'length']])
 197
 198         ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
 199         listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
 200         valistip4 = VPPType('list_ip4_t',
 201                             [['u8', 'count'],
 202                              ['ip4_address', 'addresses', 0, 'count']])
 203
 204         valistip4_legacy = VPPType('list_ip4_t',
 205                                    [['u8', 'foo'],
 206                                     ['ip4_address', 'addresses', 0]])
 207
 208         addresses = []
 209         for i in range(4):
 210             addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
 211         b = listip4.pack({'addresses': addresses})
 212         self.assertEqual(len(b), 16)
 213         nt, size = listip4.unpack(b)
 214         self.assertEqual(nt.addresses[0].address,
 215                          inet_pton(AF_INET, '2.2.2.2'))
 216
 217         b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
 218         self.assertEqual(len(b), 17)
 219
 220         nt, size = valistip4.unpack(b)
 221         self.assertEqual(nt.count, 4)
 222         self.assertEqual(nt.addresses[0].address,
 223                          inet_pton(AF_INET, '2.2.2.2'))
 224
 225         b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
 226         self.assertEqual(len(b), 17)
 227         nt, size = valistip4_legacy.unpack(b)
 228         self.assertEqual(len(nt.addresses), 4)
 229         self.assertEqual(nt.addresses[0].address,
 230                          inet_pton(AF_INET, '2.2.2.2'))
 231
 232         string = 'foobar foobar'
 233         b = s.pack({'length': len(string), 'string': string.encode()})
 234         nt, size = s.unpack(b)
 235         self.assertEqual(len(b), size)
 236
 237     def test_string(self):
 238         s = VPPType('str', [['u32', 'length'],
 239                             ['u8', 'string', 0, 'length']])
 240
 241         string = ''
 242         b = s.pack({'length': len(string), 'string': string.encode()})
 243         nt, size = s.unpack(b)
 244         self.assertEqual(len(b), size)
 245
 246     def test_message(self):
 247         foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
 248                                  ['u8', 'client_index'],
 249                                  ['u8', 'something'],
 250                                  {"crc": "0x559b9f3c"}])
 251         b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
 252                       'something': 200})
 253         nt, size = foo.unpack(b)
 254         self.assertEqual(len(b), size)
 255         self.assertEqual(nt.something, 200)
 256
 257     def test_abf(self):
 258
 259         fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
 260                                  [['u8', 'is_uniform'],
 261                                   ['u32', 'label'],
 262                                   ['u8', 'ttl'],
 263                                   ['u8', 'exp']])
 264
 265         label_stack = {'is_uniform': 0,
 266                        'label': 0,
 267                        'ttl': 0,
 268                        'exp': 0}
 269
 270         b = fib_mpls_label.pack(label_stack)
 271         self.assertEqual(len(b), 7)
 272
 273         fib_path = VPPType('vl_api_fib_path_t',
 274                            [['u32', 'sw_if_index'],
 275                             ['u32', 'table_id'],
 276                             ['u8', 'weight'],
 277                             ['u8', 'preference'],
 278                             ['u8', 'is_local'],
 279                             ['u8', 'is_drop'],
 280                             ['u8', 'is_udp_encap'],
 281                             ['u8', 'is_unreach'],
 282                             ['u8', 'is_prohibit'],
 283                             ['u8', 'is_resolve_host'],
 284                             ['u8', 'is_resolve_attached'],
 285                             ['u8', 'is_dvr'],
 286                             ['u8', 'is_source_lookup'],
 287                             ['u8', 'afi'],
 288                             ['u8', 'next_hop', 16],
 289                             ['u32', 'next_hop_id'],
 290                             ['u32', 'rpf_id'],
 291                             ['u32', 'via_label'],
 292                             ['u8', 'n_labels'],
 293                             ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
 294         label_stack_list = []
 295         for i in range(16):
 296             label_stack_list.append(label_stack)
 297
 298         paths = {'is_udp_encap': 0,
 299                  'next_hop': b'\x10\x02\x02\xac',
 300                  'table_id': 0,
 301                  'afi': 0,
 302                  'weight': 1,
 303                  'next_hop_id': 4294967295,
 304                  'label_stack': label_stack_list,
 305                  'n_labels': 0,
 306                  'sw_if_index': 4294967295,
 307                  'preference': 0}
 308
 309         b = fib_path.pack(paths)
 310         self.assertEqual(len(b), (7*16) + 49)
 311
 312         abf_policy = VPPType('vl_api_abf_policy_t',
 313                              [['u32', 'policy_id'],
 314                               ['u32', 'acl_index'],
 315                               ['u8', 'n_paths'],
 316                               ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
 317
 318         policy = {
 319             'n_paths': 1,
 320             'paths': [paths],
 321             'acl_index': 0,
 322             'policy_id': 10}
 323
 324         b = abf_policy.pack(policy)
 325         self.assertEqual(len(b), (7*16) + 49 + 9)
 326
 327         abf_policy_add_del = VPPMessage('abf_policy_add_del',
 328                                         [['u16', '_vl_msg_id'],
 329                                          ['u32', 'client_index'],
 330                                          ['u32', 'context'],
 331                                          ['u8', 'is_add'],
 332                                          ['vl_api_abf_policy_t', 'policy']])
 333
 334         b = abf_policy_add_del.pack({'is_add': 1,
 335                                      'context': 66,
 336                                      '_vl_msg_id': 1066,
 337                                      'policy': policy})
 338
 339         nt, size = abf_policy_add_del.unpack(b)
 340         self.assertEqual(nt.policy.paths[0].next_hop,
 341                          b'\x10\x02\x02\xac\x00\x00\x00\x00'
 342                          b'\x00\x00\x00\x00\x00\x00\x00\x00')
 343
 344     def test_bier(self):
 345
 346         bier_table_id = VPPType('vl_api_bier_table_id_t',
 347                                 [['u8', 'bt_set'],
 348                                  ['u8', 'bt_sub_domain'],
 349                                  ['u8', 'bt_hdr_len_id']])
 350
 351         bier_imp_add = VPPMessage('bier_imp_add',
 352                                   [['u32', 'client_index'],
 353                                    ['u32', 'context'],
 354                                    ['vl_api_bier_table_id_t', 'bi_tbl_id'],
 355                                    ['u16', 'bi_src'],
 356                                    ['u8', 'bi_n_bytes'],
 357                                    ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
 358
 359         table_id = {'bt_set': 0,
 360                     'bt_sub_domain': 0,
 361                     'bt_hdr_len_id': 0}
 362
 363         bibytes = b'foobar'
 364
 365         b = bier_imp_add.pack({'bi_tbl_id': table_id,
 366                                'bi_n_bytes': len(bibytes),
 367                                'bi_bytes': bibytes})
 368
 369         self.assertEqual(len(b), 20)
 370
 371
 372 if __name__ == '__main__':
 373     unittest.main()