test/test_tcp.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from framework import VppTestCase, VppTestRunner
   6 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
   7
   8
   9 class TestTCP(VppTestCase):
  10     """TCP Test Case"""
  11
  12     @classmethod
  13     def setUpClass(cls):
  14         super(TestTCP, cls).setUpClass()
  15
  16     @classmethod
  17     def tearDownClass(cls):
  18         super(TestTCP, cls).tearDownClass()
  19
  20     def setUp(self):
  21         super(TestTCP, self).setUp()
  22         self.vapi.session_enable_disable(is_enable=1)
  23         self.create_loopback_interfaces(2)
  24
  25         table_id = 0
  26
  27         for i in self.lo_interfaces:
  28             i.admin_up()
  29
  30             if table_id != 0:
  31                 tbl = VppIpTable(self, table_id)
  32                 tbl.add_vpp_config()
  33
  34             i.set_table_ip4(table_id)
  35             i.config_ip4()
  36             table_id += 1
  37
  38         # Configure namespaces
  39         self.vapi.app_namespace_add_del(
  40             namespace_id="0", sw_if_index=self.loop0.sw_if_index
  41         )
  42         self.vapi.app_namespace_add_del(
  43             namespace_id="1", sw_if_index=self.loop1.sw_if_index
  44         )
  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         self.vapi.session_enable_disable(is_enable=0)
  52         super(TestTCP, self).tearDown()
  53
  54     def test_tcp_transfer(self):
  55         """TCP echo client/server transfer"""
  56
  57         # Add inter-table routes
  58         ip_t01 = VppIpRoute(
  59             self,
  60             self.loop1.local_ip4,
  61             32,
  62             [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=1)],
  63         )
  64         ip_t10 = VppIpRoute(
  65             self,
  66             self.loop0.local_ip4,
  67             32,
  68             [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=0)],
  69             table_id=1,
  70         )
  71         ip_t01.add_vpp_config()
  72         ip_t10.add_vpp_config()
  73
  74         # Start builtin server and client
  75         uri = "tcp://" + self.loop0.local_ip4 + "/1234"
  76         error = self.vapi.cli("test echo server appns 0 fifo-size 4 uri " + uri)
  77         if error:
  78             self.logger.critical(error)
  79             self.assertNotIn("failed", error)
  80
  81         error = self.vapi.cli(
  82             "test echo client mbytes 10 appns 1 "
  83             + "fifo-size 4 no-output test-bytes "
  84             + "syn-timeout 2 uri "
  85             + uri
  86         )
  87         if error:
  88             self.logger.critical(error)
  89             self.assertNotIn("failed", error)
  90
  91         # Delete inter-table routes
  92         ip_t01.remove_vpp_config()
  93         ip_t10.remove_vpp_config()
  94
  95
  96 class TestTCPUnitTests(VppTestCase):
  97     "TCP Unit Tests"
  98
  99     @classmethod
 100     def setUpClass(cls):
 101         super(TestTCPUnitTests, cls).setUpClass()
 102
 103     @classmethod
 104     def tearDownClass(cls):
 105         super(TestTCPUnitTests, cls).tearDownClass()
 106
 107     def setUp(self):
 108         super(TestTCPUnitTests, self).setUp()
 109         self.vapi.session_enable_disable(is_enable=1)
 110
 111     def tearDown(self):
 112         super(TestTCPUnitTests, self).tearDown()
 113         self.vapi.session_enable_disable(is_enable=0)
 114
 115     def test_tcp_unittest(self):
 116         """TCP Unit Tests"""
 117         error = self.vapi.cli("test tcp all")
 118
 119         if error:
 120             self.logger.critical(error)
 121         self.assertNotIn("failed", error)
 122
 123
 124 if __name__ == "__main__":
 125     unittest.main(testRunner=VppTestRunner)