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