import unittest
from vpp_papi.vpp_serializer import VPPType, VPPEnumType
from vpp_papi.vpp_serializer import VPPUnionType, VPPMessage
+from vpp_papi.vpp_format import VPPFormat
from socket import inet_pton, AF_INET, AF_INET6
import logging
import sys
nt, size = message_with_va_address_list.unpack(b)
self.assertEqual(nt.is_cool, 100)
+ def test_recursive_address(self):
+ 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]])
+ 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', 'address_length']])
+ 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", VPPFormat.unformat(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", VPPFormat.unformat(nt.address))
+
+ b = message.pack({'prefix': "1.1.1.1/24"})
+ self.assertEqual(len(b), 21)
+ nt, size = message.unpack(b)
+ self.assertEqual("1.1.1.1/24", VPPFormat.unformat(nt.prefix))
+
+ 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
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()})
nt, size = s.unpack(b)
self.assertEqual(len(b), size)
['u8', 'string', 0, 'length']])
string = ''
- b = s.pack({'length': len(string), 'string': string})
+ b = s.pack({'length': len(string), 'string': string.encode()})
nt, size = s.unpack(b)
self.assertEqual(len(b), size)
--- /dev/null
+#
+# Copyright (c) 2018 Cisco and/or its affiliates.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from socket import inet_pton, inet_ntop, AF_INET6, AF_INET
+
+
+class VPPFormat:
+ @staticmethod
+ def format_vl_api_ip6_prefix_t(args):
+ prefix, len = args.split('/')
+ return {'prefix': {'address': inet_pton(AF_INET6, prefix)},
+ 'len': int(len)}
+
+ @staticmethod
+ def unformat_vl_api_ip6_prefix_t(args):
+ return "{}/{}".format(inet_ntop(AF_INET6, args.prefix.address),
+ args.len)
+
+ @staticmethod
+ def format_vl_api_ip4_prefix_t(args):
+ prefix, len = args.split('/')
+ return {'prefix': {'address': inet_pton(AF_INET, prefix)},
+ 'len': int(len)}
+
+ @staticmethod
+ def unformat_vl_api_ip4_prefix_t(args):
+ return "{}/{}".format(inet_ntop(AF_INET, args.prefix.address),
+ args.len)
+
+ @staticmethod
+ def format_vl_api_ip6_address_t(args):
+ return {'address': inet_pton(AF_INET6, args)}
+
+ @staticmethod
+ def format_vl_api_ip4_address_t(args):
+ return {'address': inet_pton(AF_INET, args)}
+
+ @staticmethod
+ def format_vl_api_address_t(args):
+ try:
+ return {'un': {'ip6': {'address': inet_pton(AF_INET6, args)}},
+ 'af': int(1)}
+ except Exception as e:
+ return {'un': {'ip4': {'address': inet_pton(AF_INET, args)}},
+ 'af': int(0)}
+
+ @staticmethod
+ def unformat_vl_api_address_t(arg):
+ if arg.af == 1:
+ return inet_ntop(AF_INET6, arg.un.ip6.address)
+ if arg.af == 0:
+ return inet_ntop(AF_INET, arg.un.ip4.address)
+ raise
+
+ @staticmethod
+ def format_vl_api_prefix_t(args):
+ prefix, len = args.split('/')
+ return {'address': VPPFormat.format_vl_api_address_t(prefix),
+ 'address_length': int(len)}
+
+ @staticmethod
+ def unformat_vl_api_prefix_t(arg):
+ if arg.address.af == 1:
+ return "{}/{}".format(inet_ntop(AF_INET6,
+ arg.address.un.ip6.address),
+ arg.address_length)
+ if arg.address.af == 0:
+ return "{}/{}".format(inet_ntop(AF_INET,
+ arg.address.un.ip4.address),
+ arg.address_length)
+ raise
+
+ @staticmethod
+ def format_u8(args):
+ try:
+ return int(args)
+ except Exception as e:
+ return args.encode()
+
+ @staticmethod
+ def format(typename, args):
+ try:
+ return getattr(VPPFormat, 'format_' + typename)(args)
+ except AttributeError:
+ # Default
+ return (int(args))
+
+ @staticmethod
+ def unformat_bytes(args):
+ try:
+ return args.decode('utf-8')
+ except Exception as e:
+ return args
+
+ @staticmethod
+ def unformat_list(args):
+ s = '['
+ for f in args:
+ t = type(f).__name__
+ if type(f) is int:
+ s2 = str(f)
+ else:
+ s2 = VPPFormat.unformat_t(t, f)
+ s += '{} '.format(s2)
+ return s[:-1] + ']'
+
+ @staticmethod
+ def unformat(args):
+ s = ''
+ return VPPFormat.unformat_t(type(args).__name__, args)
+ '''
+ for i, f in enumerate(args):
+ print('F', f)
+ t = type(f).__name__
+ if type(f) is int:
+ s2 = str(f)
+ else:
+ s2 = VPPFormat.unformat_t(t, f)
+ s += '{} {} '.format(args._fields[i], s2)
+ return s[:-1]
+ '''
+
+ @staticmethod
+ def unformat_t(typename, args):
+ try:
+ return getattr(VPPFormat, 'unformat_' + typename)(args)
+ except AttributeError:
+ # Type without explicit override
+ return VPPFormat.unformat(args)
+
+ # Default handling
+ return args
import atexit
from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType, BaseTypes
from . vpp_serializer import VPPMessage, vpp_get_type
+from . vpp_format import VPPFormat
if sys.version[0] == '2':
import Queue as queue
import collections
from enum import IntEnum
import logging
+from .vpp_format import VPPFormat
#
# Set log-level in application by doing e.g.:
.format(type, base_types[type]))
def pack(self, data, kwargs=None):
+ if not data: # Default to zero if not specified
+ data = 0
return self.packer.pack(data)
def unpack(self, data, offset, result=None):
def pack(self, list, kwargs):
"""Packs a fixed length bytestring. Left-pads with zeros
if input data is too short."""
+ if not list:
+ return b'\x00' * self.size
if len(list) > self.num:
raise ValueError('Fixed list length error for "{}", got: {}'
' expected: {}'
self.length_field = len_field_name
def pack(self, list, kwargs=None):
+ if not list:
+ return b""
if len(list) != kwargs[self.length_field]:
raise ValueError('Variable length error, got: {} expected: {}'
.format(len(list), kwargs[self.length_field]))
return True
def pack(self, data, kwargs=None):
- return types['u32'].pack(data, kwargs)
+ return types['u32'].pack(data)
def unpack(self, data, offset=0, result=None):
x, size = types['u32'].unpack(data, offset)
self.tuple = collections.namedtuple(name, fields, rename=True)
logger.debug('Adding union {}'.format(name))
+ # Union of variable length?
def pack(self, data, kwargs=None):
+ if not data:
+ return b'\x00' * self.size
+
for k, v in data.items():
logger.debug("Key: {} Value: {}".format(k, v))
b = self.packers[k].pack(v, kwargs)
kwargs = data
b = bytes()
for i, a in enumerate(self.fields):
- if a not in data:
- b += b'\x00' * self.packers[i].size
- continue
+
+ # Try one of the format functions
+ if data and type(data) is not dict and a not in data:
+ raise ValueError("Invalid argument: {} expected {}.{}".
+ format(data, self.name, a))
+
+ # Defaulting to zero.
+ if not data or a not in data: # Default to 0
+ arg = None
+ kwarg = None # No default for VLA
+ else:
+ arg = data[a]
+ kwarg = kwargs[a] if a in kwargs else None
if isinstance(self.packers[i], VPPType):
- b += self.packers[i].pack(data[a], kwargs[a])
+ try:
+ b += self.packers[i].pack(arg, kwarg)
+ except ValueError:
+ # Invalid argument, can we convert it?
+ arg = VPPFormat.format(self.packers[i].name, data[a])
+ data[a] = arg
+ kwarg = arg
+ b += self.packers[i].pack(arg, kwarg)
else:
- b += self.packers[i].pack(data[a], kwargs)
+ b += self.packers[i].pack(arg, kwargs)
+
return b
def unpack(self, data, offset=0, result=None):
if type(strings) is not list:
strings = [strings]
for s in strings:
- vec = api.stat_segment_string_vector(vec, ffi.new("char []", s.encode()))
+ vec = api.stat_segment_string_vector(vec, ffi.new("char []",
+ s.encode()))
return vec
for i in range(rv_len):
n = ffi.string(rv[i].name).decode()
e = stat_entry_to_python(self.api, rv[i])
- if e != None:
+ if e is not None:
stats[n] = e
return stats
try:
dir = self.ls(name)
return self.dump(dir).values()[0]
- except:
+ except Exception as e:
if retries > 10:
return None
retries += 1
error_names = self.ls(['/err/'])
error_counters = self.dump(error_names)
break
- except:
+ except Exception as e:
if retries > 10:
return None
retries += 1
from vpp_ip import *
from vpp_mac import *
from vpp_papi_provider import L2_PORT_TYPE
+from vpp_papi.vpp_format import VPPFormat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, ARP
sw_if_index=None, epg=None):
self._test = test
self.table_id = table_id
- self.prefix = VppIpPrefix(address, address_len)
+ self.prefix = "{}/{}".format(address, address_len)
self.is_internal = is_internal
self.sw_if_index = sw_if_index
self.epg = epg
1,
self.table_id,
self.is_internal,
- self.prefix.encode(),
+ self.prefix,
sw_if_index=self.sw_if_index if self.sw_if_index else 0xffffffff,
epg_id=self.epg if self.epg else 0xffff)
self._test.registry.register(self, self._test.logger)
0,
self.table_id,
self.is_internal,
- self.prefix.encode())
+ self.prefix)
def __str__(self):
return self.object_id()
ss = self._test.vapi.gbp_subnet_dump()
for s in ss:
if s.subnet.table_id == self.table_id and \
- s.subnet.prefix == self.prefix:
+ VPPFormat.unformat(s.subnet.prefix) == self.prefix:
return True
return False
#!/usr/bin/env python
from framework import VppTestCase, VppTestRunner
-from vpp_ip import VppIpPrefix
-
from vpp_ip_route import VppIpTable
from scapy.packet import Raw
#
for i in range(1, 4):
self.vapi.svs_route_add_del(
- table_id,
- VppIpPrefix("%d.0.0.0" % i, 8).encode(),
- i)
+ table_id, "%d.0.0.0/8" % i, i)
#
# Enable SVS on pg0/pg1 using table 1001/1002
for table_id in table_ids:
for i in range(1, 4):
self.vapi.svs_route_add_del(
- table_id,
- VppIpPrefix("%d.0.0.0" % i, 8).encode(),
+ table_id, "%d.0.0.0/8" % i,
0, is_add=0)
self.vapi.svs_table_add_del(
VppEnum.vl_api_address_family_t.ADDRESS_IP4,
#
for i in range(1, 4):
self.vapi.svs_route_add_del(
- table_id,
- VppIpPrefix("2001:%d::" % i, 32).encode(),
+ table_id, "2001:%d::/32" % i,
i)
#
for table_id in table_ids:
for i in range(1, 4):
self.vapi.svs_route_add_del(
- table_id,
- VppIpPrefix("2001:%d::" % i, 32).encode(),
+ table_id, "2001:%d::/32" % i,
0, is_add=0)
self.vapi.svs_table_add_del(
VppEnum.vl_api_address_family_t.ADDRESS_IP6,
Code Review / vpp.git / commitdiff