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
12 class TestAddType(unittest.TestCase):
15 un = VPPUnionType('test_union',
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)
25 def test_address(self):
26 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
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"]])
35 address = VPPType('vl_api_address_t',
36 [['vl_api_address_family_t', 'af'],
37 ['vl_api_address_union_t', 'un']])
39 va_address_list = VPPType('list_addresses',
41 ['vl_api_address_t', 'addresses',
44 message_with_va_address_list = VPPType('msg_with_vla',
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'))
54 b = ip6.pack({'address': inet_pton(AF_INET6, '1::1')})
55 self.assertEqual(len(b), 16)
57 b = address.pack({'af': af.ADDRESS_IP4,
60 {'address': inet_pton(AF_INET, '2.2.2.2')}}})
61 self.assertEqual(len(b), 20)
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::'))
73 address_list.append({'af': af.ADDRESS_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)
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'))
85 b = message_with_va_address_list.pack({'vla_address':
86 {'count': len(address_list),
87 'addresses': address_list},
89 self.assertEqual(len(b), 82)
90 nt, size = message_with_va_address_list.unpack(b)
91 self.assertEqual(nt.is_cool, 100)
93 def test_recursive_address(self):
94 af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0],
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"]])
103 address = VPPType('vl_api_address_t',
104 [['vl_api_address_family_t', 'af'],
105 ['vl_api_address_union_t', 'un']])
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']])
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))
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))
130 def test_zero_vla(self):
131 '''Default zero'ed out for VLAs'''
132 list = VPPType('vl_api_list_t',
133 [['u8', 'count', 10]])
135 # Define an embedded VLA type
136 valist = VPPType('vl_api_valist_t',
138 ['u8', 'string', 0, 'count']])
140 vamessage = VPPMessage('vamsg',
141 [['vl_api_valist_t', 'valist'],
142 ['u8', 'is_something']])
144 message = VPPMessage('msg',
145 [['vl_api_list_t', 'list'],
146 ['u8', 'is_something']])
148 # Pack message without VLA specified
149 b = message.pack({'is_something': 1})
150 b = vamessage.pack({'is_something': 1})
152 def test_arrays(self):
155 # 2. Fixed list of variable length sub type
156 # 3. Variable length type
158 s = VPPType('str', [['u32', 'length'],
159 ['u8', 'string', 0, 'length']])
161 ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
162 listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
163 valistip4 = VPPType('list_ip4_t',
165 ['ip4_address', 'addresses', 0, 'count']])
167 valistip4_legacy = VPPType('list_ip4_t',
169 ['ip4_address', 'addresses', 0]])
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'))
180 b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
181 self.assertEqual(len(b), 17)
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'))
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'))
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)
200 def test_string(self):
201 s = VPPType('str', [['u32', 'length'],
202 ['u8', 'string', 0, 'length']])
205 b = s.pack({'length': len(string), 'string': string.encode()})
206 nt, size = s.unpack(b)
207 self.assertEqual(len(b), size)
209 def test_message(self):
210 foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
211 ['u8', 'client_index'],
213 {"crc": "0x559b9f3c"}])
214 b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
216 nt, size = foo.unpack(b)
217 self.assertEqual(len(b), size)
218 self.assertEqual(nt.something, 200)
222 fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
223 [['u8', 'is_uniform'],
228 label_stack = {'is_uniform': 0,
233 b = fib_mpls_label.pack(label_stack)
234 self.assertEqual(len(b), 7)
236 fib_path = VPPType('vl_api_fib_path_t',
237 [['u32', 'sw_if_index'],
240 ['u8', 'preference'],
243 ['u8', 'is_udp_encap'],
244 ['u8', 'is_unreach'],
245 ['u8', 'is_prohibit'],
246 ['u8', 'is_resolve_host'],
247 ['u8', 'is_resolve_attached'],
249 ['u8', 'is_source_lookup'],
251 ['u8', 'next_hop', 16],
252 ['u32', 'next_hop_id'],
254 ['u32', 'via_label'],
256 ['vl_api_fib_mpls_label_t', 'label_stack', 16]])
257 label_stack_list = []
259 label_stack_list.append(label_stack)
261 paths = {'is_udp_encap': 0,
262 'next_hop': b'\x10\x02\x02\xac',
266 'next_hop_id': 4294967295,
267 'label_stack': label_stack_list,
269 'sw_if_index': 4294967295,
272 b = fib_path.pack(paths)
273 self.assertEqual(len(b), (7*16) + 49)
275 abf_policy = VPPType('vl_api_abf_policy_t',
276 [['u32', 'policy_id'],
277 ['u32', 'acl_index'],
279 ['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
287 b = abf_policy.pack(policy)
288 self.assertEqual(len(b), (7*16) + 49 + 9)
290 abf_policy_add_del = VPPMessage('abf_policy_add_del',
291 [['u16', '_vl_msg_id'],
292 ['u32', 'client_index'],
295 ['vl_api_abf_policy_t', 'policy']])
297 b = abf_policy_add_del.pack({'is_add': 1,
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')
309 bier_table_id = VPPType('vl_api_bier_table_id_t',
311 ['u8', 'bt_sub_domain'],
312 ['u8', 'bt_hdr_len_id']])
314 bier_imp_add = VPPMessage('bier_imp_add',
315 [['u32', 'client_index'],
317 ['vl_api_bier_table_id_t', 'bi_tbl_id'],
319 ['u8', 'bi_n_bytes'],
320 ['u8', 'bi_bytes', 0, 'bi_n_bytes']])
322 table_id = {'bt_set': 0,
328 b = bier_imp_add.pack({'bi_tbl_id': table_id,
329 'bi_n_bytes': len(bibytes),
330 'bi_bytes': bibytes})
332 self.assertEqual(len(b), 20)
335 if __name__ == '__main__':