1 from __future__ import print_function
2 import unittest, sys, time, threading, struct, logging, os
4 from ipaddress import *
5 scriptdir = os.path.dirname(os.path.realpath(__file__))
6 papi_event = threading.Event()
7 print(vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS)
8 def papi_event_handler(result):
9 if result.vl_msg_id == vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS:
11 if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_INTERFACE_COUNTERS:
12 print('Interface counters', result)
14 if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_IP6_FIB_COUNTERS:
15 print('IPv6 FIB counters', result)
19 print('Unknown message id:', result.vl_msg_id)
21 import glob, subprocess
22 class TestPAPI(unittest.TestCase):
26 # Start main VPP process
27 cls.vpp_bin = glob.glob(scriptdir+'/../../../build-root/install-vpp*-native/vpp/bin/vpp')[0]
28 print("VPP BIN:", cls.vpp_bin)
29 cls.vpp = subprocess.Popen([cls.vpp_bin, "unix", "nodaemon"], stderr=subprocess.PIPE)
31 # For some reason unless we let VPP start up the API cannot connect.
34 def tearDownClass(cls):
38 print("Connecting API")
39 r = vpp_papi.connect("test_papi")
40 self.assertEqual(r, 0)
43 r = vpp_papi.disconnect()
44 self.assertEqual(r, 0)
47 # The tests themselves
51 # Basic request / reply
53 def test_show_version(self):
54 t = vpp_papi.show_version()
56 program = t.program.decode().rstrip('\x00')
57 self.assertEqual('vpe', program)
62 def test_details_dump(self):
63 t = vpp_papi.sw_interface_dump(0, b'')
64 print('Dump/details T', t)
69 def test_arrays(self):
70 t = vpp_papi.vnet_get_summary_stats()
71 print('Summary stats', t)
72 print('Packets:', t.total_pkts[0])
73 print('Packets:', t.total_pkts[1])
75 # Variable sized arrays and counters
77 #@unittest.skip("stats")
78 def test_want_stats(self):
80 vpp_papi.register_event_callback(papi_event_handler)
83 # Need to configure IPv6 to get som IPv6 FIB stats
84 t = vpp_papi.create_loopback('')
86 self.assertEqual(t.retval, 0)
88 ifindex = t.sw_if_index
89 addr = str(IPv6Address(u'1::1').packed)
90 t = vpp_papi.sw_interface_add_del_address(ifindex, 1, 1, 0, 16, addr)
92 self.assertEqual(t.retval, 0)
94 # Check if interface is up
95 # XXX: Add new API to query interface state based on ifindex, instead of dump all.
96 t = vpp_papi.sw_interface_set_flags(ifindex, 1, 1, 0)
97 self.assertEqual(t.retval, 0)
99 t = vpp_papi.want_stats(True, pid)
104 # Wait for some stats
106 self.assertEqual(papi_event.wait(15), True)
107 t = vpp_papi.want_stats(False, pid)
115 if __name__ == '__main__':
116 #logging.basicConfig(level=logging.DEBUG)