From 6855f6cdfee8c479f1e0ae440ce87a91ff41a708 Mon Sep 17 00:00:00 2001 From: Ole Troan Date: Sat, 9 Apr 2016 03:16:30 +0200 Subject: Python-API: Inital commit of Python bindings for the VPP API. See: https://wiki.fd.io/view/VPP/Python_API Change-Id: If135fc32208c7031787e1935b399d930e0e1ea1f Signed-off-by: Ole Troan --- vpp-api/python/tests/test_papi.py | 125 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100755 vpp-api/python/tests/test_papi.py (limited to 'vpp-api/python/tests/test_papi.py') diff --git a/vpp-api/python/tests/test_papi.py b/vpp-api/python/tests/test_papi.py new file mode 100755 index 00000000000..ec51f0936cb --- /dev/null +++ b/vpp-api/python/tests/test_papi.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 + +import vpp_papi +import unittest, sys, time, threading, struct, logging +from ipaddress import * + +papi_event = threading.Event() +def papi_event_handler(result): + if result.vlmsgid == vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS: + papi_event.set() + return + if result.vlmsgid == 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 + + print('Unknown message id:', result.vlmsgid) + +class TestPAPI(unittest.TestCase): + + def setUp(self): + r = vpp_papi.connect("test_papi") + self.assertEqual(r, 0) + + def tearDown(self): + r = vpp_papi.disconnect() + self.assertEqual(r, 0) + + def test_show_version(self): + t = vpp_papi.show_version() + program = t.program.decode().rstrip('\x00') + self.assertEqual('vpe', program) + + # + # Add a few MAP domains, then dump them later + # + 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) + + 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.totalbindings, 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.vlmsgid == vpp_papi.VL_API_SW_INTERFACE_DETAILS: + print(interface.interfacename.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") + def test_want_stats(self): + pid = 123 + vpp_papi.register_event_callback(papi_event_handler) + papi_event.clear() + t = vpp_papi.want_stats(True, pid) + + print (t) + + # + # Wait for some stats + # + self.assertEqual(papi_event.wait(30), 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.swifindex + + 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) + + +if __name__ == '__main__': + #logging.basicConfig(level=logging.DEBUG) + unittest.main() -- cgit 1.2.3-korg