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