ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Python API: Add enum and union support.

Browse Source 
As well as a rewrite of the encoders/decoders to make it more readable and extensible.
(Re-commit after fix to verify build.)

Change-Id: Ic244d3cebe070bb2570491f8a24f4a1e203f889a
Signed-off-by: Ole Troan <ot@cisco.com>
This commit is contained in:
Ole Troan
2018-06-12 21:06:44 +02:00
committed by Florin Coras
parent d600ffe13a
commit a7564e8004
15 changed files with 682 additions and 1523 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -93,8 +93,6 @@ GTAGS
/src/vpp-api/python/build
/src/vpp-api/python/dist
/src/vpp-api/python/vpp_papi.egg-info
/src/vpp-api/python/vpp_papi/memclnt.py
/src/vpp-api/python/vpp_papi/vpe.py
# Build files in the test directory
/test/*.ok

2
extras/rpm/vpp-suse.spec
View File

@ -330,7 +330,7 @@ export NO_BRP_CHECK_RPATH=true
%files api-python
%dir %{python_sitelib}/vpp_papi*
%{python_sitelib}/vpp_papi*
%{python_sitelib}/vpp_*
%files devel
%dir %{python_sitelib}/jvppgen

2
extras/rpm/vpp.spec
View File

@ -397,7 +397,7 @@ fi
%files api-python
%defattr(644,root,root)
%{python2_sitelib}/vpp_papi*
%{python2_sitelib}/vpp_*
%files selinux-policy
%defattr(-,root,root,0755)

32
src/vpp-api/python/setup.py
View File

@ -13,21 +13,21 @@
# limitations under the License.
try:
from setuptools import setup
from setuptools import setup, find_packages
except ImportError:
from distutils.core import setup
from distutils.core import setup, find_packages
setup (name = 'vpp_papi',
version = '1.4',
description = 'VPP Python binding',
author = 'Ole Troan',
author_email = 'ot@cisco.com',
url = 'https://wiki.fd.io/view/VPP/Python_API',
python_requires='>=2.7, >=3.3',
license = 'Apache-2.0',
test_suite = 'tests',
install_requires=['cffi >= 1.6'],
py_modules=['vpp_papi'],
long_description = '''VPP Python language binding.''',
zip_safe = True,
)
setup(name='vpp_papi',
version='1.5',
description='VPP Python binding',
author='Ole Troan',
author_email='ot@cisco.com',
url='https://wiki.fd.io/view/VPP/Python_API',
license='Apache-2.0',
test_suite='tests',
# Add when we don't need to support 2.7.5
# 'enum34;python_version<"3.4"'],
install_requires=['cffi >= 1.6', 'enum34'],
packages=find_packages(),
long_description='''VPP Python language binding.''',
zip_safe=True)

52
src/vpp-api/python/tests/test_cli.py
View File

@ -1,52 +0,0 @@
#!/usr/bin/env python
from __future__ import print_function
import unittest, sys, time, threading, struct
import test_base
import vpp_papi
from ipaddress import *
import glob, subprocess
class TestPAPI(unittest.TestCase):
@classmethod
def setUpClass(cls):
#
# Start main VPP process
cls.vpp_bin = glob.glob(test_base.scriptdir+'/../../../build-root/install-vpp*-native/vpp/bin/vpp')[0]
print("VPP BIN:", cls.vpp_bin)
cls.vpp = subprocess.Popen([cls.vpp_bin, "unix", "nodaemon"], stderr=subprocess.PIPE)
print('Started VPP')
# For some reason unless we let VPP start up the API cannot connect.
time.sleep(0.3)
@classmethod
def tearDownClass(cls):
cls.vpp.terminate()
def setUp(self):
print("Connecting API")
r = vpp_papi.connect("test_papi")
self.assertEqual(r, 0)
def tearDown(self):
r = vpp_papi.disconnect()
self.assertEqual(r, 0)
#
# The tests themselves
#
#
# Basic request / reply
#
def test_cli_request(self):
print(vpp_papi.cli_exec('show version verbose'))
#t = vpp_papi.cli_inband_request(len(cmd), cmd)
#print('T:',t)
#reply = t.reply[0].decode().rstrip('\x00')
#print(reply)
#program = t.program.decode().rstrip('\x00')
#self.assertEqual('vpe', program)
if __name__ == '__main__':
unittest.main()

18
src/vpp-api/python/tests/test_modules.py
View File

@ -1,18 +0,0 @@
from __future__ import print_function
import unittest
import vpp_papi
import pot, snat
print('Plugins:')
vpp_papi.plugin_show()
r = vpp_papi.connect('ole')
r = vpp_papi.show_version()
print('R:', r)
r = snat.snat_interface_add_del_feature(1, 1, 1)
print('R:', r)
list_name = 'foobar'
r = pot.pot_profile_add(0, 1, 123, 123, 0, 12, 0, 23, len(list_name), list_name)
print('R:', r)
vpp_papi.disconnect()

119
src/vpp-api/python/tests/test_papi.py
View File

@ -1,119 +0,0 @@
from __future__ import print_function
import unittest, sys, time, threading, struct, logging, os
import vpp_papi
from ipaddress import *
scriptdir = os.path.dirname(os.path.realpath(__file__))
papi_event = threading.Event()
print(vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS)
def papi_event_handler(result):
if result.vl_msg_id == vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS:
return
if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_INTERFACE_COUNTERS:
print('Interface counters', result)
return
if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_IP6_FIB_COUNTERS:
print('IPv6 FIB counters', result)
papi_event.set()
return
print('Unknown message id:', result.vl_msg_id)
import glob, subprocess
class TestPAPI(unittest.TestCase):
@classmethod
def setUpClass(cls):
#
# Start main VPP process
cls.vpp_bin = glob.glob(scriptdir+'/../../../build-root/install-vpp*-native/vpp/bin/vpp')[0]
print("VPP BIN:", cls.vpp_bin)
cls.vpp = subprocess.Popen([cls.vpp_bin, "unix", "nodaemon"], stderr=subprocess.PIPE)
print('Started VPP')
# For some reason unless we let VPP start up the API cannot connect.
time.sleep(0.3)
@classmethod
def tearDownClass(cls):
cls.vpp.terminate()
def setUp(self):
print("Connecting API")
r = vpp_papi.connect("test_papi")
self.assertEqual(r, 0)
def tearDown(self):
r = vpp_papi.disconnect()
self.assertEqual(r, 0)
#
# The tests themselves
#
#
# Basic request / reply
#
def test_show_version(self):
t = vpp_papi.show_version()
print('T', t);
program = t.program.decode().rstrip('\x00')
self.assertEqual('vpe', program)
#
# Details / Dump
#
def test_details_dump(self):
t = vpp_papi.sw_interface_dump(0, b'')
print('Dump/details T', t)
#
# Arrays
#
def test_arrays(self):
t = vpp_papi.vnet_get_summary_stats()
print('Summary stats', t)
print('Packets:', t.total_pkts[0])
print('Packets:', t.total_pkts[1])
#
# Variable sized arrays and counters
#
#@unittest.skip("stats")
def test_want_stats(self):
pid = 123
vpp_papi.register_event_callback(papi_event_handler)
papi_event.clear()
# Need to configure IPv6 to get som IPv6 FIB stats
t = vpp_papi.create_loopback('')
print(t)
self.assertEqual(t.retval, 0)
ifindex = t.sw_if_index
addr = str(IPv6Address(u'1::1').packed)
t = vpp_papi.sw_interface_add_del_address(ifindex, 1, 1, 0, 16, addr)
print(t)
self.assertEqual(t.retval, 0)
# Check if interface is up
# XXX: Add new API to query interface state based on ifindex, instead of dump all.
t = vpp_papi.sw_interface_set_flags(ifindex, 1, 1, 0)
self.assertEqual(t.retval, 0)
t = vpp_papi.want_stats(True, pid)
print (t)
#
# Wait for some stats
#
self.assertEqual(papi_event.wait(15), True)
t = vpp_papi.want_stats(False, pid)
print (t)
#
# Plugins?
#
if __name__ == '__main__':
#logging.basicConfig(level=logging.DEBUG)
unittest.main()
def test_papi():
print('test')

35
src/vpp-api/python/tests/test_version.py
View File

@ -1,35 +0,0 @@
from __future__ import print_function
import unittest, sys, time, threading, struct
import vpp_papi
from ipaddress import *
import glob, subprocess
class TestPAPI(unittest.TestCase):
def setUp(self):
print("Connecting API")
r = vpp_papi.connect("test_papi")
self.assertEqual(r, 0)
def tearDown(self):
r = vpp_papi.disconnect()
self.assertEqual(r, 0)
#
# The tests themselves
#
#
# Basic request / reply
#
def test_show_version(self):
print(vpp_papi.show_version())
#
# Details / Dump
#
def test_details_dump(self):
t = vpp_papi.sw_interface_dump(0, b'')
print('Dump/details T', t)
if __name__ == '__main__':
unittest.main()

487
src/vpp-api/python/tests/test_vpp_papi2.py
View File

File diff suppressed because it is too large Load Diff

226
src/vpp-api/python/tests/test_vpp_serializer.py Executable file
View File

@ -0,0 +1,226 @@
#!/usr/bin/env python
import unittest
from vpp_papi.vpp_serializer import VPPType, VPPEnumType
from vpp_papi.vpp_serializer import VPPUnionType, VPPMessage
from socket import inet_pton, AF_INET, AF_INET6
import logging
import sys
class TestAddType(unittest.TestCase):
def test_union(self):
un = VPPUnionType('test_union',
[['u8', 'is_bool'],
['u32', 'is_int']])
b = un.pack({'is_int': 0x1234})
self.assertEqual(len(b), 4)
nt = un.unpack(b)
self.assertEqual(nt.is_bool, 52)
self.assertEqual(nt.is_int, 0x1234)
def test_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('address', [['vl_api_address_family_t', 'af'],
['vl_api_address_union_t', 'un']])
b = ip4.pack({'address': inet_pton(AF_INET, '1.1.1.1')})
self.assertEqual(len(b), 4)
nt = ip4.unpack(b)
self.assertEqual(nt.address, inet_pton(AF_INET, '1.1.1.1'))
b = ip6.pack({'address': 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')}}})
self.assertEqual(len(b), 20)
nt = 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'))
def test_arrays(self):
# Test cases
# 1. Fixed list
# 2. Fixed list of variable length sub type
# 3. Variable length type
#
ip4 = VPPType('ip4_address', [['u8', 'address', 4]])
listip4 = VPPType('list_ip4_t', [['ip4_address', 'addresses', 4]])
valistip4 = VPPType('list_ip4_t',
[['u8', 'count'],
['ip4_address', 'addresses', 0, 'count']])
valistip4_legacy = VPPType('list_ip4_t',
[['u8', 'foo'],
['ip4_address', 'addresses', 0]])
addresses = []
for i in range(4):
addresses.append({'address': inet_pton(AF_INET, '2.2.2.2')})
b = listip4.pack({'addresses': addresses})
self.assertEqual(len(b), 16)
nt = listip4.unpack(b)
self.assertEqual(nt.addresses[0].address,
inet_pton(AF_INET, '2.2.2.2'))
b = valistip4.pack({'count': len(addresses), 'addresses': addresses})
self.assertEqual(len(b), 17)
nt = valistip4.unpack(b)
self.assertEqual(nt.count, 4)
self.assertEqual(nt.addresses[0].address,
inet_pton(AF_INET, '2.2.2.2'))
b = valistip4_legacy.pack({'foo': 1, 'addresses': addresses})
self.assertEqual(len(b), 17)
nt = valistip4_legacy.unpack(b)
self.assertEqual(len(nt.addresses), 4)
self.assertEqual(nt.addresses[0].address,
inet_pton(AF_INET, '2.2.2.2'))
def test_message(self):
foo = VPPMessage('foo', [['u16', '_vl_msg_id'],
['u8', 'client_index'],
['u8', 'something'],
{"crc": "0x559b9f3c"}])
b = foo.pack({'_vl_msg_id': 1, 'client_index': 5,
'something': 200})
self.assertEqual(len(b), 4)
nt = foo.unpack(b)
self.assertEqual(nt.something, 200)
def test_abf(self):
fib_mpls_label = VPPType('vl_api_fib_mpls_label_t',
[['u8', 'is_uniform'],
['u32', 'label'],
['u8', 'ttl'],
['u8', 'exp']])
label_stack = {'is_uniform': 0,
'label': 0,
'ttl': 0,
'exp': 0}
b = fib_mpls_label.pack(label_stack)
self.assertEqual(len(b), 7)
fib_path = VPPType('vl_api_fib_path_t',
[['u32', 'sw_if_index'],
['u32', 'table_id'],
['u8', 'weight'],
['u8', 'preference'],
['u8', 'is_local'],
['u8', 'is_drop'],
['u8', 'is_udp_encap'],
['u8', 'is_unreach'],
['u8', 'is_prohibit'],
['u8', 'is_resolve_host'],
['u8', 'is_resolve_attached'],
['u8', 'is_dvr'],
['u8', 'is_source_lookup'],
['u8', 'afi'],
['u8', 'next_hop', 16],
['u32', 'next_hop_id'],
['u32', 'rpf_id'],
['u32', 'via_label'],
['u8', 'n_labels'],
['vl_api_fib_mpls_label_t', 'label_stack', 16]])
label_stack_list = []
for i in range(16):
label_stack_list.append(label_stack)
paths = {'is_udp_encap': 0,
'next_hop': b'\x10\x02\x02\xac',
'table_id': 0,
'afi': 0,
'weight': 1,
'next_hop_id': 4294967295,
'label_stack': label_stack_list,
'n_labels': 0,
'sw_if_index': 4294967295,
'preference': 0}
b = fib_path.pack(paths)
self.assertEqual(len(b), (7*16) + 49)
abf_policy = VPPType('vl_api_abf_policy_t',
[['u32', 'policy_id'],
['u32', 'acl_index'],
['u8', 'n_paths'],
['vl_api_fib_path_t', 'paths', 0, 'n_paths']])
policy = {
'n_paths': 1,
'paths': [paths],
'acl_index': 0,
'policy_id': 10}
b = abf_policy.pack(policy)
self.assertEqual(len(b), (7*16) + 49 + 9)
abf_policy_add_del = VPPMessage('abf_policy_add_del',
[['u16', '_vl_msg_id'],
['u32', 'client_index'],
['u32', 'context'],
['u8', 'is_add'],
['vl_api_abf_policy_t', 'policy']])
b = abf_policy_add_del.pack({'is_add': 1,
'context': 66,
'_vl_msg_id': 1066,
'policy': policy})
nt = abf_policy_add_del.unpack(b)
self.assertEqual(nt.policy.paths[0].next_hop,
b'\x10\x02\x02\xac\x00\x00\x00\x00'
b'\x00\x00\x00\x00\x00\x00\x00\x00')
def test_bier(self):
bier_table_id = VPPType('vl_api_bier_table_id_t',
[['u8', 'bt_set'],
['u8', 'bt_sub_domain'],
['u8', 'bt_hdr_len_id']])
bier_imp_add = VPPMessage('bier_imp_add',
[['u32', 'client_index'],
['u32', 'context'],
['vl_api_bier_table_id_t', 'bi_tbl_id'],
['u16', 'bi_src'],
['u8', 'bi_n_bytes'],
['u8', 'bi_bytes', 0, 'bi_n_bytes']])
table_id = {'bt_set': 0,
'bt_sub_domain': 0,
'bt_hdr_len_id': 0}
bibytes = b'foobar'
b = bier_imp_add.pack({'bi_tbl_id': table_id,
'bi_n_bytes': len(bibytes),
'bi_bytes': bibytes})
self.assertEqual(len(b), 20)
if __name__ == '__main__':
unittest.main()

1
src/vpp-api/python/vpp_papi/__init__.py Normal file
View File

@ -0,0 +1 @@
from .vpp_papi import *

396
src/vpp-api/python/vpp_papi.py → src/vpp-api/python/vpp_papi/vpp_papi.py
View File

File diff suppressed because it is too large Load Diff

343
src/vpp-api/python/vpp_papi/vpp_serializer.py Normal file
View File

File diff suppressed because it is too large Load Diff

486
test/test_papi.py
View File

File diff suppressed because it is too large Load Diff

2
test/vpp_papi_provider.py
View File

@ -70,7 +70,7 @@ class VppPapiProvider(object):
for filename in fnmatch.filter(filenames, '*.api.json'):
jsonfiles.append(os.path.join(root, filename))
self.vpp = VPP(jsonfiles, logger=test_class.logger)
self.vpp = VPP(jsonfiles, logger=test_class.logger, read_timeout=5)
self._events = deque()
def __enter__(self):