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.vlmsgid == vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS:
  10         papi_event.set()
  11         return
  12     if result.vlmsgid == 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
  18     print('Unknown message id:', result.vlmsgid)
  19
  20 class TestPAPI(unittest.TestCase):
  21
  22     def setUp(self):
  23         r = vpp_papi.connect("test_papi")
  24         self.assertEqual(r, 0)
  25
  26     def tearDown(self):
  27         r = vpp_papi.disconnect()
  28         self.assertEqual(r, 0)
  29
  30     def test_show_version(self):
  31         t = vpp_papi.show_version()
  32         program = t.program.decode().rstrip('\x00')
  33         self.assertEqual('vpe', program)
  34
  35     #
  36     # Add a few MAP domains, then dump them later
  37     #
  38     def test_map(self):
  39         t = vpp_papi.map_summary_stats()
  40         print(t)
  41         ip6 = IPv6Address(u'2001:db8::1').packed
  42         ip4 = IPv4Address(u'10.0.0.0').packed
  43         ip6_src = IPv6Address(u'2001:db9::1').packed
  44         t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
  45         print(t)
  46         self.assertEqual(t.retval, 0)
  47
  48         ip4 = IPv4Address(u'10.0.1.0').packed
  49         t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
  50         print(t)
  51         self.assertEqual(t.retval, 0)
  52
  53         t = vpp_papi.map_summary_stats()
  54         print(t)
  55         self.assertEqual(t.totalbindings, 2)
  56
  57         t = vpp_papi.map_domain_dump()
  58         print (t)
  59         self.assertEqual(len(t), 2)
  60
  61     def test_sw_interface_dump(self):
  62         #
  63         # Dump interfaces
  64         #
  65         t = vpp_papi.sw_interface_dump(0, b'ignored')
  66         for interface in t:
  67             if interface.vlmsgid == vpp_papi.VL_API_SW_INTERFACE_DETAILS:
  68                 print(interface.interfacename.decode())
  69
  70     def test_want_interface_events(self):
  71         pid = 123
  72         vpp_papi.register_event_callback(papi_event_handler)
  73         papi_event.clear()
  74         t = vpp_papi.want_interface_events(True, pid)
  75         print (t)
  76         print('Setting interface up')
  77         t = vpp_papi.sw_interface_set_flags(0, 1, 1, 0)
  78         print (t)
  79         self.assertEqual(papi_event.wait(5), True)
  80         t = vpp_papi.sw_interface_set_flags(0, 0, 0, 0)
  81         print (t)
  82         self.assertEqual(papi_event.wait(5), True)
  83
  84     @unittest.skip("not quite ready yet")
  85     def test_want_stats(self):
  86         pid = 123
  87         vpp_papi.register_event_callback(papi_event_handler)
  88         papi_event.clear()
  89         t = vpp_papi.want_stats(True, pid)
  90
  91         print (t)
  92
  93         #
  94         # Wait for some stats
  95         #
  96         self.assertEqual(papi_event.wait(30), True)
  97         t = vpp_papi.want_stats(False, pid)
  98         print (t)
  99
 100     def test_tap(self):
 101         pid = 123
 102         vpp_papi.register_event_callback(papi_event_handler)
 103         papi_event.clear()
 104         t = vpp_papi.want_stats(True, pid)
 105
 106         print (t)
 107
 108         t = vpp_papi.tap_connect(1, b'tap', b'foo', 1, 0)
 109         print (t)
 110         self.assertEqual(t.retval, 0)
 111         swifindex = t.swifindex
 112
 113         t = vpp_papi.sw_interface_set_flags(swifindex, 1, 1, 0)
 114         print (t)
 115         self.assertEqual(t.retval, 0)
 116
 117         ip6 = IPv6Address(u'2001:db8::1').packed
 118         t = vpp_papi.sw_interface_add_del_address(swifindex, 1, 1, 0, 16, ip6)
 119         print (t)
 120         time.sleep(40)
 121
 122
 123 if __name__ == '__main__':
 124     #logging.basicConfig(level=logging.DEBUG)
 125     unittest.main()