test/asf/test_session.py

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