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
  70         b = default_type.pack({})
  71         self.assertEqual(len(b), 2)
  72
  73         nt, size = default_type.unpack(b)
  74         self.assertEqual(len(b), size)
  75         self.assertEqual(nt.mtu, 1500)
  76
  77
  78 class TestAddType(unittest.TestCase):
  79
  80     def test_union(self):
  81         un = VPPUnionType('test_union',
  82                           [['u8', 'is_bool'],
  83                            ['u32', 'is_int']])
  84
  85         b = un.pack({'is_int': 0x12345678})
  86         nt, size = un.unpack(b)
  87         self.assertEqual(len(b), size)
  88         self.assertEqual(nt.is_bool, 0x12)
  89         self.assertEqual(nt.is_int, 0x12345678)
  90
  91     def test_address(self):
  92         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
  93                                                      ["ADDRESS_IP6", 1],
  94                                                      {"enumtype": "u32"}])
  95         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
  96                                                     'length': 4})
  97         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
  98                                                     'length': 16})
  99         VPPUnionType('vl_api_address_union_t',
 100                      [["vl_api_ip4_address_t", "ip4"],
 101                       ["vl_api_ip6_address_t", "ip6"]])
 102
 103         address = VPPType('vl_api_address_t',
 104                           [['vl_api_address_family_t', 'af'],
 105                            ['vl_api_address_union_t', 'un']])
 106
 107         prefix = VPPType('vl_api_prefix_t',
 108                          [['vl_api_address_t', 'address'],
 109                           ['u8', 'len']])
 110
 111         va_address_list = VPPType('list_addresses',
 112                                   [['u8', 'count'],
 113                                    ['vl_api_address_t', 'addresses',
 114                                     0, 'count']])
 115
 116         message_with_va_address_list = VPPType('msg_with_vla',
 117                                                [['list_addresses',
 118                                                  'vla_address'],
 119                                                 ['u8', 'is_cool']])
 120
 121         b = ip4.pack(inet_pton(AF_INET, '1.1.1.1'))
 122         self.assertEqual(len(b), 4)
 123         nt, size = ip4.unpack(b)
 124         self.assertEqual(str(nt), '1.1.1.1')
 125
 126         b = ip6.pack(inet_pton(AF_INET6, '1::1'))
 127         self.assertEqual(len(b), 16)
 128
 129         b = address.pack({'af': af.ADDRESS_IP4,
 130                           'un':
 131                           {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
 132         self.assertEqual(len(b), 20)
 133
 134         nt, size = address.unpack(b)
 135         self.assertEqual(str(nt), '2.2.2.2')
 136
 137         # List of addresses
 138         address_list = []
 139         for i in range(4):
 140             address_list.append({'af': af.ADDRESS_IP4,
 141                                  'un':
 142                                  {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
 143         b = va_address_list.pack({'count': len(address_list),
 144                                   'addresses': address_list})
 145         self.assertEqual(len(b), 81)
 146
 147         nt, size = va_address_list.unpack(b)
 148         self.assertEqual(str(nt.addresses[0]), '2.2.2.2')
 149
 150         b = message_with_va_address_list.pack({'vla_address':
 151                                                {'count': len(address_list),
 152                                                 'addresses': address_list},
 153                                                'is_cool': 100})
 154         self.assertEqual(len(b), 82)
 155         nt, size = message_with_va_address_list.unpack(b)
 156         self.assertEqual(nt.is_cool, 100)
 157
 158     def test_address_with_prefix(self):
 159         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
 160                                                      ["ADDRESS_IP6", 1],
 161                                                      {"enumtype": "u32"}])
 162         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
 163                                                     'length': 4})
 164         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
 165                                                     'length': 16})
 166         VPPUnionType('vl_api_address_union_t',
 167                      [["vl_api_ip4_address_t", "ip4"],
 168                       ["vl_api_ip6_address_t", "ip6"]])
 169
 170         address = VPPType('vl_api_address_t',
 171                           [['vl_api_address_family_t', 'af'],
 172                            ['vl_api_address_union_t', 'un']])
 173
 174
 175         prefix = VPPType('vl_api_prefix_t',
 176                          [['vl_api_address_t', 'address'],
 177                           ['u8', 'len']])
 178         prefix4 = VPPType('vl_api_ip4_prefix_t',
 179                           [['vl_api_ip4_address_t', 'address'],
 180                           ['u8', 'len']])
 181         prefix6 = VPPType('vl_api_ip6_prefix_t',
 182                           [['vl_api_ip6_address_t', 'address'],
 183                           ['u8', 'len']])
 184
 185         address_with_prefix = VPPTypeAlias('vl_api_address_with_prefix_t', {'type': 'vl_api_prefix_t' })
 186         address4_with_prefix = VPPTypeAlias('vl_api_ip4_address_with_prefix_t',
 187                                             {'type': 'vl_api_ip4_prefix_t' })
 188         address6_with_prefix = VPPTypeAlias('vl_api_ip6_address_with_prefix_t',
 189                                             {'type': 'vl_api_ip6_prefix_t' })
 190
 191         awp_type = VPPType('foobar_t',
 192                            [['vl_api_address_with_prefix_t', 'address']])
 193
 194         # address with prefix
 195         b = address_with_prefix.pack(IPv4Interface('2.2.2.2/24'))
 196         self.assertEqual(len(b), 21)
 197         nt, size = address_with_prefix.unpack(b)
 198         self.assertTrue(isinstance(nt, IPv4Interface))
 199         self.assertEqual(str(nt), '2.2.2.2/24')
 200
 201         b = address_with_prefix.pack(IPv6Interface('2::2/64'))
 202         self.assertEqual(len(b), 21)
 203         nt, size = address_with_prefix.unpack(b)
 204         self.assertTrue(isinstance(nt, IPv6Interface))
 205         self.assertEqual(str(nt), '2::2/64')
 206
 207         b = address_with_prefix.pack(IPv4Network('2.2.2.2/24', strict=False))
 208         self.assertEqual(len(b), 21)
 209         nt, size = address_with_prefix.unpack(b)
 210         self.assertTrue(isinstance(nt, IPv4Interface))
 211         self.assertEqual(str(nt), '2.2.2.0/24')
 212
 213         b = address4_with_prefix.pack('2.2.2.2/24')
 214         self.assertEqual(len(b), 5)
 215         nt, size = address4_with_prefix.unpack(b)
 216         self.assertTrue(isinstance(nt, IPv4Interface))
 217         self.assertEqual(str(nt), '2.2.2.2/24')
 218         b = address4_with_prefix.pack(IPv4Interface('2.2.2.2/24'))
 219         self.assertEqual(len(b), 5)
 220
 221         b = address6_with_prefix.pack('2::2/64')
 222         self.assertEqual(len(b), 17)
 223         nt, size = address6_with_prefix.unpack(b)
 224         self.assertTrue(isinstance(nt, IPv6Interface))
 225         self.assertEqual(str(nt), '2::2/64')
 226         b = address6_with_prefix.pack(IPv6Interface('2::2/64'))
 227         self.assertEqual(len(b), 17)
 228
 229         b = prefix.pack('192.168.10.0/24')
 230         self.assertEqual(len(b), 21)
 231         nt, size = prefix.unpack(b)
 232         self.assertTrue(isinstance(nt, IPv4Network))
 233         self.assertEqual(str(nt), '192.168.10.0/24')
 234
 235         b = awp_type.pack({'address': '1.2.3.4/24'})
 236         self.assertEqual(len(b), 21)
 237         nt, size = awp_type.unpack(b)
 238         self.assertTrue(isinstance(nt.address, IPv4Interface))
 239         self.assertEqual(str(nt.address), '1.2.3.4/24')
 240
 241         b = awp_type.pack({'address': IPv4Interface('1.2.3.4/24')})
 242         self.assertEqual(len(b), 21)
 243         nt, size = awp_type.unpack(b)
 244         self.assertTrue(isinstance(nt.address, IPv4Interface))
 245         self.assertEqual(str(nt.address), '1.2.3.4/24')
 246
 247
 248     def test_recursive_address(self):
 249         af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
 250                                                      ["ADDRESS_IP6", 1],
 251                                                      {"enumtype": "u32"}])
 252         ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
 253                                                     'length': 4})
 254         b = ip4.pack('1.1.1.1')
 255         self.assertEqual(len(b), 4)
 256         nt, size = ip4.unpack(b)
 257
 258         self.assertEqual(str(nt), '1.1.1.1')
 259
 260         ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
 261                                                     'length': 16})
 262         VPPUnionType('vl_api_address_union_t',
 263                      [["vl_api_ip4_address_t", "ip4"],
 264                       ["vl_api_ip6_address_t", "ip6"]])
 265
 266         address = VPPType('vl_api_address_t',
 267                           [['vl_api_address_family_t', 'af'],
 268                            ['vl_api_address_union_t', 'un']])
 269
 270         prefix = VPPType('vl_api_prefix_t',
 271                          [['vl_api_address_t', 'address'],
 272                           ['u8', 'len']])
 273         message = VPPMessage('svs',
 274                              [['vl_api_prefix_t', 'prefix']])
 275         message_addr = VPPMessage('svs_address',
 276                                   [['vl_api_address_t', 'address']])
 277
 278         b = message_addr.pack({'address': "1::1"})
 279         self.assertEqual(len(b), 20)
 280         nt, size = message_addr.unpack(b)
 281         self.assertEqual("1::1", str(nt.address))
 282         b = message_addr.pack({'address': "1.1.1.1"})
 283         self.assertEqual(len(b), 20)
 284         nt, size = message_addr.unpack(b)
 285         self.assertEqual("1.1.1.1", str(nt.address))
 286
 287         b = message.pack({'prefix': "1.1.1.0/24"})
 288         self.assertEqual(len(b), 21)
 289         nt, size = message.unpack(b)
 290         self.assertEqual("1.1.1.0/24", str(nt.prefix))
 291
 292         message_array = VPPMessage('address_array',
 293                                    [['vl_api_ip6_address_t',
 294                                      'addresses', 2]])
 295         b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
 296         self.assertEqual(len(b), 32)
 297         message_array_vla = VPPMessage('address_array_vla',
 298                                        [['u32', 'num'],
 299                                         ['vl_api_ip6_address_t',
 300                                          'addresses', 0, 'num']])
 301         b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
 302         self.assertEqual(len(b), 36)
 303
 304         message_array4 = VPPMessage('address_array4',
 305                                     [['vl_api_ip4_address_t',
 306                                       'addresses', 2]])
 307         b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
 308         self.assertEqual(len(b), 8)
 309         b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
 310                                                "2.2.2.2"]})
 311         self.assertEqual(len(b), 8)
 312
 313         message = VPPMessage('address', [['vl_api_address_t', 'address']])
 314         b = message.pack({'address': '1::1'})
 315         self.assertEqual(len(b), 20)
 316         b = message.pack({'address': '1.1.1.1'})
 317         self.assertEqual(len(b), 20)
 318         message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
 319         b = message.pack({'prefix': '1::1/130'})
 320         self.assertEqual(len(b), 21)
 321         b = message.pack({'prefix': IPv6Network(u'1::/119')})
 322         self.assertEqual(len(b), 21)
 323         b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
 324         self.assertEqual(len(b), 21)
 325
 326     def test_zero_vla(self):
 327         '''Default zero'ed out for VLAs'''
 328         list = VPPType('vl_api_list_t',
 329                        [['u8', 'count', 10]])
 330
 331         # Define an embedded VLA type
 332         valist = VPPType('vl_api_valist_t',
 333                          [['u8', 'count'],
 334                           ['u8', 'string', 0, 'count']])
 335         # Define a message
 336         vamessage = VPPMessage('vamsg',
 337                                [['vl_api_valist_t', 'valist'],
 338                                 ['u8', 'is_something']])
 339
 340         message = VPPMessage('msg',
 341                              [['vl_api_list_t', 'list'],
 342                               ['u8', 'is_something']])
 343
 344         # Pack message without VLA specified
 345         b = message.pack({'is_something': 1})
 346         b = vamessage.pack({'is_something': 1})
 347
 348     def test_arrays(self):
 349         # Test cases
 350         # 1. Fixed list
 351         # 2. Fixed list of variable length sub type
 352         # 3. Variable length type
 353         #
 354         s = VPPType('str', [['u32', 'length'],
 355                             ['u8', 'string', 0, 'length']])
 356
 357         ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
 358         listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
 359         valistip4 = VPPType('list_ip4_t',
 360                             [['u8', 'count'],
 361                              ['ip4_address', 'addresses', 0, 'count']])
 362
 363         valistip4_legacy = VPPType('list_ip4_t',
 364                                    [['u8', 'foo'],
 365                                     ['ip4_address', 'addresses', 0]])
 366
 367         addresses = []
 368         for i in range(4):
 369             addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
 370         b = listip4.pack({'addresses': addresses})
 371         self.assertEqual(len(b), 16)
 372         nt, size = listip4.unpack(b)
 373         self.assertEqual(nt.addresses[0].address,
 374                          inet_pton(AF_INET, '2.2.2.2'))
 375
 376         b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
 377         self.assertEqual(len(b), 17)
 378
 379         nt, size = valistip4.unpack(b)
 380         self.assertEqual(nt.count, 4)
 381         self.assertEqual(nt.addresses[0].address,
 382                          inet_pton(AF_INET, '2.2.2.2'))
 383
 384         b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
 385         self.assertEqual(len(b), 17)
 386         nt, size = valistip4_legacy.unpack(b)
 387         self.assertEqual(len(nt.addresses), 4)
 388         self.assertEqual(nt.addresses[0].address,
 389                          inet_pton(AF_INET, '2.2.2.2'))
 390
 391         string = 'foobar foobar'
 392         b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
 393         nt, size = s.unpack(b)
 394         self.assertEqual(len(b), size)
 395
 396     def test_string(self):
 397         s = VPPType('str', [['u32', 'length'],
 398                             ['u8', 'string', 0, 'length']])
 399
 400         string = ''
 401         b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
 402         nt, size = s.unpack(b)
 403         self.assertEqual(len(b), size)
 404
 405     def test_message(self):
 406         foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
 407                                  ['u8', 'client_index'],
 408                                  ['u8', 'something'],
 409                                  {"crc": "0x559b9f3c"}])
 410         b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
 411                       'something': 200})
 412         nt, size = foo.unpack(b)
 413         self.assertEqual(len(b), size)
 414         self.assertEqual(nt.something, 200)
 415
 416     def test_abf(self):
 417
 418         fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
 419                                  [['u8', 'is_uniform'],
 420                                   ['u32', 'label'],
 421                                   ['u8', 'ttl'],
 422                                   ['u8', 'exp']])
 423
 424         label_stack = {'is_uniform': 0,
 425                        'label': 0,
 426                        'ttl': 0,
 427                        'exp': 0}
 428
 429         b = fib_mpls_label.pack(label_stack)
 430         self.assertEqual(len(b), 7)
 431
 432         fib_path = VPPType('vl_api_fib_path_t',
 433                            [['u32', 'sw_if_index'],
 434                             ['u32', 'table_id'],
 435                             ['u8', 'weight'],
 436                             ['u8', 'preference'],
 437                             ['u8', 'is_local'],
 438                             ['u8', 'is_drop'],
 439                             ['u8', 'is_udp_encap'],
 440                             ['u8', 'is_unreach'],
 441                             ['u8', 'is_prohibit'],
 442                             ['u8', 'is_resolve_host'],
 443                             ['u8', 'is_resolve_attached'],
 444                             ['u8', 'is_dvr'],
 445                             ['u8', 'is_source_lookup'],
 446                             ['u8', 'afi'],
 447                             ['u8', 'next_hop', 16],
 448                             ['u32', 'next_hop_id'],
 449                             ['u32', 'rpf_id'],
 450                             ['u32', 'via_label'],
 451                             ['u8', 'n_labels'],
 452                             ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
 453         label_stack_list = []
 454         for i in range(16):
 455             label_stack_list.append(label_stack)
 456
 457         paths = {'is_udp_encap': 0,
 458                  'next_hop': b'\x10\x02\x02\xac',
 459                  'table_id': 0,
 460                  'afi': 0,
 461                  'weight': 1,
 462                  'next_hop_id': 4294967295,
 463                  'label_stack': label_stack_list,
 464                  'n_labels': 0,
 465                  'sw_if_index': 4294967295,
 466                  'preference': 0}
 467
 468         b = fib_path.pack(paths)
 469         self.assertEqual(len(b), (7*16) + 49)
 470
 471         abf_policy = VPPType('vl_api_abf_policy_t',
 472                              [['u32', 'policy_id'],
 473                               ['u32', 'acl_index'],
 474                               ['u8', 'n_paths'],
 475                               ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
 476
 477         policy = {
 478             'n_paths': 1,
 479             'paths': [paths],
 480             'acl_index': 0,
 481             'policy_id': 10}
 482
 483         b = abf_policy.pack(policy)
 484         self.assertEqual(len(b), (7*16) + 49 + 9)
 485
 486         abf_policy_add_del = VPPMessage('abf_policy_add_del',
 487                                         [['u16', '_vl_msg_id'],
 488                                          ['u32', 'client_index'],
 489                                          ['u32', 'context'],
 490                                          ['u8', 'is_add'],
 491                                          ['vl_api_abf_policy_t', 'policy']])
 492
 493         b = abf_policy_add_del.pack({'is_add': 1,
 494                                      'context': 66,
 495                                      '_vl_msg_id': 1066,
 496                                      'policy': policy})
 497
 498         nt, size = abf_policy_add_del.unpack(b)
 499         self.assertEqual(nt.policy.paths[0].next_hop,
 500                          b'\x10\x02\x02\xac\x00\x00\x00\x00'
 501                          b'\x00\x00\x00\x00\x00\x00\x00\x00')
 502
 503     def test_bier(self):
 504
 505         bier_table_id = VPPType('vl_api_bier_table_id_t',
 506                                 [['u8', 'bt_set'],
 507                                  ['u8', 'bt_sub_domain'],
 508                                  ['u8', 'bt_hdr_len_id']])
 509
 510         bier_imp_add = VPPMessage('bier_imp_add',
 511                                   [['u32', 'client_index'],
 512                                    ['u32', 'context'],
 513                                    ['vl_api_bier_table_id_t', 'bi_tbl_id'],
 514                                    ['u16', 'bi_src'],
 515                                    ['u8', 'bi_n_bytes'],
 516                                    ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
 517
 518         table_id = {'bt_set': 0,
 519                     'bt_sub_domain': 0,
 520                     'bt_hdr_len_id': 0}
 521
 522         bibytes = b'foobar'
 523
 524         b = bier_imp_add.pack({'bi_tbl_id': table_id,
 525                                'bi_n_bytes': len(bibytes),
 526                                'bi_bytes': bibytes})
 527
 528         self.assertEqual(len(b), 20)
 529
 530
 531 if __name__ == '__main__':
 532     unittest.main()