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