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 *
12 class TestLimits(unittest.TestCase):
14 limited_type = VPPType('limited_type_t',
15 [['string', 'name', {'limit': 16}]])
16 unlimited_type = VPPType('limited_type_t',
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)
25 with self.assertRaises(VPPSerializerValueError):
26 b = limited_type.pack({'name':'foobar'*3})
28 class TestAddType(unittest.TestCase):
31 un = VPPUnionType('test_union',
35 b = un.pack({'is_int': 0x12345678})
36 nt, size = un.unpack(b)
37 self.assertEqual(len(b), size)
38 self.assertEqual(nt.is_bool, 0x12)
39 self.assertEqual(nt.is_int, 0x12345678)
41 def test_address(self):
42 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
45 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
47 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
49 VPPUnionType('vl_api_address_union_t',
50 [["vl_api_ip4_address_t", "ip4"],
51 ["vl_api_ip6_address_t", "ip6"]])
53 address = VPPType('vl_api_address_t',
54 [['vl_api_address_family_t', 'af'],
55 ['vl_api_address_union_t', 'un']])
57 va_address_list = VPPType('list_addresses',
59 ['vl_api_address_t', 'addresses',
62 message_with_va_address_list = VPPType('msg_with_vla',
67 b = ip4.pack(inet_pton(AF_INET, '1.1.1.1'))
68 self.assertEqual(len(b), 4)
69 nt, size = ip4.unpack(b)
70 self.assertEqual(str(nt), '1.1.1.1')
72 b = ip6.pack(inet_pton(AF_INET6, '1::1'))
73 self.assertEqual(len(b), 16)
75 b = address.pack({'af': af.ADDRESS_IP4,
77 {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
78 self.assertEqual(len(b), 20)
80 nt, size = address.unpack(b)
81 self.assertEqual(str(nt), '2.2.2.2')
86 address_list.append({'af': af.ADDRESS_IP4,
88 {'ip4': inet_pton(AF_INET, '2.2.2.2')}})
89 b = va_address_list.pack({'count': len(address_list),
90 'addresses': address_list})
91 self.assertEqual(len(b), 81)
93 nt, size = va_address_list.unpack(b)
94 self.assertEqual(str(nt.addresses[0]), '2.2.2.2')
96 b = message_with_va_address_list.pack({'vla_address':
97 {'count': len(address_list),
98 'addresses': address_list},
100 self.assertEqual(len(b), 82)
101 nt, size = message_with_va_address_list.unpack(b)
102 self.assertEqual(nt.is_cool, 100)
104 def test_recursive_address(self):
105 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
107 {"enumtype": "u32"}])
108 ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8',
110 b = ip4.pack('1.1.1.1')
111 self.assertEqual(len(b), 4)
112 nt, size = ip4.unpack(b)
114 self.assertEqual(str(nt), '1.1.1.1')
116 ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8',
118 VPPUnionType('vl_api_address_union_t',
119 [["vl_api_ip4_address_t", "ip4"],
120 ["vl_api_ip6_address_t", "ip6"]])
122 address = VPPType('vl_api_address_t',
123 [['vl_api_address_family_t', 'af'],
124 ['vl_api_address_union_t', 'un']])
126 prefix = VPPType('vl_api_prefix_t',
127 [['vl_api_address_t', 'address'],
128 ['u8', 'address_length']])
129 message = VPPMessage('svs',
130 [['vl_api_prefix_t', 'prefix']])
131 message_addr = VPPMessage('svs_address',
132 [['vl_api_address_t', 'address']])
134 b = message_addr.pack({'address': "1::1"})
135 self.assertEqual(len(b), 20)
136 nt, size = message_addr.unpack(b)
137 self.assertEqual("1::1", str(nt.address))
138 b = message_addr.pack({'address': "1.1.1.1"})
139 self.assertEqual(len(b), 20)
140 nt, size = message_addr.unpack(b)
141 self.assertEqual("1.1.1.1", str(nt.address))
143 b = message.pack({'prefix': "1.1.1.0/24"})
144 self.assertEqual(len(b), 21)
145 nt, size = message.unpack(b)
146 self.assertEqual("1.1.1.0/24", str(nt.prefix))
148 message_array = VPPMessage('address_array',
149 [['vl_api_ip6_address_t',
151 b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]})
152 self.assertEqual(len(b), 32)
153 message_array_vla = VPPMessage('address_array_vla',
155 ['vl_api_ip6_address_t',
156 'addresses', 0, 'num']])
157 b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2})
158 self.assertEqual(len(b), 36)
160 message_array4 = VPPMessage('address_array4',
161 [['vl_api_ip4_address_t',
163 b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]})
164 self.assertEqual(len(b), 8)
165 b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"),
167 self.assertEqual(len(b), 8)
169 message = VPPMessage('address', [['vl_api_address_t', 'address']])
170 b = message.pack({'address': '1::1'})
171 self.assertEqual(len(b), 20)
172 b = message.pack({'address': '1.1.1.1'})
173 self.assertEqual(len(b), 20)
174 message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']])
175 b = message.pack({'prefix': '1::1/130'})
176 self.assertEqual(len(b), 21)
177 b = message.pack({'prefix': IPv6Network(u'1::/119')})
178 self.assertEqual(len(b), 21)
179 b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')})
180 self.assertEqual(len(b), 21)
182 def test_zero_vla(self):
183 '''Default zero'ed out for VLAs'''
184 list = VPPType('vl_api_list_t',
185 [['u8', 'count', 10]])
187 # Define an embedded VLA type
188 valist = VPPType('vl_api_valist_t',
190 ['u8', 'string', 0, 'count']])
192 vamessage = VPPMessage('vamsg',
193 [['vl_api_valist_t', 'valist'],
194 ['u8', 'is_something']])
196 message = VPPMessage('msg',
197 [['vl_api_list_t', 'list'],
198 ['u8', 'is_something']])
200 # Pack message without VLA specified
201 b = message.pack({'is_something': 1})
202 b = vamessage.pack({'is_something': 1})
204 def test_arrays(self):
207 # 2. Fixed list of variable length sub type
208 # 3. Variable length type
210 s = VPPType('str', [['u32', 'length'],
211 ['u8', 'string', 0, 'length']])
213 ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
214 listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
215 valistip4 = VPPType('list_ip4_t',
217 ['ip4_address', 'addresses', 0, 'count']])
219 valistip4_legacy = VPPType('list_ip4_t',
221 ['ip4_address', 'addresses', 0]])
225 addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
226 b = listip4.pack({'addresses': addresses})
227 self.assertEqual(len(b), 16)
228 nt, size = listip4.unpack(b)
229 self.assertEqual(nt.addresses[0].address,
230 inet_pton(AF_INET, '2.2.2.2'))
232 b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
233 self.assertEqual(len(b), 17)
235 nt, size = valistip4.unpack(b)
236 self.assertEqual(nt.count, 4)
237 self.assertEqual(nt.addresses[0].address,
238 inet_pton(AF_INET, '2.2.2.2'))
240 b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
241 self.assertEqual(len(b), 17)
242 nt, size = valistip4_legacy.unpack(b)
243 self.assertEqual(len(nt.addresses), 4)
244 self.assertEqual(nt.addresses[0].address,
245 inet_pton(AF_INET, '2.2.2.2'))
247 string = 'foobar foobar'
248 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
249 nt, size = s.unpack(b)
250 self.assertEqual(len(b), size)
252 def test_string(self):
253 s = VPPType('str', [['u32', 'length'],
254 ['u8', 'string', 0, 'length']])
257 b = s.pack({'length': len(string), 'string': string.encode('utf-8')})
258 nt, size = s.unpack(b)
259 self.assertEqual(len(b), size)
261 def test_message(self):
262 foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
263 ['u8', 'client_index'],
265 {"crc": "0x559b9f3c"}])
266 b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
268 nt, size = foo.unpack(b)
269 self.assertEqual(len(b), size)
270 self.assertEqual(nt.something, 200)
274 fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
275 [['u8', 'is_uniform'],
280 label_stack = {'is_uniform': 0,
285 b = fib_mpls_label.pack(label_stack)
286 self.assertEqual(len(b), 7)
288 fib_path = VPPType('vl_api_fib_path_t',
289 [['u32', 'sw_if_index'],
292 ['u8', 'preference'],
295 ['u8', 'is_udp_encap'],
296 ['u8', 'is_unreach'],
297 ['u8', 'is_prohibit'],
298 ['u8', 'is_resolve_host'],
299 ['u8', 'is_resolve_attached'],
301 ['u8', 'is_source_lookup'],
303 ['u8', 'next_hop', 16],
304 ['u32', 'next_hop_id'],
306 ['u32', 'via_label'],
308 ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
309 label_stack_list = []
311 label_stack_list.append(label_stack)
313 paths = {'is_udp_encap': 0,
314 'next_hop': b'\x10\x02\x02\xac',
318 'next_hop_id': 4294967295,
319 'label_stack': label_stack_list,
321 'sw_if_index': 4294967295,
324 b = fib_path.pack(paths)
325 self.assertEqual(len(b), (7*16) + 49)
327 abf_policy = VPPType('vl_api_abf_policy_t',
328 [['u32', 'policy_id'],
329 ['u32', 'acl_index'],
331 ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
339 b = abf_policy.pack(policy)
340 self.assertEqual(len(b), (7*16) + 49 + 9)
342 abf_policy_add_del = VPPMessage('abf_policy_add_del',
343 [['u16', '_vl_msg_id'],
344 ['u32', 'client_index'],
347 ['vl_api_abf_policy_t', 'policy']])
349 b = abf_policy_add_del.pack({'is_add': 1,
354 nt, size = abf_policy_add_del.unpack(b)
355 self.assertEqual(nt.policy.paths[0].next_hop,
356 b'\x10\x02\x02\xac\x00\x00\x00\x00'
357 b'\x00\x00\x00\x00\x00\x00\x00\x00')
361 bier_table_id = VPPType('vl_api_bier_table_id_t',
363 ['u8', 'bt_sub_domain'],
364 ['u8', 'bt_hdr_len_id']])
366 bier_imp_add = VPPMessage('bier_imp_add',
367 [['u32', 'client_index'],
369 ['vl_api_bier_table_id_t', 'bi_tbl_id'],
371 ['u8', 'bi_n_bytes'],
372 ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
374 table_id = {'bt_set': 0,
380 b = bier_imp_add.pack({'bi_tbl_id': table_id,
381 'bi_n_bytes': len(bibytes),
382 'bi_bytes': bibytes})
384 self.assertEqual(len(b), 20)
387 if __name__ == '__main__':