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
10 from ipaddress import *
13 class TestLimits(unittest.TestCase):
14 def test_string(self):
15 fixed_string = VPPType('fixed_string',
16 [['string', 'name', 16]])
18 b = fixed_string.pack({'name': 'foobar'})
19 self.assertEqual(len(b), 16)
21 # Ensure string is nul terminated
22 self.assertEqual(b.decode('ascii')[6], '\x00')
24 nt, size = fixed_string.unpack(b)
25 self.assertEqual(size, 16)
26 self.assertEqual(nt.name, 'foobar')
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, '')
36 with self.assertRaises(VPPSerializerValueError):
37 b = fixed_string.pack({'name': 'foobarfoobar1234'})
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'))
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'))
51 limited_type = VPPType('limited_type_t',
52 [['string', 'name', 0, {'limit': 16}]])
53 unlimited_type = VPPType('limited_type_t',
54 [['string', 'name', 0]])
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)
61 with self.assertRaises(VPPSerializerValueError):
62 b = limited_type.pack({'name': 'foobar'*3})
65 class TestDefaults(unittest.TestCase):
66 def test_defaults(self):
67 default_type = VPPType('default_type_t',
68 [['u16', 'mtu', {'default': 1500, 'limit': 0}]])
70 b = default_type.pack({})
71 self.assertEqual(len(b), 2)
73 nt, size = default_type.unpack(b)
74 self.assertEqual(len(b), size)
75 self.assertEqual(nt.mtu, 1500)
78 class TestAddType(unittest.TestCase):
81 un = VPPUnionType('test_union',
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)
91 def test_address(self):
92 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
95 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
97 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
99 VPPUnionType('vl_api_address_union_t',
100 [["vl_api_ip4_address_t", "ip4"],
101 ["vl_api_ip6_address_t", "ip6"]])
103 address = VPPType('vl_api_address_t',
104 [['vl_api_address_family_t', 'af'],
105 ['vl_api_address_union_t', 'un']])
107 va_address_list = VPPType('list_addresses',
109 ['vl_api_address_t', 'addresses',
112 message_with_va_address_list = VPPType('msg_with_vla',
117 b = ip4.pack(inet_pton(AF_INET, '1.1.1.1'))
118 self.assertEqual(len(b), 4)
119 nt, size = ip4.unpack(b)
120 self.assertEqual(str(nt), '1.1.1.1')
122 b = ip6.pack(inet_pton(AF_INET6, '1::1'))
123 self.assertEqual(len(b), 16)
125 b = address.pack({'af': af.ADDRESS_IP4,
127 {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
128 self.assertEqual(len(b), 20)
130 nt, size = address.unpack(b)
131 self.assertEqual(str(nt), '2.2.2.2')
136 address_list.append({'af': af.ADDRESS_IP4,
138 {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
139 b = va_address_list.pack({'count': len(address_list),
140 'addresses': address_list})
141 self.assertEqual(len(b), 81)
143 nt, size = va_address_list.unpack(b)
144 self.assertEqual(str(nt.addresses[0]), '2.2.2.2')
146 b = message_with_va_address_list.pack({'vla_address':
147 {'count': len(address_list),
148 'addresses': address_list},
150 self.assertEqual(len(b), 82)
151 nt, size = message_with_va_address_list.unpack(b)
152 self.assertEqual(nt.is_cool, 100)
154 def test_recursive_address(self):
155 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
157 {"enumtype": "u32"}])
158 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
160 b = ip4.pack('1.1.1.1')
161 self.assertEqual(len(b), 4)
162 nt, size = ip4.unpack(b)
164 self.assertEqual(str(nt), '1.1.1.1')
166 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
168 VPPUnionType('vl_api_address_union_t',
169 [["vl_api_ip4_address_t", "ip4"],
170 ["vl_api_ip6_address_t", "ip6"]])
172 address = VPPType('vl_api_address_t',
173 [['vl_api_address_family_t', 'af'],
174 ['vl_api_address_union_t', 'un']])
176 prefix = VPPType('vl_api_prefix_t',
177 [['vl_api_address_t', 'address'],
179 message = VPPMessage('svs',
180 [['vl_api_prefix_t', 'prefix']])
181 message_addr = VPPMessage('svs_address',
182 [['vl_api_address_t', 'address']])
184 b = message_addr.pack({'address': "1::1"})
185 self.assertEqual(len(b), 20)
186 nt, size = message_addr.unpack(b)
187 self.assertEqual("1::1", str(nt.address))
188 b = message_addr.pack({'address': "1.1.1.1"})
189 self.assertEqual(len(b), 20)
190 nt, size = message_addr.unpack(b)
191 self.assertEqual("1.1.1.1", str(nt.address))
193 b = message.pack({'prefix': "1.1.1.0/24"})
194 self.assertEqual(len(b), 21)
195 nt, size = message.unpack(b)
196 self.assertEqual("1.1.1.0/24", str(nt.prefix))
198 message_array = VPPMessage('address_array',
199 [['vl_api_ip6_address_t',
201 b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
202 self.assertEqual(len(b), 32)
203 message_array_vla = VPPMessage('address_array_vla',
205 ['vl_api_ip6_address_t',
206 'addresses', 0, 'num']])
207 b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
208 self.assertEqual(len(b), 36)
210 message_array4 = VPPMessage('address_array4',
211 [['vl_api_ip4_address_t',
213 b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
214 self.assertEqual(len(b), 8)
215 b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
217 self.assertEqual(len(b), 8)
219 message = VPPMessage('address', [['vl_api_address_t', 'address']])
220 b = message.pack({'address': '1::1'})
221 self.assertEqual(len(b), 20)
222 b = message.pack({'address': '1.1.1.1'})
223 self.assertEqual(len(b), 20)
224 message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
225 b = message.pack({'prefix': '1::1/130'})
226 self.assertEqual(len(b), 21)
227 b = message.pack({'prefix': IPv6Network(u'1::/119')})
228 self.assertEqual(len(b), 21)
229 b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
230 self.assertEqual(len(b), 21)
232 def test_zero_vla(self):
233 '''Default zero'ed out for VLAs'''
234 list = VPPType('vl_api_list_t',
235 [['u8', 'count', 10]])
237 # Define an embedded VLA type
238 valist = VPPType('vl_api_valist_t',
240 ['u8', 'string', 0, 'count']])
242 vamessage = VPPMessage('vamsg',
243 [['vl_api_valist_t', 'valist'],
244 ['u8', 'is_something']])
246 message = VPPMessage('msg',
247 [['vl_api_list_t', 'list'],
248 ['u8', 'is_something']])
250 # Pack message without VLA specified
251 b = message.pack({'is_something': 1})
252 b = vamessage.pack({'is_something': 1})
254 def test_arrays(self):
257 # 2. Fixed list of variable length sub type
258 # 3. Variable length type
260 s = VPPType('str', [['u32', 'length'],
261 ['u8', 'string', 0, 'length']])
263 ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
264 listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
265 valistip4 = VPPType('list_ip4_t',
267 ['ip4_address', 'addresses', 0, 'count']])
269 valistip4_legacy = VPPType('list_ip4_t',
271 ['ip4_address', 'addresses', 0]])
275 addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
276 b = listip4.pack({'addresses': addresses})
277 self.assertEqual(len(b), 16)
278 nt, size = listip4.unpack(b)
279 self.assertEqual(nt.addresses[0].address,
280 inet_pton(AF_INET, '2.2.2.2'))
282 b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
283 self.assertEqual(len(b), 17)
285 nt, size = valistip4.unpack(b)
286 self.assertEqual(nt.count, 4)
287 self.assertEqual(nt.addresses[0].address,
288 inet_pton(AF_INET, '2.2.2.2'))
290 b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
291 self.assertEqual(len(b), 17)
292 nt, size = valistip4_legacy.unpack(b)
293 self.assertEqual(len(nt.addresses), 4)
294 self.assertEqual(nt.addresses[0].address,
295 inet_pton(AF_INET, '2.2.2.2'))
297 string = 'foobar foobar'
298 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
299 nt, size = s.unpack(b)
300 self.assertEqual(len(b), size)
302 def test_string(self):
303 s = VPPType('str', [['u32', 'length'],
304 ['u8', 'string', 0, 'length']])
307 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
308 nt, size = s.unpack(b)
309 self.assertEqual(len(b), size)
311 def test_message(self):
312 foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
313 ['u8', 'client_index'],
315 {"crc": "0x559b9f3c"}])
316 b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
318 nt, size = foo.unpack(b)
319 self.assertEqual(len(b), size)
320 self.assertEqual(nt.something, 200)
324 fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
325 [['u8', 'is_uniform'],
330 label_stack = {'is_uniform': 0,
335 b = fib_mpls_label.pack(label_stack)
336 self.assertEqual(len(b), 7)
338 fib_path = VPPType('vl_api_fib_path_t',
339 [['u32', 'sw_if_index'],
342 ['u8', 'preference'],
345 ['u8', 'is_udp_encap'],
346 ['u8', 'is_unreach'],
347 ['u8', 'is_prohibit'],
348 ['u8', 'is_resolve_host'],
349 ['u8', 'is_resolve_attached'],
351 ['u8', 'is_source_lookup'],
353 ['u8', 'next_hop', 16],
354 ['u32', 'next_hop_id'],
356 ['u32', 'via_label'],
358 ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
359 label_stack_list = []
361 label_stack_list.append(label_stack)
363 paths = {'is_udp_encap': 0,
364 'next_hop': b'\x10\x02\x02\xac',
368 'next_hop_id': 4294967295,
369 'label_stack': label_stack_list,
371 'sw_if_index': 4294967295,
374 b = fib_path.pack(paths)
375 self.assertEqual(len(b), (7*16) + 49)
377 abf_policy = VPPType('vl_api_abf_policy_t',
378 [['u32', 'policy_id'],
379 ['u32', 'acl_index'],
381 ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
389 b = abf_policy.pack(policy)
390 self.assertEqual(len(b), (7*16) + 49 + 9)
392 abf_policy_add_del = VPPMessage('abf_policy_add_del',
393 [['u16', '_vl_msg_id'],
394 ['u32', 'client_index'],
397 ['vl_api_abf_policy_t', 'policy']])
399 b = abf_policy_add_del.pack({'is_add': 1,
404 nt, size = abf_policy_add_del.unpack(b)
405 self.assertEqual(nt.policy.paths[0].next_hop,
406 b'\x10\x02\x02\xac\x00\x00\x00\x00'
407 b'\x00\x00\x00\x00\x00\x00\x00\x00')
411 bier_table_id = VPPType('vl_api_bier_table_id_t',
413 ['u8', 'bt_sub_domain'],
414 ['u8', 'bt_hdr_len_id']])
416 bier_imp_add = VPPMessage('bier_imp_add',
417 [['u32', 'client_index'],
419 ['vl_api_bier_table_id_t', 'bi_tbl_id'],
421 ['u8', 'bi_n_bytes'],
422 ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
424 table_id = {'bt_set': 0,
430 b = bier_imp_add.pack({'bi_tbl_id': table_id,
431 'bi_n_bytes': len(bibytes),
432 'bi_bytes': bibytes})
434 self.assertEqual(len(b), 20)
437 if __name__ == '__main__':