vpp-api/python/tests/test_papi.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import print_function
   4 import unittest, sys, time, threading, struct, logging
   5 import test_base
   6 import vpp_papi
   7 from ipaddress import *
   8
   9 papi_event = threading.Event()
  10 print(vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS)
  11 def papi_event_handler(result):
  12     if result.vl_msg_id == vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS:
  13         return
  14     if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_INTERFACE_COUNTERS:
  15         print('Interface counters', result)
  16         return
  17     if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_IP6_FIB_COUNTERS:
  18         print('IPv6 FIB counters', result)
  19         papi_event.set()
  20         return
  21
  22     print('Unknown message id:', result.vl_msg_id)
  23
  24 import glob, subprocess
  25 class TestPAPI(unittest.TestCase):
  26     @classmethod
  27     def setUpClass(cls):
  28         #
  29         # Start main VPP process
  30         cls.vpp_bin = glob.glob(test_base.scriptdir+'/../../../build-root/install-vpp*-native/vpp/bin/vpp')[0]
  31         print("VPP BIN:", cls.vpp_bin)
  32         cls.vpp = subprocess.Popen([cls.vpp_bin, "unix", "nodaemon"], stderr=subprocess.PIPE)
  33         print('Started VPP')
  34         # For some reason unless we let VPP start up the API cannot connect.
  35         time.sleep(0.3)
  36     @classmethod
  37     def tearDownClass(cls):
  38         cls.vpp.terminate()
  39
  40     def setUp(self):
  41         print("Connecting API")
  42         r = vpp_papi.connect("test_papi")
  43         self.assertEqual(r, 0)
  44
  45     def tearDown(self):
  46         r = vpp_papi.disconnect()
  47         self.assertEqual(r, 0)
  48
  49     #
  50     # The tests themselves
  51     #
  52
  53     #
  54     # Basic request / reply
  55     #
  56     def test_show_version(self):
  57         t = vpp_papi.show_version()
  58         print('T', t);
  59         program = t.program.decode().rstrip('\x00')
  60         self.assertEqual('vpe', program)
  61
  62     #
  63     # Details / Dump
  64     #
  65     def test_details_dump(self):
  66         t = vpp_papi.sw_interface_dump(0, b'')
  67         print('Dump/details T', t)
  68
  69     #
  70     # Arrays
  71     #
  72     def test_arrays(self):
  73         t = vpp_papi.vnet_get_summary_stats()
  74         print('Summary stats', t)
  75         print('Packets:', t.total_pkts[0])
  76         print('Packets:', t.total_pkts[1])
  77     #
  78     # Variable sized arrays and counters
  79     #
  80     #@unittest.skip("stats")
  81     def test_want_stats(self):
  82         pid = 123
  83         vpp_papi.register_event_callback(papi_event_handler)
  84         papi_event.clear()
  85
  86         # Need to configure IPv6 to get som IPv6 FIB stats
  87         t = vpp_papi.create_loopback('')
  88         print(t)
  89         self.assertEqual(t.retval, 0)
  90
  91         ifindex = t.sw_if_index
  92         addr = str(IPv6Address('1::1').packed)
  93         t = vpp_papi.sw_interface_add_del_address(ifindex, 1, 1, 0, 16, addr)
  94         print(t)
  95         self.assertEqual(t.retval, 0)
  96
  97         # Check if interface is up
  98         # XXX: Add new API to query interface state based on ifindex, instead of dump all.
  99         t = vpp_papi.sw_interface_set_flags(ifindex, 1, 1, 0)
 100         self.assertEqual(t.retval, 0)
 101
 102         t = vpp_papi.want_stats(True, pid)
 103
 104         print (t)
 105
 106         #
 107         # Wait for some stats
 108         #
 109         self.assertEqual(papi_event.wait(15), True)
 110         t = vpp_papi.want_stats(False, pid)
 111         print (t)
 112
 113
 114     #
 115     # Plugins?
 116     #
 117
 118 if __name__ == '__main__':
 119     #logging.basicConfig(level=logging.DEBUG)
 120     unittest.main()
 121 def test_papi():
 122     print('test')