vpp-api/python/tests/test_papi.py

   1 from __future__ import print_function
   2 import unittest, sys, time, threading, struct, logging, os
   3 import vpp_papi
   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:
  10         return
  11     if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_INTERFACE_COUNTERS:
  12         print('Interface counters', result)
  13         return
  14     if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_IP6_FIB_COUNTERS:
  15         print('IPv6 FIB counters', result)
  16         papi_event.set()
  17         return
  18
  19     print('Unknown message id:', result.vl_msg_id)
  20
  21 import glob, subprocess
  22 class TestPAPI(unittest.TestCase):
  23     @classmethod
  24     def setUpClass(cls):
  25         #
  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)
  30         print('Started VPP')
  31         # For some reason unless we let VPP start up the API cannot connect.
  32         time.sleep(0.3)
  33     @classmethod
  34     def tearDownClass(cls):
  35         cls.vpp.terminate()
  36
  37     def setUp(self):
  38         print("Connecting API")
  39         r = vpp_papi.connect("test_papi")
  40         self.assertEqual(r, 0)
  41
  42     def tearDown(self):
  43         r = vpp_papi.disconnect()
  44         self.assertEqual(r, 0)
  45
  46     #
  47     # The tests themselves
  48     #
  49
  50     #
  51     # Basic request / reply
  52     #
  53     def test_show_version(self):
  54         t = vpp_papi.show_version()
  55         print('T', t);
  56         program = t.program.decode().rstrip('\x00')
  57         self.assertEqual('vpe', program)
  58
  59     #
  60     # Details / Dump
  61     #
  62     def test_details_dump(self):
  63         t = vpp_papi.sw_interface_dump(0, b'')
  64         print('Dump/details T', t)
  65
  66     #
  67     # Arrays
  68     #
  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])
  74     #
  75     # Variable sized arrays and counters
  76     #
  77     #@unittest.skip("stats")
  78     def test_want_stats(self):
  79         pid = 123
  80         vpp_papi.register_event_callback(papi_event_handler)
  81         papi_event.clear()
  82
  83         # Need to configure IPv6 to get som IPv6 FIB stats
  84         t = vpp_papi.create_loopback('')
  85         print(t)
  86         self.assertEqual(t.retval, 0)
  87
  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)
  91         print(t)
  92         self.assertEqual(t.retval, 0)
  93
  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)
  98
  99         t = vpp_papi.want_stats(True, pid)
 100
 101         print (t)
 102
 103         #
 104         # Wait for some stats
 105         #
 106         self.assertEqual(papi_event.wait(15), True)
 107         t = vpp_papi.want_stats(False, pid)
 108         print (t)
 109
 110
 111     #
 112     # Plugins?
 113     #
 114
 115 if __name__ == '__main__':
 116     #logging.basicConfig(level=logging.DEBUG)
 117     unittest.main()
 118 def test_papi():
 119     print('test')