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