X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvpp-api%2Fpython%2Fvpp_papi%2Ftests%2Ftest_vpp_serializer.py;h=c9b3d672d6a496dfda49a5ef9dac461ad13156da;hb=3825d93af;hp=4e8a417c6fdcef196558e88554fec1072aeb7797;hpb=c84cbad785fa11c1dba92b8e87ad234d779d1cbd;p=vpp.git diff --git a/src/vpp-api/python/vpp_papi/tests/test_vpp_serializer.py b/src/vpp-api/python/vpp_papi/tests/test_vpp_serializer.py index 4e8a417c6fd..c9b3d672d6a 100755 --- a/src/vpp-api/python/vpp_papi/tests/test_vpp_serializer.py +++ b/src/vpp-api/python/vpp_papi/tests/test_vpp_serializer.py @@ -1,11 +1,108 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import unittest -from vpp_papi.vpp_serializer import VPPType, VPPEnumType +from vpp_papi.vpp_serializer import VPPType, VPPEnumType, VPPEnumFlagType from vpp_papi.vpp_serializer import VPPUnionType, VPPMessage +from vpp_papi.vpp_serializer import VPPTypeAlias, VPPSerializerValueError +from vpp_papi import MACAddress from socket import inet_pton, AF_INET, AF_INET6 import logging import sys +from ipaddress import * + + +class TestLimits(unittest.TestCase): + def test_string(self): + fixed_string = VPPType('fixed_string', + [['string', 'name', 16]]) + + b = fixed_string.pack({'name': 'foobar'}) + self.assertEqual(len(b), 16) + + # Ensure string is nul terminated + self.assertEqual(b.decode('ascii')[6], '\x00') + + nt, size = fixed_string.unpack(b) + self.assertEqual(size, 16) + self.assertEqual(nt.name, 'foobar') + + # Empty string + b = fixed_string.pack({'name': ''}) + self.assertEqual(len(b), 16) + nt, size = fixed_string.unpack(b) + self.assertEqual(size, 16) + self.assertEqual(nt.name, '') + + # String too long + with self.assertRaises(VPPSerializerValueError): + b = fixed_string.pack({'name': 'foobarfoobar1234'}) + + variable_string = VPPType('variable_string', + [['string', 'name', 0]]) + b = variable_string.pack({'name': 'foobar'}) + self.assertEqual(len(b), 4 + len('foobar')) + + nt, size = variable_string.unpack(b) + self.assertEqual(size, 4 + len('foobar')) + self.assertEqual(nt.name, 'foobar') + self.assertEqual(len(nt.name), len('foobar')) + + def test_limit(self): + limited_type = VPPType('limited_type_t', + [['string', 'name', 0, {'limit': 16}]]) + unlimited_type = VPPType('limited_type_t', + [['string', 'name', 0]]) + + b = limited_type.pack({'name': 'foobar'}) + self.assertEqual(len(b), 10) + b = unlimited_type.pack({'name': 'foobar'}) + self.assertEqual(len(b), 10) + + with self.assertRaises(VPPSerializerValueError): + b = limited_type.pack({'name': 'foobar'*3}) + + +class TestDefaults(unittest.TestCase): + def test_defaults(self): + default_type = VPPType('default_type_t', + [['u16', 'mtu', {'default': 1500, 'limit': 0}]]) + without_default_type = VPPType('without_default_type_t', + [['u16', 'mtu']]) + + b = default_type.pack({}) + self.assertEqual(len(b), 2) + nt, size = default_type.unpack(b) + self.assertEqual(len(b), size) + self.assertEqual(nt.mtu, 1500) + + # distinguish between parameter 0 and parameter not passed + b = default_type.pack({'mtu': 0}) + self.assertEqual(len(b), 2) + nt, size = default_type.unpack(b) + self.assertEqual(len(b), size) + self.assertEqual(nt.mtu, 0) + + # Ensure that basetypes does not inherit default + b = without_default_type.pack({}) + self.assertEqual(len(b), 2) + nt, size = default_type.unpack(b) + self.assertEqual(len(b), size) + self.assertEqual(nt.mtu, 0) + + # default enum type + VPPEnumType('vl_api_enum_t', [["ADDRESS_IP4", 0], + ["ADDRESS_IP6", 1], + {"enumtype": "u32"}]) + + default_with_enum = VPPType('default_enum_type_t', + [['u16', 'mtu'], ['vl_api_enum_t', + 'e', {'default': 1}]]) + + b = default_with_enum.pack({}) + self.assertEqual(len(b), 6) + nt, size = default_with_enum.unpack(b) + self.assertEqual(len(b), size) + self.assertEqual(nt.e, 1) class TestAddType(unittest.TestCase): @@ -25,8 +122,13 @@ class TestAddType(unittest.TestCase): af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0], ["ADDRESS_IP6", 1], {"enumtype": "u32"}]) - ip4 = VPPType('vl_api_ip4_address_t', [['u8', 'address', 4]]) - ip6 = VPPType('vl_api_ip6_address_t', [['u8', 'address', 16]]) + aff = VPPEnumFlagType('vl_api_address_family_flag_t', [["ADDRESS_IP4", 0], + ["ADDRESS_IP6", 1], + {"enumtype": "u32"}]) + ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8', + 'length': 4}) + ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8', + 'length': 16}) VPPUnionType('vl_api_address_union_t', [["vl_api_ip4_address_t", "ip4"], ["vl_api_ip6_address_t", "ip6"]]) @@ -35,6 +137,10 @@ class TestAddType(unittest.TestCase): [['vl_api_address_family_t', 'af'], ['vl_api_address_union_t', 'un']]) + prefix = VPPType('vl_api_prefix_t', + [['vl_api_address_t', 'address'], + ['u8', 'len']]) + va_address_list = VPPType('list_addresses', [['u8', 'count'], ['vl_api_address_t', 'addresses', @@ -45,41 +151,34 @@ class TestAddType(unittest.TestCase): 'vla_address'], ['u8', 'is_cool']]) - b = ip4.pack({'address': inet_pton(AF_INET, '1.1.1.1')}) + b = ip4.pack(inet_pton(AF_INET, '1.1.1.1')) self.assertEqual(len(b), 4) nt, size = ip4.unpack(b) - self.assertEqual(nt.address, inet_pton(AF_INET, '1.1.1.1')) + self.assertEqual(str(nt), '1.1.1.1') - b = ip6.pack({'address': inet_pton(AF_INET6, '1::1')}) + b = ip6.pack(inet_pton(AF_INET6, '1::1')) self.assertEqual(len(b), 16) b = address.pack({'af': af.ADDRESS_IP4, 'un': - {'ip4': - {'address': inet_pton(AF_INET, '2.2.2.2')}}}) + {'ip4': inet_pton(AF_INET, '2.2.2.2')}}) self.assertEqual(len(b), 20) nt, size = address.unpack(b) - self.assertEqual(nt.af, af.ADDRESS_IP4) - self.assertEqual(nt.un.ip4.address, - inet_pton(AF_INET, '2.2.2.2')) - self.assertEqual(nt.un.ip6.address, - inet_pton(AF_INET6, '0202:0202::')) + self.assertEqual(str(nt), '2.2.2.2') # List of addresses address_list = [] for i in range(4): address_list.append({'af': af.ADDRESS_IP4, 'un': - {'ip4': - {'address': inet_pton(AF_INET, '2.2.2.2')}}}) + {'ip4': inet_pton(AF_INET, '2.2.2.2')}}) b = va_address_list.pack({'count': len(address_list), 'addresses': address_list}) self.assertEqual(len(b), 81) nt, size = va_address_list.unpack(b) - self.assertEqual(nt.addresses[0].un.ip4.address, - inet_pton(AF_INET, '2.2.2.2')) + self.assertEqual(str(nt.addresses[0]), '2.2.2.2') b = message_with_va_address_list.pack({'vla_address': {'count': len(address_list), @@ -89,6 +188,194 @@ class TestAddType(unittest.TestCase): nt, size = message_with_va_address_list.unpack(b) self.assertEqual(nt.is_cool, 100) + def test_address_with_prefix(self): + af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0], + ["ADDRESS_IP6", 1], + {"enumtype": "u32"}]) + ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8', + 'length': 4}) + ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8', + 'length': 16}) + VPPUnionType('vl_api_address_union_t', + [["vl_api_ip4_address_t", "ip4"], + ["vl_api_ip6_address_t", "ip6"]]) + + address = VPPType('vl_api_address_t', + [['vl_api_address_family_t', 'af'], + ['vl_api_address_union_t', 'un']]) + + prefix = VPPType('vl_api_prefix_t', + [['vl_api_address_t', 'address'], + ['u8', 'len']]) + prefix4 = VPPType('vl_api_ip4_prefix_t', + [['vl_api_ip4_address_t', 'address'], + ['u8', 'len']]) + prefix6 = VPPType('vl_api_ip6_prefix_t', + [['vl_api_ip6_address_t', 'address'], + ['u8', 'len']]) + + address_with_prefix = VPPTypeAlias('vl_api_address_with_prefix_t', {'type': 'vl_api_prefix_t' }) + address4_with_prefix = VPPTypeAlias('vl_api_ip4_address_with_prefix_t', + {'type': 'vl_api_ip4_prefix_t' }) + address6_with_prefix = VPPTypeAlias('vl_api_ip6_address_with_prefix_t', + {'type': 'vl_api_ip6_prefix_t' }) + + awp_type = VPPType('foobar_t', + [['vl_api_address_with_prefix_t', 'address']]) + + # address with prefix + b = address_with_prefix.pack(IPv4Interface('2.2.2.2/24')) + self.assertEqual(len(b), 21) + nt, size = address_with_prefix.unpack(b) + self.assertTrue(isinstance(nt, IPv4Interface)) + self.assertEqual(str(nt), '2.2.2.2/24') + + b = address_with_prefix.pack(IPv6Interface('2::2/64')) + self.assertEqual(len(b), 21) + nt, size = address_with_prefix.unpack(b) + self.assertTrue(isinstance(nt, IPv6Interface)) + self.assertEqual(str(nt), '2::2/64') + + b = address_with_prefix.pack(IPv4Network('2.2.2.2/24', strict=False)) + self.assertEqual(len(b), 21) + nt, size = address_with_prefix.unpack(b) + self.assertTrue(isinstance(nt, IPv4Interface)) + self.assertEqual(str(nt), '2.2.2.0/24') + + b = address4_with_prefix.pack('2.2.2.2/24') + self.assertEqual(len(b), 5) + nt, size = address4_with_prefix.unpack(b) + self.assertTrue(isinstance(nt, IPv4Interface)) + self.assertEqual(str(nt), '2.2.2.2/24') + b = address4_with_prefix.pack(IPv4Interface('2.2.2.2/24')) + self.assertEqual(len(b), 5) + + b = address6_with_prefix.pack('2::2/64') + self.assertEqual(len(b), 17) + nt, size = address6_with_prefix.unpack(b) + self.assertTrue(isinstance(nt, IPv6Interface)) + self.assertEqual(str(nt), '2::2/64') + b = address6_with_prefix.pack(IPv6Interface('2::2/64')) + self.assertEqual(len(b), 17) + + b = prefix.pack('192.168.10.0/24') + self.assertEqual(len(b), 21) + nt, size = prefix.unpack(b) + self.assertTrue(isinstance(nt, IPv4Network)) + self.assertEqual(str(nt), '192.168.10.0/24') + + b = awp_type.pack({'address': '1.2.3.4/24'}) + self.assertEqual(len(b), 21) + nt, size = awp_type.unpack(b) + self.assertTrue(isinstance(nt.address, IPv4Interface)) + self.assertEqual(str(nt.address), '1.2.3.4/24') + + b = awp_type.pack({'address': IPv4Interface('1.2.3.4/24')}) + self.assertEqual(len(b), 21) + nt, size = awp_type.unpack(b) + self.assertTrue(isinstance(nt.address, IPv4Interface)) + self.assertEqual(str(nt.address), '1.2.3.4/24') + + def test_recursive_address(self): + af = VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0], + ["ADDRESS_IP6", 1], + {"enumtype": "u32"}]) + ip4 = VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8', + 'length': 4}) + b = ip4.pack('1.1.1.1') + self.assertEqual(len(b), 4) + nt, size = ip4.unpack(b) + + self.assertEqual(str(nt), '1.1.1.1') + + ip6 = VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8', + 'length': 16}) + VPPUnionType('vl_api_address_union_t', + [["vl_api_ip4_address_t", "ip4"], + ["vl_api_ip6_address_t", "ip6"]]) + + address = VPPType('vl_api_address_t', + [['vl_api_address_family_t', 'af'], + ['vl_api_address_union_t', 'un']]) + + prefix = VPPType('vl_api_prefix_t', + [['vl_api_address_t', 'address'], + ['u8', 'len']]) + message = VPPMessage('svs', + [['vl_api_prefix_t', 'prefix']]) + message_addr = VPPMessage('svs_address', + [['vl_api_address_t', 'address']]) + + b = message_addr.pack({'address': "1::1"}) + self.assertEqual(len(b), 20) + nt, size = message_addr.unpack(b) + self.assertEqual("1::1", str(nt.address)) + b = message_addr.pack({'address': "1.1.1.1"}) + self.assertEqual(len(b), 20) + nt, size = message_addr.unpack(b) + self.assertEqual("1.1.1.1", str(nt.address)) + + b = message.pack({'prefix': "1.1.1.0/24"}) + self.assertEqual(len(b), 21) + nt, size = message.unpack(b) + self.assertEqual("1.1.1.0/24", str(nt.prefix)) + + message_array = VPPMessage('address_array', + [['vl_api_ip6_address_t', + 'addresses', 2]]) + b = message_array.pack({'addresses': [IPv6Address(u"1::1"), "2::2"]}) + self.assertEqual(len(b), 32) + message_array_vla = VPPMessage('address_array_vla', + [['u32', 'num'], + ['vl_api_ip6_address_t', + 'addresses', 0, 'num']]) + b = message_array_vla.pack({'addresses': ["1::1", "2::2"], 'num': 2}) + self.assertEqual(len(b), 36) + + message_array4 = VPPMessage('address_array4', + [['vl_api_ip4_address_t', + 'addresses', 2]]) + b = message_array4.pack({'addresses': ["1.1.1.1", "2.2.2.2"]}) + self.assertEqual(len(b), 8) + b = message_array4.pack({'addresses': [IPv4Address(u"1.1.1.1"), + "2.2.2.2"]}) + self.assertEqual(len(b), 8) + + message = VPPMessage('address', [['vl_api_address_t', 'address']]) + b = message.pack({'address': '1::1'}) + self.assertEqual(len(b), 20) + b = message.pack({'address': '1.1.1.1'}) + self.assertEqual(len(b), 20) + message = VPPMessage('prefix', [['vl_api_prefix_t', 'prefix']]) + b = message.pack({'prefix': '1::1/130'}) + self.assertEqual(len(b), 21) + b = message.pack({'prefix': IPv6Network(u'1::/119')}) + self.assertEqual(len(b), 21) + b = message.pack({'prefix': IPv4Network(u'1.1.0.0/16')}) + self.assertEqual(len(b), 21) + + def test_zero_vla(self): + '''Default zero'ed out for VLAs''' + list = VPPType('vl_api_list_t', + [['u8', 'count', 10]]) + + # Define an embedded VLA type + valist = VPPType('vl_api_valist_t', + [['u8', 'count'], + ['u8', 'string', 0, 'count']]) + # Define a message + vamessage = VPPMessage('vamsg', + [['vl_api_valist_t', 'valist'], + ['u8', 'is_something']]) + + message = VPPMessage('msg', + [['vl_api_list_t', 'list'], + ['u8', 'is_something']]) + + # Pack message without VLA specified + b = message.pack({'is_something': 1}) + b = vamessage.pack({'is_something': 1}) + def test_arrays(self): # Test cases # 1. Fixed list @@ -133,7 +420,7 @@ class TestAddType(unittest.TestCase): inet_pton(AF_INET, '2.2.2.2')) string = 'foobar foobar' - b = s.pack({'length': len(string), 'string': string}) + b = s.pack({'length': len(string), 'string': string.encode('utf-8')}) nt, size = s.unpack(b) self.assertEqual(len(b), size) @@ -142,7 +429,7 @@ class TestAddType(unittest.TestCase): ['u8', 'string', 0, 'length']]) string = '' - b = s.pack({'length': len(string), 'string': string}) + b = s.pack({'length': len(string), 'string': string.encode('utf-8')}) nt, size = s.unpack(b) self.assertEqual(len(b), size) @@ -271,6 +558,78 @@ class TestAddType(unittest.TestCase): self.assertEqual(len(b), 20) + def test_lisp(self): + VPPEnumType('vl_api_eid_type_t', + [["EID_TYPE_API_PREFIX", 0], + ["EID_TYPE_API_MAC", 1], + ["EID_TYPE_API_NSH", 2], + {"enumtype": "u32"}]) + + VPPTypeAlias('vl_api_mac_address_t', {'type': 'u8', + 'length': 6}) + + VPPType('vl_api_nsh_t', + [["u32", "spi"], + ["u8", "si"]]) + + VPPEnumType('vl_api_address_family_t', [["ADDRESS_IP4", 0], + ["ADDRESS_IP6", 1], + {"enumtype": "u32"}]) + VPPTypeAlias('vl_api_ip4_address_t', {'type': 'u8', + 'length': 4}) + VPPTypeAlias('vl_api_ip6_address_t', {'type': 'u8', + 'length': 16}) + VPPUnionType('vl_api_address_union_t', + [["vl_api_ip4_address_t", "ip4"], + ["vl_api_ip6_address_t", "ip6"]]) + + VPPType('vl_api_address_t', + [['vl_api_address_family_t', 'af'], + ['vl_api_address_union_t', 'un']]) + + VPPType('vl_api_prefix_t', + [['vl_api_address_t', 'address'], + ['u8', 'len']]) + + VPPUnionType('vl_api_eid_address_t', + [["vl_api_prefix_t", "prefix"], + ["vl_api_mac_address_t", "mac"], + ["vl_api_nsh_t", "nsh"]]) + + eid = VPPType('vl_api_eid_t', + [["vl_api_eid_type_t", "type"], + ["vl_api_eid_address_t", "address"]]) + + b = eid.pack({'type':1, + 'address': { + 'mac': MACAddress('aa:bb:cc:dd:ee:ff')}}) + self.assertEqual(len(b), 25) + nt, size = eid.unpack(b) + self.assertEqual(str(nt.address.mac), 'aa:bb:cc:dd:ee:ff') + self.assertIsNone(nt.address.prefix) + + +class TestVppSerializerLogging(unittest.TestCase): + + def test_logger(self): + # test logger name 'vpp_papi.serializer' + with self.assertRaises(VPPSerializerValueError) as ctx: + with self.assertLogs('vpp_papi.serializer', level='DEBUG') as cm: + u = VPPUnionType('vl_api_eid_address_t', + [["vl_api_prefix_t", "prefix"], + ["vl_api_mac_address_t", "mac"], + ["vl_api_nsh_t", "nsh"]]) + self.assertEqual(cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"]) + + # test parent logger name 'vpp_papi' + with self.assertRaises(VPPSerializerValueError) as ctx: + with self.assertLogs('vpp_papi', level='DEBUG') as cm: + u = VPPUnionType('vl_api_eid_address_t', + [["vl_api_prefix_t", "prefix"], + ["vl_api_mac_address_t", "mac"], + ["vl_api_nsh_t", "nsh"]]) + self.assertEqual(cm.output, ["DEBUG:vpp_papi.serializer:Unknown union type vl_api_prefix_t"]) + if __name__ == '__main__': unittest.main()