VPP Python language binding - plugin support
- Moved Python generator tool to tools directory - Added build-vpp-api Makefile target - Generator now only creates a Python representation of the .api the rest of the framework is in the vpp_papi script - Each plugin has its own namespace. - Plugin Python files are installed in vpp_papi_plugins for easy use inside the build tree. Change-Id: I272c83bb7e5d5e416bdbd8a790a3cc35c5a04e38 Signed-off-by: Ole Troan <ot@cisco.com>
This commit is contained in:
Ole Troan
7
vpp-api/python/tests/test_base.py
Normal file
7
vpp-api/python/tests/test_base.py
Normal file
@ -0,0 +1,7 @@
|
||||
# Manipulate sys.path to allow tests be run inside the build environment.
|
||||
import os, sys, glob
|
||||
scriptdir = os.path.dirname(os.path.realpath(__file__))
|
||||
sys.path.append(os.path.dirname(glob.glob(scriptdir+'/../../../build-root/install*/vpp-api/lib64/vpp_api.so')[0]))
|
||||
sys.path.append(os.path.dirname(glob.glob(scriptdir+'/../../../build-root/install*/vlib-api/vlibmemory/memclnt.py')[0]))
|
||||
sys.path.append(os.path.dirname(glob.glob(scriptdir+'/../../../build-root/install*/vpp/vpp-api/vpe.py')[0]))
|
||||
sys.path.append(glob.glob(scriptdir+'/../../../build-root/install*/plugins/vpp_papi_plugins')[0])
|
||||
17
vpp-api/python/tests/test_modules.py
Executable file
17
vpp-api/python/tests/test_modules.py
Executable file
@ -0,0 +1,17 @@
|
||||
#!/usr/bin/env python
|
||||
from __future__ import print_function
|
||||
import unittest
|
||||
import test_base
|
||||
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)
|
||||
|
||||
vpp_papi.disconnect()
|
||||
@ -1,102 +1,104 @@
|
||||
#!/usr/bin/env python3
|
||||
#!/usr/bin/env python
|
||||
|
||||
import vpp_papi
|
||||
from __future__ import print_function
|
||||
import unittest, sys, time, threading, struct, logging
|
||||
import test_base
|
||||
import vpp_papi
|
||||
from ipaddress import *
|
||||
|
||||
papi_event = threading.Event()
|
||||
print(vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS)
|
||||
def papi_event_handler(result):
|
||||
if result.vl_msg_id == vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS:
|
||||
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
|
||||
if result.vl_msg_id == vpp_papi.VL_API_VNET_INTERFACE_COUNTERS:
|
||||
format = '>' + str(int(len(result.data) / 8)) + 'Q'
|
||||
counters = struct.unpack(format, result.data)
|
||||
print('Counters:', counters)
|
||||
return
|
||||
if result.vl_msg_id == vpp_papi.VL_API_VNET_IP6_FIB_COUNTERS:
|
||||
print('IP6 FIB Counters:', result.count, len(result.c), len(result))
|
||||
i = 0
|
||||
# FIB counters allocate a large (1000 bytes) block so message length does not match reality
|
||||
for c in struct.iter_unpack('>16sBQQ', result.c):
|
||||
# In Python 3.5 we can use a tuple for prefix, length
|
||||
print(str(IPv6Address(c[0])) + '/' + str(c[1]), str(c[2]), str(c[3]))
|
||||
i += 1
|
||||
if i >= result.count:
|
||||
break
|
||||
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(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_show_version(self):
|
||||
t = vpp_papi.show_version()
|
||||
print('T', t);
|
||||
program = t.program.decode().rstrip('\x00')
|
||||
self.assertEqual('vpe', program)
|
||||
|
||||
#
|
||||
# Add a few MAP domains, then dump them later
|
||||
# Details / Dump
|
||||
#
|
||||
def test_map(self):
|
||||
t = vpp_papi.map_summary_stats()
|
||||
print(t)
|
||||
ip6 = IPv6Address(u'2001:db8::1').packed
|
||||
ip4 = IPv4Address(u'10.0.0.0').packed
|
||||
ip6_src = IPv6Address(u'2001:db9::1').packed
|
||||
t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
|
||||
print(t)
|
||||
self.assertEqual(t.retval, 0)
|
||||
def test_details_dump(self):
|
||||
t = vpp_papi.sw_interface_dump(0, b'')
|
||||
print('Dump/details T', t)
|
||||
|
||||
ip4 = IPv4Address(u'10.0.1.0').packed
|
||||
t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
|
||||
print(t)
|
||||
self.assertEqual(t.retval, 0)
|
||||
|
||||
t = vpp_papi.map_summary_stats()
|
||||
print(t)
|
||||
self.assertEqual(t.total_bindings, 2)
|
||||
|
||||
t = vpp_papi.map_domain_dump()
|
||||
print (t)
|
||||
self.assertEqual(len(t), 2)
|
||||
|
||||
def test_sw_interface_dump(self):
|
||||
#
|
||||
# Dump interfaces
|
||||
#
|
||||
t = vpp_papi.sw_interface_dump(0, b'ignored')
|
||||
for interface in t:
|
||||
if interface.vl_msg_id == vpp_papi.VL_API_SW_INTERFACE_DETAILS:
|
||||
print(interface.interface_name.decode())
|
||||
|
||||
def test_want_interface_events(self):
|
||||
pid = 123
|
||||
vpp_papi.register_event_callback(papi_event_handler)
|
||||
papi_event.clear()
|
||||
t = vpp_papi.want_interface_events(True, pid)
|
||||
print (t)
|
||||
print('Setting interface up')
|
||||
t = vpp_papi.sw_interface_set_flags(0, 1, 1, 0)
|
||||
print (t)
|
||||
self.assertEqual(papi_event.wait(5), True)
|
||||
t = vpp_papi.sw_interface_set_flags(0, 0, 0, 0)
|
||||
print (t)
|
||||
self.assertEqual(papi_event.wait(5), True)
|
||||
|
||||
@unittest.skip("not quite ready yet")
|
||||
#
|
||||
# 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('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)
|
||||
@ -104,33 +106,17 @@ class TestPAPI(unittest.TestCase):
|
||||
#
|
||||
# Wait for some stats
|
||||
#
|
||||
self.assertEqual(papi_event.wait(30), True)
|
||||
self.assertEqual(papi_event.wait(15), True)
|
||||
t = vpp_papi.want_stats(False, pid)
|
||||
print (t)
|
||||
|
||||
def test_tap(self):
|
||||
pid = 123
|
||||
vpp_papi.register_event_callback(papi_event_handler)
|
||||
papi_event.clear()
|
||||
t = vpp_papi.want_stats(True, pid)
|
||||
|
||||
print (t)
|
||||
|
||||
t = vpp_papi.tap_connect(1, b'tap', b'foo', 1, 0)
|
||||
print (t)
|
||||
self.assertEqual(t.retval, 0)
|
||||
swifindex = t.sw_if_index
|
||||
|
||||
t = vpp_papi.sw_interface_set_flags(swifindex, 1, 1, 0)
|
||||
print (t)
|
||||
self.assertEqual(t.retval, 0)
|
||||
|
||||
ip6 = IPv6Address(u'2001:db8::1').packed
|
||||
t = vpp_papi.sw_interface_add_del_address(swifindex, 1, 1, 0, 16, ip6)
|
||||
print (t)
|
||||
time.sleep(40)
|
||||
|
||||
#
|
||||
# Plugins?
|
||||
#
|
||||
|
||||
if __name__ == '__main__':
|
||||
#logging.basicConfig(level=logging.DEBUG)
|
||||
unittest.main()
|
||||
def test_papi():
|
||||
print('test')
|
||||
|
||||
Reference in New Issue
Block a user