4 from vpp_papi.vpp_serializer import VPPType, VPPEnumType, VPPEnumFlagType
5 from vpp_papi.vpp_serializer import VPPUnionType, VPPMessage
6 from vpp_papi.vpp_serializer import VPPTypeAlias, VPPSerializerValueError
7 from vpp_papi import MACAddress
8 from socket import inet_pton, AF_INET, AF_INET6
11 from ipaddress import *
14 class TestLimits(unittest.TestCase):
15 def test_string(self):
16 fixed_string = VPPType("fixed_string", [["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", [["string", "name", 0]])
40 b = variable_string.pack({"name": "foobar"})
41 self.assertEqual(len(b), 4 + len("foobar"))
43 nt, size = variable_string.unpack(b)
44 self.assertEqual(size, 4 + len("foobar"))
45 self.assertEqual(nt.name, "foobar")
46 self.assertEqual(len(nt.name), len("foobar"))
49 limited_type = VPPType("limited_type_t", [["string", "name", 0, {"limit": 16}]])
50 unlimited_type = VPPType("limited_type_t", [["string", "name", 0]])
52 b = limited_type.pack({"name": "foobar"})
53 self.assertEqual(len(b), 10)
54 b = unlimited_type.pack({"name": "foobar"})
55 self.assertEqual(len(b), 10)
57 with self.assertRaises(VPPSerializerValueError):
58 b = limited_type.pack({"name": "foobar" * 3})
61 class TestDefaults(unittest.TestCase):
62 def test_defaults(self):
63 default_type = VPPType(
64 "default_type_t", [["u16", "mtu", {"default": 1500, "limit": 0}]]
66 without_default_type = VPPType("without_default_type_t", [["u16", "mtu"]])
68 b = default_type.pack({})
69 self.assertEqual(len(b), 2)
70 nt, size = default_type.unpack(b)
71 self.assertEqual(len(b), size)
72 self.assertEqual(nt.mtu, 1500)
74 # distinguish between parameter 0 and parameter not passed
75 b = default_type.pack({"mtu": 0})
76 self.assertEqual(len(b), 2)
77 nt, size = default_type.unpack(b)
78 self.assertEqual(len(b), size)
79 self.assertEqual(nt.mtu, 0)
81 # Ensure that basetypes does not inherit default
82 b = without_default_type.pack({})
83 self.assertEqual(len(b), 2)
84 nt, size = default_type.unpack(b)
85 self.assertEqual(len(b), size)
86 self.assertEqual(nt.mtu, 0)
91 [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
94 default_with_enum = VPPType(
95 "default_enum_type_t",
96 [["u16", "mtu"], ["vl_api_enum_t", "e", {"default": 1}]],
99 b = default_with_enum.pack({})
100 self.assertEqual(len(b), 6)
101 nt, size = default_with_enum.unpack(b)
102 self.assertEqual(len(b), size)
103 self.assertEqual(nt.e, 1)
106 class TestAddType(unittest.TestCase):
107 def test_union(self):
108 un = VPPUnionType("test_union", [["u8", "is_bool"], ["u32", "is_int"]])
110 b = un.pack({"is_int": 0x12345678})
111 nt, size = un.unpack(b)
112 self.assertEqual(len(b), size)
113 self.assertEqual(nt.is_bool, 0x12)
114 self.assertEqual(nt.is_int, 0x12345678)
116 def test_address(self):
118 "vl_api_address_family_t",
119 [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
121 aff = VPPEnumFlagType(
122 "vl_api_address_family_flag_t",
123 [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
125 ip4 = VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
126 ip6 = VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
128 "vl_api_address_union_t",
129 [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
134 [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
138 "vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]]
141 va_address_list = VPPType(
143 [["u8", "count"], ["vl_api_address_t", "addresses", 0, "count"]],
146 message_with_va_address_list = VPPType(
147 "msg_with_vla", [["list_addresses", "vla_address"], ["u8", "is_cool"]]
150 b = ip4.pack(inet_pton(AF_INET, "1.1.1.1"))
151 self.assertEqual(len(b), 4)
152 nt, size = ip4.unpack(b)
153 self.assertEqual(str(nt), "1.1.1.1")
155 b = ip6.pack(inet_pton(AF_INET6, "1::1"))
156 self.assertEqual(len(b), 16)
159 {"af": af.ADDRESS_IP4, "un": {"ip4": inet_pton(AF_INET, "2.2.2.2")}}
161 self.assertEqual(len(b), 20)
163 nt, size = address.unpack(b)
164 self.assertEqual(str(nt), "2.2.2.2")
170 {"af": af.ADDRESS_IP4, "un": {"ip4": inet_pton(AF_INET, "2.2.2.2")}}
172 b = va_address_list.pack(
173 {"count": len(address_list), "addresses": address_list}
175 self.assertEqual(len(b), 81)
177 nt, size = va_address_list.unpack(b)
178 self.assertEqual(str(nt.addresses[0]), "2.2.2.2")
180 b = message_with_va_address_list.pack(
182 "vla_address": {"count": len(address_list), "addresses": address_list},
186 self.assertEqual(len(b), 82)
187 nt, size = message_with_va_address_list.unpack(b)
188 self.assertEqual(nt.is_cool, 100)
190 def test_address_with_prefix(self):
192 "vl_api_address_family_t",
193 [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
195 ip4 = VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
196 ip6 = VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
198 "vl_api_address_union_t",
199 [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
204 [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
208 "vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]]
211 "vl_api_ip4_prefix_t", [["vl_api_ip4_address_t", "address"], ["u8", "len"]]
214 "vl_api_ip6_prefix_t", [["vl_api_ip6_address_t", "address"], ["u8", "len"]]
217 address_with_prefix = VPPTypeAlias(
218 "vl_api_address_with_prefix_t", {"type": "vl_api_prefix_t"}
220 address4_with_prefix = VPPTypeAlias(
221 "vl_api_ip4_address_with_prefix_t", {"type": "vl_api_ip4_prefix_t"}
223 address6_with_prefix = VPPTypeAlias(
224 "vl_api_ip6_address_with_prefix_t", {"type": "vl_api_ip6_prefix_t"}
227 awp_type = VPPType("foobar_t", [["vl_api_address_with_prefix_t", "address"]])
229 # address with prefix
230 b = address_with_prefix.pack(IPv4Interface("2.2.2.2/24"))
231 self.assertEqual(len(b), 21)
232 nt, size = address_with_prefix.unpack(b)
233 self.assertTrue(isinstance(nt, IPv4Interface))
234 self.assertEqual(str(nt), "2.2.2.2/24")
236 b = address_with_prefix.pack(IPv6Interface("2::2/64"))
237 self.assertEqual(len(b), 21)
238 nt, size = address_with_prefix.unpack(b)
239 self.assertTrue(isinstance(nt, IPv6Interface))
240 self.assertEqual(str(nt), "2::2/64")
242 b = address_with_prefix.pack(IPv4Network("2.2.2.2/24", strict=False))
243 self.assertEqual(len(b), 21)
244 nt, size = address_with_prefix.unpack(b)
245 self.assertTrue(isinstance(nt, IPv4Interface))
246 self.assertEqual(str(nt), "2.2.2.0/24")
248 b = address4_with_prefix.pack("2.2.2.2/24")
249 self.assertEqual(len(b), 5)
250 nt, size = address4_with_prefix.unpack(b)
251 self.assertTrue(isinstance(nt, IPv4Interface))
252 self.assertEqual(str(nt), "2.2.2.2/24")
253 b = address4_with_prefix.pack(IPv4Interface("2.2.2.2/24"))
254 self.assertEqual(len(b), 5)
256 b = address6_with_prefix.pack("2::2/64")
257 self.assertEqual(len(b), 17)
258 nt, size = address6_with_prefix.unpack(b)
259 self.assertTrue(isinstance(nt, IPv6Interface))
260 self.assertEqual(str(nt), "2::2/64")
261 b = address6_with_prefix.pack(IPv6Interface("2::2/64"))
262 self.assertEqual(len(b), 17)
264 b = prefix.pack("192.168.10.0/24")
265 self.assertEqual(len(b), 21)
266 nt, size = prefix.unpack(b)
267 self.assertTrue(isinstance(nt, IPv4Network))
268 self.assertEqual(str(nt), "192.168.10.0/24")
270 b = awp_type.pack({"address": "1.2.3.4/24"})
271 self.assertEqual(len(b), 21)
272 nt, size = awp_type.unpack(b)
273 self.assertTrue(isinstance(nt.address, IPv4Interface))
274 self.assertEqual(str(nt.address), "1.2.3.4/24")
276 b = awp_type.pack({"address": IPv4Interface("1.2.3.4/24")})
277 self.assertEqual(len(b), 21)
278 nt, size = awp_type.unpack(b)
279 self.assertTrue(isinstance(nt.address, IPv4Interface))
280 self.assertEqual(str(nt.address), "1.2.3.4/24")
282 def test_recursive_address(self):
284 "vl_api_address_family_t",
285 [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
287 ip4 = VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
288 b = ip4.pack("1.1.1.1")
289 self.assertEqual(len(b), 4)
290 nt, size = ip4.unpack(b)
292 self.assertEqual(str(nt), "1.1.1.1")
294 ip6 = VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
296 "vl_api_address_union_t",
297 [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
302 [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
306 "vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]]
308 message = VPPMessage("svs", [["vl_api_prefix_t", "prefix"]])
309 message_addr = VPPMessage("svs_address", [["vl_api_address_t", "address"]])
311 b = message_addr.pack({"address": "1::1"})
312 self.assertEqual(len(b), 20)
313 nt, size = message_addr.unpack(b)
314 self.assertEqual("1::1", str(nt.address))
315 b = message_addr.pack({"address": "1.1.1.1"})
316 self.assertEqual(len(b), 20)
317 nt, size = message_addr.unpack(b)
318 self.assertEqual("1.1.1.1", str(nt.address))
320 b = message.pack({"prefix": "1.1.1.0/24"})
321 self.assertEqual(len(b), 21)
322 nt, size = message.unpack(b)
323 self.assertEqual("1.1.1.0/24", str(nt.prefix))
325 message_array = VPPMessage(
326 "address_array", [["vl_api_ip6_address_t", "addresses", 2]]
328 b = message_array.pack({"addresses": [IPv6Address("1::1"), "2::2"]})
329 self.assertEqual(len(b), 32)
330 message_array_vla = VPPMessage(
332 [["u32", "num"], ["vl_api_ip6_address_t", "addresses", 0, "num"]],
334 b = message_array_vla.pack({"addresses": ["1::1", "2::2"], "num": 2})
335 self.assertEqual(len(b), 36)
337 message_array4 = VPPMessage(
338 "address_array4", [["vl_api_ip4_address_t", "addresses", 2]]
340 b = message_array4.pack({"addresses": ["1.1.1.1", "2.2.2.2"]})
341 self.assertEqual(len(b), 8)
342 b = message_array4.pack({"addresses": [IPv4Address("1.1.1.1"), "2.2.2.2"]})
343 self.assertEqual(len(b), 8)
345 message = VPPMessage("address", [["vl_api_address_t", "address"]])
346 b = message.pack({"address": "1::1"})
347 self.assertEqual(len(b), 20)
348 b = message.pack({"address": "1.1.1.1"})
349 self.assertEqual(len(b), 20)
350 message = VPPMessage("prefix", [["vl_api_prefix_t", "prefix"]])
351 b = message.pack({"prefix": "1::1/130"})
352 self.assertEqual(len(b), 21)
353 b = message.pack({"prefix": IPv6Network("1::/119")})
354 self.assertEqual(len(b), 21)
355 b = message.pack({"prefix": IPv4Network("1.1.0.0/16")})
356 self.assertEqual(len(b), 21)
358 def test_zero_vla(self):
359 """Default zero'ed out for VLAs"""
360 list = VPPType("vl_api_list_t", [["u8", "count", 10]])
362 # Define an embedded VLA type
364 "vl_api_valist_t", [["u8", "count"], ["u8", "string", 0, "count"]]
367 vamessage = VPPMessage(
368 "vamsg", [["vl_api_valist_t", "valist"], ["u8", "is_something"]]
371 message = VPPMessage("msg", [["vl_api_list_t", "list"], ["u8", "is_something"]])
373 # Pack message without VLA specified
374 b = message.pack({"is_something": 1})
375 b = vamessage.pack({"is_something": 1})
377 def test_arrays(self):
380 # 2. Fixed list of variable length sub type
381 # 3. Variable length type
383 s = VPPType("str", [["u32", "length"], ["u8", "string", 0, "length"]])
385 ip4 = VPPType("ip4_address", [["u8", "address", 4]])
386 listip4 = VPPType("list_ip4_t", [["ip4_address", "addresses", 4]])
388 "list_ip4_t", [["u8", "count"], ["ip4_address", "addresses", 0, "count"]]
391 valistip4_legacy = VPPType(
392 "list_ip4_t", [["u8", "foo"], ["ip4_address", "addresses", 0]]
397 addresses.append({"address": inet_pton(AF_INET, "2.2.2.2")})
398 b = listip4.pack({"addresses": addresses})
399 self.assertEqual(len(b), 16)
400 nt, size = listip4.unpack(b)
401 self.assertEqual(nt.addresses[0].address, inet_pton(AF_INET, "2.2.2.2"))
403 b = valistip4.pack({"count": len(addresses), "addresses": addresses})
404 self.assertEqual(len(b), 17)
406 nt, size = valistip4.unpack(b)
407 self.assertEqual(nt.count, 4)
408 self.assertEqual(nt.addresses[0].address, inet_pton(AF_INET, "2.2.2.2"))
410 b = valistip4_legacy.pack({"foo": 1, "addresses": addresses})
411 self.assertEqual(len(b), 17)
412 nt, size = valistip4_legacy.unpack(b)
413 self.assertEqual(len(nt.addresses), 4)
414 self.assertEqual(nt.addresses[0].address, inet_pton(AF_INET, "2.2.2.2"))
416 string = "foobar foobar"
417 b = s.pack({"length": len(string), "string": string.encode("utf-8")})
418 nt, size = s.unpack(b)
419 self.assertEqual(len(b), size)
421 def test_string(self):
422 s = VPPType("str", [["u32", "length"], ["u8", "string", 0, "length"]])
425 b = s.pack({"length": len(string), "string": string.encode("utf-8")})
426 nt, size = s.unpack(b)
427 self.assertEqual(len(b), size)
429 # Try same with VLA u8
430 byte_array = [b"\0"] * (10)
431 vla_u8 = VPPType("vla_u8", [["u8", "length"], ["u8", "data", 0, "length"]])
432 b = vla_u8.pack({"length": len(byte_array), "data": byte_array})
433 nt, size = vla_u8.unpack(b)
435 # VLA Array of fixed length strings
436 fixed_string = VPPType("fixed_string", [["string", "data", 32]])
438 "string_vla", [["u32", "length"], ["fixed_string", "services", 0, "length"]]
441 string_list = [{"data": "foobar1"}, {"data": "foobar2"}]
442 b = s.pack({"length": 2, "services": string_list})
443 nt, size = s.unpack(b)
446 fixed_u8 = VPPType("fixed_u8", [["u8", "data", 32]])
448 "u8_vla", [["u32", "length"], ["fixed_string", "services", 0, "length"]]
451 u8_list = [{"data": "foobar1"}, {"data": "foobar2"}]
452 b = s.pack({"length": 2, "services": u8_list})
453 nt, size = s.unpack(b)
455 def test_message(self):
459 ["u16", "_vl_msg_id"],
460 ["u8", "client_index"],
462 {"crc": "0x559b9f3c"},
465 b = foo.pack({"_vl_msg_id": 1, "client_index": 5, "something": 200})
466 nt, size = foo.unpack(b)
467 self.assertEqual(len(b), size)
468 self.assertEqual(nt.something, 200)
471 fib_mpls_label = VPPType(
472 "vl_api_fib_mpls_label_t",
473 [["u8", "is_uniform"], ["u32", "label"], ["u8", "ttl"], ["u8", "exp"]],
476 label_stack = {"is_uniform": 0, "label": 0, "ttl": 0, "exp": 0}
478 b = fib_mpls_label.pack(label_stack)
479 self.assertEqual(len(b), 7)
484 ["u32", "sw_if_index"],
487 ["u8", "preference"],
490 ["u8", "is_udp_encap"],
491 ["u8", "is_unreach"],
492 ["u8", "is_prohibit"],
493 ["u8", "is_resolve_host"],
494 ["u8", "is_resolve_attached"],
496 ["u8", "is_source_lookup"],
498 ["u8", "next_hop", 16],
499 ["u32", "next_hop_id"],
501 ["u32", "via_label"],
503 ["vl_api_fib_mpls_label_t", "label_stack", 16],
506 label_stack_list = []
508 label_stack_list.append(label_stack)
512 "next_hop": b"\x10\x02\x02\xac",
516 "next_hop_id": 4294967295,
517 "label_stack": label_stack_list,
519 "sw_if_index": 4294967295,
523 b = fib_path.pack(paths)
524 self.assertEqual(len(b), (7 * 16) + 49)
526 abf_policy = VPPType(
527 "vl_api_abf_policy_t",
529 ["u32", "policy_id"],
530 ["u32", "acl_index"],
532 ["vl_api_fib_path_t", "paths", 0, "n_paths"],
536 policy = {"n_paths": 1, "paths": [paths], "acl_index": 0, "policy_id": 10}
538 b = abf_policy.pack(policy)
539 self.assertEqual(len(b), (7 * 16) + 49 + 9)
541 abf_policy_add_del = VPPMessage(
542 "abf_policy_add_del",
544 ["u16", "_vl_msg_id"],
545 ["u32", "client_index"],
548 ["vl_api_abf_policy_t", "policy"],
552 b = abf_policy_add_del.pack(
553 {"is_add": 1, "context": 66, "_vl_msg_id": 1066, "policy": policy}
556 nt, size = abf_policy_add_del.unpack(b)
558 nt.policy.paths[0].next_hop,
559 b"\x10\x02\x02\xac\x00\x00\x00\x00" b"\x00\x00\x00\x00\x00\x00\x00\x00",
563 bier_table_id = VPPType(
564 "vl_api_bier_table_id_t",
565 [["u8", "bt_set"], ["u8", "bt_sub_domain"], ["u8", "bt_hdr_len_id"]],
568 bier_imp_add = VPPMessage(
571 ["u32", "client_index"],
573 ["vl_api_bier_table_id_t", "bi_tbl_id"],
575 ["u8", "bi_n_bytes"],
576 ["u8", "bi_bytes", 0, "bi_n_bytes"],
580 table_id = {"bt_set": 0, "bt_sub_domain": 0, "bt_hdr_len_id": 0}
584 b = bier_imp_add.pack(
585 {"bi_tbl_id": table_id, "bi_n_bytes": len(bibytes), "bi_bytes": bibytes}
588 self.assertEqual(len(b), 20)
594 ["EID_TYPE_API_PREFIX", 0],
595 ["EID_TYPE_API_MAC", 1],
596 ["EID_TYPE_API_NSH", 2],
601 VPPTypeAlias("vl_api_mac_address_t", {"type": "u8", "length": 6})
603 VPPType("vl_api_nsh_t", [["u32", "spi"], ["u8", "si"]])
606 "vl_api_address_family_t",
607 [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
609 VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
610 VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
612 "vl_api_address_union_t",
613 [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
618 [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
621 VPPType("vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]])
624 "vl_api_eid_address_t",
626 ["vl_api_prefix_t", "prefix"],
627 ["vl_api_mac_address_t", "mac"],
628 ["vl_api_nsh_t", "nsh"],
634 [["vl_api_eid_type_t", "type"], ["vl_api_eid_address_t", "address"]],
637 b = eid.pack({"type": 1, "address": {"mac": MACAddress("aa:bb:cc:dd:ee:ff")}})
638 self.assertEqual(len(b), 25)
639 nt, size = eid.unpack(b)
640 self.assertEqual(str(nt.address.mac), "aa:bb:cc:dd:ee:ff")
641 self.assertIsNone(nt.address.prefix)
644 class TestVppSerializerLogging(unittest.TestCase):
645 def test_logger(self):
646 # test logger name 'vpp_papi.serializer'
647 with self.assertRaises(VPPSerializerValueError) as ctx:
648 with self.assertLogs("vpp_papi.serializer", level="DEBUG") as cm:
650 "vl_api_eid_address_t",
652 ["vl_api_prefix_t", "prefix"],
653 ["vl_api_mac_address_t", "mac"],
654 ["vl_api_nsh_t", "nsh"],
658 cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"]
661 # test parent logger name 'vpp_papi'
662 with self.assertRaises(VPPSerializerValueError) as ctx:
663 with self.assertLogs("vpp_papi", level="DEBUG") as cm:
665 "vl_api_eid_address_t",
667 ["vl_api_prefix_t", "prefix"],
668 ["vl_api_mac_address_t", "mac"],
669 ["vl_api_nsh_t", "nsh"],
673 cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"]
677 if __name__ == "__main__":