vcl: allow more rx events on peek
[vpp.git] / src / vpp-api / python / vpp_papi / tests / test_vpp_serializer.py
1 #!/usr/bin/env python3
2
3 import unittest
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
9 import logging
10 import sys
11 from ipaddress import *
12
13
14 class TestLimits(unittest.TestCase):
15     def test_string(self):
16         fixed_string = VPPType("fixed_string", [["string", "name", 16]])
17
18         b = fixed_string.pack({"name": "foobar"})
19         self.assertEqual(len(b), 16)
20
21         # Ensure string is nul terminated
22         self.assertEqual(b.decode("ascii")[6], "\x00")
23
24         nt, size = fixed_string.unpack(b)
25         self.assertEqual(size, 16)
26         self.assertEqual(nt.name, "foobar")
27
28         # Empty string
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, "")
34
35         # String too long
36         with self.assertRaises(VPPSerializerValueError):
37             b = fixed_string.pack({"name": "foobarfoobar1234"})
38
39         variable_string = VPPType("variable_string", [["string", "name", 0]])
40         b = variable_string.pack({"name": "foobar"})
41         self.assertEqual(len(b), 4 + len("foobar"))
42
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"))
47
48     def test_limit(self):
49         limited_type = VPPType("limited_type_t", [["string", "name", 0, {"limit": 16}]])
50         unlimited_type = VPPType("limited_type_t", [["string", "name", 0]])
51
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)
56
57         with self.assertRaises(VPPSerializerValueError):
58             b = limited_type.pack({"name": "foobar" * 3})
59
60
61 class TestDefaults(unittest.TestCase):
62     def test_defaults(self):
63         default_type = VPPType(
64             "default_type_t", [["u16", "mtu", {"default": 1500, "limit": 0}]]
65         )
66         without_default_type = VPPType("without_default_type_t", [["u16", "mtu"]])
67
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)
73
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)
80
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)
87
88         # default enum type
89         VPPEnumType(
90             "vl_api_enum_t",
91             [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
92         )
93
94         default_with_enum = VPPType(
95             "default_enum_type_t",
96             [["u16", "mtu"], ["vl_api_enum_t", "e", {"default": 1}]],
97         )
98
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)
104
105
106 class TestAddType(unittest.TestCase):
107     def test_union(self):
108         un = VPPUnionType("test_union", [["u8", "is_bool"], ["u32", "is_int"]])
109
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)
115
116     def test_address(self):
117         af = VPPEnumType(
118             "vl_api_address_family_t",
119             [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
120         )
121         aff = VPPEnumFlagType(
122             "vl_api_address_family_flag_t",
123             [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
124         )
125         ip4 = VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
126         ip6 = VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
127         VPPUnionType(
128             "vl_api_address_union_t",
129             [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
130         )
131
132         address = VPPType(
133             "vl_api_address_t",
134             [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
135         )
136
137         prefix = VPPType(
138             "vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]]
139         )
140
141         va_address_list = VPPType(
142             "list_addresses",
143             [["u8", "count"], ["vl_api_address_t", "addresses", 0, "count"]],
144         )
145
146         message_with_va_address_list = VPPType(
147             "msg_with_vla", [["list_addresses", "vla_address"], ["u8", "is_cool"]]
148         )
149
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")
154
155         b = ip6.pack(inet_pton(AF_INET6, "1::1"))
156         self.assertEqual(len(b), 16)
157
158         b = address.pack(
159             {"af": af.ADDRESS_IP4, "un": {"ip4": inet_pton(AF_INET, "2.2.2.2")}}
160         )
161         self.assertEqual(len(b), 20)
162
163         nt, size = address.unpack(b)
164         self.assertEqual(str(nt), "2.2.2.2")
165
166         # List of addresses
167         address_list = []
168         for i in range(4):
169             address_list.append(
170                 {"af": af.ADDRESS_IP4, "un": {"ip4": inet_pton(AF_INET, "2.2.2.2")}}
171             )
172         b = va_address_list.pack(
173             {"count": len(address_list), "addresses": address_list}
174         )
175         self.assertEqual(len(b), 81)
176
177         nt, size = va_address_list.unpack(b)
178         self.assertEqual(str(nt.addresses[0]), "2.2.2.2")
179
180         b = message_with_va_address_list.pack(
181             {
182                 "vla_address": {"count": len(address_list), "addresses": address_list},
183                 "is_cool": 100,
184             }
185         )
186         self.assertEqual(len(b), 82)
187         nt, size = message_with_va_address_list.unpack(b)
188         self.assertEqual(nt.is_cool, 100)
189
190     def test_address_with_prefix(self):
191         af = VPPEnumType(
192             "vl_api_address_family_t",
193             [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
194         )
195         ip4 = VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
196         ip6 = VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
197         VPPUnionType(
198             "vl_api_address_union_t",
199             [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
200         )
201
202         address = VPPType(
203             "vl_api_address_t",
204             [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
205         )
206
207         prefix = VPPType(
208             "vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]]
209         )
210         prefix4 = VPPType(
211             "vl_api_ip4_prefix_t", [["vl_api_ip4_address_t", "address"], ["u8", "len"]]
212         )
213         prefix6 = VPPType(
214             "vl_api_ip6_prefix_t", [["vl_api_ip6_address_t", "address"], ["u8", "len"]]
215         )
216
217         address_with_prefix = VPPTypeAlias(
218             "vl_api_address_with_prefix_t", {"type": "vl_api_prefix_t"}
219         )
220         address4_with_prefix = VPPTypeAlias(
221             "vl_api_ip4_address_with_prefix_t", {"type": "vl_api_ip4_prefix_t"}
222         )
223         address6_with_prefix = VPPTypeAlias(
224             "vl_api_ip6_address_with_prefix_t", {"type": "vl_api_ip6_prefix_t"}
225         )
226
227         awp_type = VPPType("foobar_t", [["vl_api_address_with_prefix_t", "address"]])
228
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")
235
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")
241
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")
247
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)
255
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)
263
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")
269
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")
275
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")
281
282     def test_recursive_address(self):
283         af = VPPEnumType(
284             "vl_api_address_family_t",
285             [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
286         )
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)
291
292         self.assertEqual(str(nt), "1.1.1.1")
293
294         ip6 = VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
295         VPPUnionType(
296             "vl_api_address_union_t",
297             [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
298         )
299
300         address = VPPType(
301             "vl_api_address_t",
302             [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
303         )
304
305         prefix = VPPType(
306             "vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]]
307         )
308         message = VPPMessage("svs", [["vl_api_prefix_t", "prefix"]])
309         message_addr = VPPMessage("svs_address", [["vl_api_address_t", "address"]])
310
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))
319
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))
324
325         message_array = VPPMessage(
326             "address_array", [["vl_api_ip6_address_t", "addresses", 2]]
327         )
328         b = message_array.pack({"addresses": [IPv6Address("1::1"), "2::2"]})
329         self.assertEqual(len(b), 32)
330         message_array_vla = VPPMessage(
331             "address_array_vla",
332             [["u32", "num"], ["vl_api_ip6_address_t", "addresses", 0, "num"]],
333         )
334         b = message_array_vla.pack({"addresses": ["1::1", "2::2"], "num": 2})
335         self.assertEqual(len(b), 36)
336
337         message_array4 = VPPMessage(
338             "address_array4", [["vl_api_ip4_address_t", "addresses", 2]]
339         )
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)
344
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)
357
358     def test_zero_vla(self):
359         """Default zero'ed out for VLAs"""
360         list = VPPType("vl_api_list_t", [["u8", "count", 10]])
361
362         # Define an embedded VLA type
363         valist = VPPType(
364             "vl_api_valist_t", [["u8", "count"], ["u8", "string", 0, "count"]]
365         )
366         # Define a message
367         vamessage = VPPMessage(
368             "vamsg", [["vl_api_valist_t", "valist"], ["u8", "is_something"]]
369         )
370
371         message = VPPMessage("msg", [["vl_api_list_t", "list"], ["u8", "is_something"]])
372
373         # Pack message without VLA specified
374         b = message.pack({"is_something": 1})
375         b = vamessage.pack({"is_something": 1})
376
377     def test_arrays(self):
378         # Test cases
379         # 1. Fixed list
380         # 2. Fixed list of variable length sub type
381         # 3. Variable length type
382         #
383         s = VPPType("str", [["u32", "length"], ["u8", "string", 0, "length"]])
384
385         ip4 = VPPType("ip4_address", [["u8", "address", 4]])
386         listip4 = VPPType("list_ip4_t", [["ip4_address", "addresses", 4]])
387         valistip4 = VPPType(
388             "list_ip4_t", [["u8", "count"], ["ip4_address", "addresses", 0, "count"]]
389         )
390
391         valistip4_legacy = VPPType(
392             "list_ip4_t", [["u8", "foo"], ["ip4_address", "addresses", 0]]
393         )
394
395         addresses = []
396         for i in range(4):
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"))
402
403         b = valistip4.pack({"count": len(addresses), "addresses": addresses})
404         self.assertEqual(len(b), 17)
405
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"))
409
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"))
415
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)
420
421     def test_string(self):
422         s = VPPType("str", [["u32", "length"], ["u8", "string", 0, "length"]])
423
424         string = ""
425         b = s.pack({"length": len(string), "string": string.encode("utf-8")})
426         nt, size = s.unpack(b)
427         self.assertEqual(len(b), size)
428
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)
434
435         # VLA Array of fixed length strings
436         fixed_string = VPPType("fixed_string", [["string", "data", 32]])
437         s = VPPType(
438             "string_vla", [["u32", "length"], ["fixed_string", "services", 0, "length"]]
439         )
440
441         string_list = [{"data": "foobar1"}, {"data": "foobar2"}]
442         b = s.pack({"length": 2, "services": string_list})
443         nt, size = s.unpack(b)
444
445         # Try same with u8
446         fixed_u8 = VPPType("fixed_u8", [["u8", "data", 32]])
447         s = VPPType(
448             "u8_vla", [["u32", "length"], ["fixed_string", "services", 0, "length"]]
449         )
450
451         u8_list = [{"data": "foobar1"}, {"data": "foobar2"}]
452         b = s.pack({"length": 2, "services": u8_list})
453         nt, size = s.unpack(b)
454
455     def test_message(self):
456         foo = VPPMessage(
457             "foo",
458             [
459                 ["u16", "_vl_msg_id"],
460                 ["u8", "client_index"],
461                 ["u8", "something"],
462                 {"crc": "0x559b9f3c"},
463             ],
464         )
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)
469
470     def test_abf(self):
471         fib_mpls_label = VPPType(
472             "vl_api_fib_mpls_label_t",
473             [["u8", "is_uniform"], ["u32", "label"], ["u8", "ttl"], ["u8", "exp"]],
474         )
475
476         label_stack = {"is_uniform": 0, "label": 0, "ttl": 0, "exp": 0}
477
478         b = fib_mpls_label.pack(label_stack)
479         self.assertEqual(len(b), 7)
480
481         fib_path = VPPType(
482             "vl_api_fib_path_t",
483             [
484                 ["u32", "sw_if_index"],
485                 ["u32", "table_id"],
486                 ["u8", "weight"],
487                 ["u8", "preference"],
488                 ["u8", "is_local"],
489                 ["u8", "is_drop"],
490                 ["u8", "is_udp_encap"],
491                 ["u8", "is_unreach"],
492                 ["u8", "is_prohibit"],
493                 ["u8", "is_resolve_host"],
494                 ["u8", "is_resolve_attached"],
495                 ["u8", "is_dvr"],
496                 ["u8", "is_source_lookup"],
497                 ["u8", "afi"],
498                 ["u8", "next_hop", 16],
499                 ["u32", "next_hop_id"],
500                 ["u32", "rpf_id"],
501                 ["u32", "via_label"],
502                 ["u8", "n_labels"],
503                 ["vl_api_fib_mpls_label_t", "label_stack", 16],
504             ],
505         )
506         label_stack_list = []
507         for i in range(16):
508             label_stack_list.append(label_stack)
509
510         paths = {
511             "is_udp_encap": 0,
512             "next_hop": b"\x10\x02\x02\xac",
513             "table_id": 0,
514             "afi": 0,
515             "weight": 1,
516             "next_hop_id": 4294967295,
517             "label_stack": label_stack_list,
518             "n_labels": 0,
519             "sw_if_index": 4294967295,
520             "preference": 0,
521         }
522
523         b = fib_path.pack(paths)
524         self.assertEqual(len(b), (7 * 16) + 49)
525
526         abf_policy = VPPType(
527             "vl_api_abf_policy_t",
528             [
529                 ["u32", "policy_id"],
530                 ["u32", "acl_index"],
531                 ["u8", "n_paths"],
532                 ["vl_api_fib_path_t", "paths", 0, "n_paths"],
533             ],
534         )
535
536         policy = {"n_paths": 1, "paths": [paths], "acl_index": 0, "policy_id": 10}
537
538         b = abf_policy.pack(policy)
539         self.assertEqual(len(b), (7 * 16) + 49 + 9)
540
541         abf_policy_add_del = VPPMessage(
542             "abf_policy_add_del",
543             [
544                 ["u16", "_vl_msg_id"],
545                 ["u32", "client_index"],
546                 ["u32", "context"],
547                 ["u8", "is_add"],
548                 ["vl_api_abf_policy_t", "policy"],
549             ],
550         )
551
552         b = abf_policy_add_del.pack(
553             {"is_add": 1, "context": 66, "_vl_msg_id": 1066, "policy": policy}
554         )
555
556         nt, size = abf_policy_add_del.unpack(b)
557         self.assertEqual(
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",
560         )
561
562     def test_bier(self):
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"]],
566         )
567
568         bier_imp_add = VPPMessage(
569             "bier_imp_add",
570             [
571                 ["u32", "client_index"],
572                 ["u32", "context"],
573                 ["vl_api_bier_table_id_t", "bi_tbl_id"],
574                 ["u16", "bi_src"],
575                 ["u8", "bi_n_bytes"],
576                 ["u8", "bi_bytes", 0, "bi_n_bytes"],
577             ],
578         )
579
580         table_id = {"bt_set": 0, "bt_sub_domain": 0, "bt_hdr_len_id": 0}
581
582         bibytes = b"foobar"
583
584         b = bier_imp_add.pack(
585             {"bi_tbl_id": table_id, "bi_n_bytes": len(bibytes), "bi_bytes": bibytes}
586         )
587
588         self.assertEqual(len(b), 20)
589
590     def test_lisp(self):
591         VPPEnumType(
592             "vl_api_eid_type_t",
593             [
594                 ["EID_TYPE_API_PREFIX", 0],
595                 ["EID_TYPE_API_MAC", 1],
596                 ["EID_TYPE_API_NSH", 2],
597                 {"enumtype": "u32"},
598             ],
599         )
600
601         VPPTypeAlias("vl_api_mac_address_t", {"type": "u8", "length": 6})
602
603         VPPType("vl_api_nsh_t", [["u32", "spi"], ["u8", "si"]])
604
605         VPPEnumType(
606             "vl_api_address_family_t",
607             [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}],
608         )
609         VPPTypeAlias("vl_api_ip4_address_t", {"type": "u8", "length": 4})
610         VPPTypeAlias("vl_api_ip6_address_t", {"type": "u8", "length": 16})
611         VPPUnionType(
612             "vl_api_address_union_t",
613             [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]],
614         )
615
616         VPPType(
617             "vl_api_address_t",
618             [["vl_api_address_family_t", "af"], ["vl_api_address_union_t", "un"]],
619         )
620
621         VPPType("vl_api_prefix_t", [["vl_api_address_t", "address"], ["u8", "len"]])
622
623         VPPUnionType(
624             "vl_api_eid_address_t",
625             [
626                 ["vl_api_prefix_t", "prefix"],
627                 ["vl_api_mac_address_t", "mac"],
628                 ["vl_api_nsh_t", "nsh"],
629             ],
630         )
631
632         eid = VPPType(
633             "vl_api_eid_t",
634             [["vl_api_eid_type_t", "type"], ["vl_api_eid_address_t", "address"]],
635         )
636
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)
642
643
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:
649                 u = VPPUnionType(
650                     "vl_api_eid_address_t",
651                     [
652                         ["vl_api_prefix_t", "prefix"],
653                         ["vl_api_mac_address_t", "mac"],
654                         ["vl_api_nsh_t", "nsh"],
655                     ],
656                 )
657         self.assertEqual(
658             cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"]
659         )
660
661         # test parent logger name 'vpp_papi'
662         with self.assertRaises(VPPSerializerValueError) as ctx:
663             with self.assertLogs("vpp_papi", level="DEBUG") as cm:
664                 u = VPPUnionType(
665                     "vl_api_eid_address_t",
666                     [
667                         ["vl_api_prefix_t", "prefix"],
668                         ["vl_api_mac_address_t", "mac"],
669                         ["vl_api_nsh_t", "nsh"],
670                     ],
671                 )
672         self.assertEqual(
673             cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"]
674         )
675
676
677 if __name__ == "__main__":
678     unittest.main()