diff options
Diffstat (limited to 'vpp-api/python/tests/test_papi.py')
| -rwxr-xr-x | vpp-api/python/tests/test_papi.py | 158 |
1 files changed, 72 insertions, 86 deletions
diff --git a/vpp-api/python/tests/test_papi.py b/vpp-api/python/tests/test_papi.py index bede7171519..ab90eeaa45c 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: - papi_event.set() + if result.vl_msg_id == vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS: 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) + 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.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 + 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(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) + # + # 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) - t = vpp_papi.map_summary_stats() + 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.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()) + self.assertEqual(t.retval, 0) - 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) + # 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) - @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) @@ -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') |