plugins/acl-plugin/test/test_acl_plugin.py

   1 from __future__ import print_function
   2 import unittest, sys, time, threading, struct, logging, os
   3 import vpp_papi
   4 # import vpp_papi_plugins.acl
   5 from ipaddress import *
   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 TestAclPlugin(unittest.TestCase):
  23     @classmethod
  24     def setUpClass(cls):
  25         print("Setup")
  26     @classmethod
  27     def tearDownClass(cls):
  28         print("Teardown")
  29
  30     def setUp(self):
  31         print("Connecting API")
  32         r = vpp_papi.connect("test_papi")
  33         self.assertEqual(r, 0)
  34
  35     def tearDown(self):
  36         r = vpp_papi.disconnect()
  37         self.assertEqual(r, 0)
  38
  39     #
  40     # The tests themselves
  41     #
  42
  43     #
  44     # Basic request / reply
  45     #
  46     def test_show_version(self):
  47         t = vpp_papi.show_version()
  48         print('T', t);
  49         program = t.program.decode().rstrip('\x00')
  50         self.assertEqual('vpe', program)
  51
  52     def x_test_acl_add(self):
  53         print("Test ACL add")
  54         self.assertEqual(1, 1)
  55
  56     #
  57     # Details / Dump
  58     #
  59     def x_test_details_dump(self):
  60         t = vpp_papi.sw_interface_dump(0, b'')
  61         print('Dump/details T', t)
  62
  63     #
  64     # Arrays
  65     #
  66     def x_test_arrays(self):
  67         t = vpp_papi.vnet_get_summary_stats()
  68         print('Summary stats', t)
  69         print('Packets:', t.total_pkts[0])
  70         print('Packets:', t.total_pkts[1])
  71     #
  72     # Variable sized arrays and counters
  73     #
  74     #@unittest.skip("stats")
  75     def x_test_want_stats(self):
  76         pid = 123
  77         vpp_papi.register_event_callback(papi_event_handler)
  78         papi_event.clear()
  79
  80         # Need to configure IPv6 to get som IPv6 FIB stats
  81         t = vpp_papi.create_loopback('')
  82         print(t)
  83         self.assertEqual(t.retval, 0)
  84
  85         ifindex = t.sw_if_index
  86         addr = str(IPv6Address(u'1::1').packed)
  87         t = vpp_papi.sw_interface_add_del_address(ifindex, 1, 1, 0, 16, addr)
  88         print(t)
  89         self.assertEqual(t.retval, 0)
  90
  91         # Check if interface is up
  92         # XXX: Add new API to query interface state based on ifindex, instead of dump all.
  93         t = vpp_papi.sw_interface_set_flags(ifindex, 1, 1, 0)
  94         self.assertEqual(t.retval, 0)
  95
  96         t = vpp_papi.want_stats(True, pid)
  97
  98         print (t)
  99
 100         #
 101         # Wait for some stats
 102         #
 103         self.assertEqual(papi_event.wait(15), True)
 104         t = vpp_papi.want_stats(False, pid)
 105         print (t)
 106
 107
 108     #
 109     # Plugins?
 110     #
 111
 112 if __name__ == '__main__' or __name__ == '__builtin__':
 113     print("This is main")
 114     suite = unittest.TestLoader().loadTestsFromTestCase(TestAclPlugin)
 115     unittest.TextTestRunner(verbosity=2).run(suite)
 116     #logging.basicConfig(level=logging.DEBUG)
 117     # unittest.main()
 118