vpp-api/python/tests/test_papi.py

   1 #!/usr/bin/env python3
   2
   3 import vpp_papi
   4 import unittest, sys, time, threading, struct, logging
   5 from ipaddress import *
   6
   7 papi_event = threading.Event()
   8 def papi_event_handler(result):
   9     if result.vl_msg_id == vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS:
  10         papi_event.set()
  11         return
  12     if result.vl_msg_id == vpp_papi.VL_API_VNET_INTERFACE_COUNTERS:
  13         format = '>' + str(int(len(result.data) / 8)) + 'Q'
  14         counters = struct.unpack(format, result.data)
  15         print('Counters:', counters)
  16         return
  17     if result.vl_msg_id == vpp_papi.VL_API_VNET_IP6_FIB_COUNTERS:
  18         print('IP6 FIB Counters:', result.count, len(result.c), len(result))
  19         i = 0
  20         # FIB counters allocate a large (1000 bytes) block so message length does not match reality
  21         for c in struct.iter_unpack('>16sBQQ', result.c):
  22             # In Python 3.5 we can use a tuple for prefix, length
  23             print(str(IPv6Address(c[0])) + '/' + str(c[1]), str(c[2]), str(c[3]))
  24             i += 1
  25             if i >= result.count:
  26                 break
  27         return
  28
  29     print('Unknown message id:', result.vl_msg_id)
  30
  31 class TestPAPI(unittest.TestCase):
  32
  33     def setUp(self):
  34         r = vpp_papi.connect("test_papi")
  35         self.assertEqual(r, 0)
  36
  37     def tearDown(self):
  38         r = vpp_papi.disconnect()
  39         self.assertEqual(r, 0)
  40
  41     def test_show_version(self):
  42         t = vpp_papi.show_version()
  43         program = t.program.decode().rstrip('\x00')
  44         self.assertEqual('vpe', program)
  45
  46     #
  47     # Add a few MAP domains, then dump them later
  48     #
  49     def test_map(self):
  50         t = vpp_papi.map_summary_stats()
  51         print(t)
  52         ip6 = IPv6Address(u'2001:db8::1').packed
  53         ip4 = IPv4Address(u'10.0.0.0').packed
  54         ip6_src = IPv6Address(u'2001:db9::1').packed
  55         t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
  56         print(t)
  57         self.assertEqual(t.retval, 0)
  58
  59         ip4 = IPv4Address(u'10.0.1.0').packed
  60         t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
  61         print(t)
  62         self.assertEqual(t.retval, 0)
  63
  64         t = vpp_papi.map_summary_stats()
  65         print(t)
  66         self.assertEqual(t.total_bindings, 2)
  67
  68         t = vpp_papi.map_domain_dump()
  69         print (t)
  70         self.assertEqual(len(t), 2)
  71
  72     def test_sw_interface_dump(self):
  73         #
  74         # Dump interfaces
  75         #
  76         t = vpp_papi.sw_interface_dump(0, b'ignored')
  77         for interface in t:
  78             if interface.vl_msg_id == vpp_papi.VL_API_SW_INTERFACE_DETAILS:
  79                 print(interface.interface_name.decode())
  80
  81     def test_want_interface_events(self):
  82         pid = 123
  83         vpp_papi.register_event_callback(papi_event_handler)
  84         papi_event.clear()
  85         t = vpp_papi.want_interface_events(True, pid)
  86         print (t)
  87         print('Setting interface up')
  88         t = vpp_papi.sw_interface_set_flags(0, 1, 1, 0)
  89         print (t)
  90         self.assertEqual(papi_event.wait(5), True)
  91         t = vpp_papi.sw_interface_set_flags(0, 0, 0, 0)
  92         print (t)
  93         self.assertEqual(papi_event.wait(5), True)
  94
  95     @unittest.skip("not quite ready yet")
  96     def test_want_stats(self):
  97         pid = 123
  98         vpp_papi.register_event_callback(papi_event_handler)
  99         papi_event.clear()
 100         t = vpp_papi.want_stats(True, pid)
 101
 102         print (t)
 103
 104         #
 105         # Wait for some stats
 106         #
 107         self.assertEqual(papi_event.wait(30), True)
 108         t = vpp_papi.want_stats(False, pid)
 109         print (t)
 110
 111     def test_tap(self):
 112         pid = 123
 113         vpp_papi.register_event_callback(papi_event_handler)
 114         papi_event.clear()
 115         t = vpp_papi.want_stats(True, pid)
 116
 117         print (t)
 118
 119         t = vpp_papi.tap_connect(1, b'tap', b'foo', 1, 0)
 120         print (t)
 121         self.assertEqual(t.retval, 0)
 122         swifindex = t.sw_if_index
 123
 124         t = vpp_papi.sw_interface_set_flags(swifindex, 1, 1, 0)
 125         print (t)
 126         self.assertEqual(t.retval, 0)
 127
 128         ip6 = IPv6Address(u'2001:db8::1').packed
 129         t = vpp_papi.sw_interface_add_del_address(swifindex, 1, 1, 0, 16, ip6)
 130         print (t)
 131         time.sleep(40)
 132
 133
 134 if __name__ == '__main__':
 135     #logging.basicConfig(level=logging.DEBUG)
 136     unittest.main()