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>
diff --git a/vpp-api/python/tests/test_base.py b/vpp-api/python/tests/test_base.py
new file mode 100644
index 0000000..8ff5dd4
--- /dev/null
+++ b/vpp-api/python/tests/test_base.py
@@ -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])
diff --git a/vpp-api/python/tests/test_modules.py b/vpp-api/python/tests/test_modules.py
new file mode 100755
index 0000000..f3066b2
--- /dev/null
+++ b/vpp-api/python/tests/test_modules.py
@@ -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()
diff --git a/vpp-api/python/tests/test_papi.py b/vpp-api/python/tests/test_papi.py
index bede717..ab90eea 100755
--- a/vpp-api/python/tests/test_papi.py
+++ b/vpp-api/python/tests/test_papi.py
@@ -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 @@
#
# 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')