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