test/test_session.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from framework import VppTestCase, VppTestRunner
   6 from framework import tag_run_solo
   7 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
   8
   9
  10 class TestSession(VppTestCase):
  11     """ Session Test Case """
  12
  13     @classmethod
  14     def setUpClass(cls):
  15         super(TestSession, cls).setUpClass()
  16
  17     @classmethod
  18     def tearDownClass(cls):
  19         super(TestSession, cls).tearDownClass()
  20
  21     def setUp(self):
  22         super(TestSession, self).setUp()
  23
  24         self.vapi.session_enable_disable(is_enable=1)
  25         self.create_loopback_interfaces(2)
  26
  27         table_id = 0
  28
  29         for i in self.lo_interfaces:
  30             i.admin_up()
  31
  32             if table_id != 0:
  33                 tbl = VppIpTable(self, table_id)
  34                 tbl.add_vpp_config()
  35
  36             i.set_table_ip4(table_id)
  37             i.config_ip4()
  38             table_id += 1
  39
  40         # Configure namespaces
  41         self.vapi.app_namespace_add_del(namespace_id="0",
  42                                         sw_if_index=self.loop0.sw_if_index)
  43         self.vapi.app_namespace_add_del(namespace_id="1",
  44                                         sw_if_index=self.loop1.sw_if_index)
  45
  46     def tearDown(self):
  47         for i in self.lo_interfaces:
  48             i.unconfig_ip4()
  49             i.set_table_ip4(0)
  50             i.admin_down()
  51
  52         super(TestSession, self).tearDown()
  53         self.vapi.session_enable_disable(is_enable=1)
  54
  55     def test_segment_manager_alloc(self):
  56         """ Session Segment Manager Multiple Segment Allocation """
  57
  58         # Add inter-table routes
  59         ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
  60                             [VppRoutePath("0.0.0.0",
  61                                           0xffffffff,
  62                                           nh_table_id=1)])
  63         ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
  64                             [VppRoutePath("0.0.0.0",
  65                                           0xffffffff,
  66                                           nh_table_id=0)], table_id=1)
  67         ip_t01.add_vpp_config()
  68         ip_t10.add_vpp_config()
  69
  70         # Start builtin server and client with small private segments
  71         uri = "tcp://" + self.loop0.local_ip4 + "/1234"
  72         error = self.vapi.cli("test echo server appns 0 fifo-size 64 " +
  73                               "private-segment-size 1m uri " + uri)
  74         if error:
  75             self.logger.critical(error)
  76             self.assertNotIn("failed", error)
  77
  78         error = self.vapi.cli("test echo client nclients 100 appns 1 " +
  79                               "no-output fifo-size 64 syn-timeout 2 " +
  80                               "private-segment-size 1m uri " + uri)
  81         if error:
  82             self.logger.critical(error)
  83             self.assertNotIn("failed", error)
  84
  85         if self.vpp_dead:
  86             self.assert_equal(0)
  87
  88         # Delete inter-table routes
  89         ip_t01.remove_vpp_config()
  90         ip_t10.remove_vpp_config()
  91
  92
  93 class TestSessionUnitTests(VppTestCase):
  94     """ Session Unit Tests Case """
  95
  96     @classmethod
  97     def setUpClass(cls):
  98         super(TestSessionUnitTests, cls).setUpClass()
  99
 100     @classmethod
 101     def tearDownClass(cls):
 102         super(TestSessionUnitTests, cls).tearDownClass()
 103
 104     def setUp(self):
 105         super(TestSessionUnitTests, self).setUp()
 106         self.vapi.session_enable_disable(is_enable=1)
 107
 108     def test_session(self):
 109         """ Session Unit Tests """
 110         error = self.vapi.cli("test session all")
 111
 112         if error:
 113             self.logger.critical(error)
 114         self.assertNotIn("failed", error)
 115
 116     def tearDown(self):
 117         super(TestSessionUnitTests, self).tearDown()
 118         self.vapi.session_enable_disable(is_enable=0)
 119
 120
 121 @tag_run_solo
 122 class TestSvmFifoUnitTests(VppTestCase):
 123     """ SVM Fifo Unit Tests Case """
 124
 125     @classmethod
 126     def setUpClass(cls):
 127         super(TestSvmFifoUnitTests, cls).setUpClass()
 128
 129     @classmethod
 130     def tearDownClass(cls):
 131         super(TestSvmFifoUnitTests, cls).tearDownClass()
 132
 133     def setUp(self):
 134         super(TestSvmFifoUnitTests, self).setUp()
 135
 136     def test_svm_fifo(self):
 137         """ SVM Fifo Unit Tests """
 138         error = self.vapi.cli("test svm fifo all")
 139
 140         if error:
 141             self.logger.critical(error)
 142         self.assertNotIn("failed", error)
 143
 144     def tearDown(self):
 145         super(TestSvmFifoUnitTests, self).tearDown()
 146
 147 if __name__ == '__main__':
 148     unittest.main(testRunner=VppTestRunner)