test/test_quic.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import os
   5 from framework import VppTestCase, VppTestRunner, running_extended_tests
   6 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
   7
   8
   9 class QUICTestCase(VppTestCase):
  10     """ QUIC Test Case """
  11
  12     @classmethod
  13     def setUpClass(cls):
  14         super(QUICTestCase, cls).setUpClass()
  15
  16     @classmethod
  17     def tearDownClass(cls):
  18         super(QUICTestCase, cls).tearDownClass()
  19
  20     def setUp(self):
  21         var = "VPP_BUILD_DIR"
  22         self.build_dir = os.getenv(var, None)
  23         if self.build_dir is None:
  24             raise Exception("Environment variable `%s' not set" % var)
  25         self.vppDebug = 'vpp_debug' in self.build_dir
  26         self.timeout = 20
  27         self.pre_test_sleep = 0.3
  28         self.post_test_sleep = 0.3
  29         self.vapi.session_enable_disable(is_enabled=1)
  30
  31     def tearDown(self):
  32         self.vapi.session_enable_disable(is_enabled=0)
  33
  34     def thru_host_stack_ipv4_setup(self):
  35         super(QUICTestCase, self).setUp()
  36
  37         self.create_loopback_interfaces(2)
  38         self.uri = "quic://%s/1234" % self.loop0.local_ip4
  39         common_args = ["uri", self.uri, "fifo-size", "4"]
  40         self.server_echo_test_args = common_args + ["appns", "server"]
  41         self.client_echo_test_args = common_args + ["appns", "client",
  42                                                     "bytes", "1024",
  43                                                     "test-bytes",
  44                                                     "no-output"]
  45         table_id = 1
  46         for i in self.lo_interfaces:
  47             i.admin_up()
  48
  49             if table_id != 0:
  50                 tbl = VppIpTable(self, table_id)
  51                 tbl.add_vpp_config()
  52
  53             i.set_table_ip4(table_id)
  54             i.config_ip4()
  55             table_id += 1
  56
  57         # Configure namespaces
  58         self.vapi.app_namespace_add_del(namespace_id=b"server",
  59                                         sw_if_index=self.loop0.sw_if_index)
  60         self.vapi.app_namespace_add_del(namespace_id=b"client",
  61                                         sw_if_index=self.loop1.sw_if_index)
  62
  63         # Add inter-table routes
  64         self.ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
  65                                  [VppRoutePath("0.0.0.0",
  66                                                0xffffffff,
  67                                                nh_table_id=2)], table_id=1)
  68         self.ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
  69                                  [VppRoutePath("0.0.0.0",
  70                                                0xffffffff,
  71                                                nh_table_id=1)], table_id=2)
  72         self.ip_t01.add_vpp_config()
  73         self.ip_t10.add_vpp_config()
  74         self.logger.debug(self.vapi.cli("show ip fib"))
  75
  76     def thru_host_stack_ipv4_tear_down(self):
  77         # Delete inter-table routes
  78         self.ip_t01.remove_vpp_config()
  79         self.ip_t10.remove_vpp_config()
  80
  81         for i in self.lo_interfaces:
  82             i.unconfig_ip4()
  83             i.set_table_ip4(0)
  84             i.admin_down()
  85
  86     def start_internal_echo_server(self, args):
  87         error = self.vapi.cli("test echo server %s" % ' '.join(args))
  88         if error:
  89             self.logger.critical(error)
  90             self.assertNotIn("failed", error)
  91
  92     def start_internal_echo_client(self, args):
  93         error = self.vapi.cli("test echo client %s" % ' '.join(args))
  94         if error:
  95             self.logger.critical(error)
  96             self.assertNotIn("failed", error)
  97
  98     def internal_ipv4_transfer_test(self, server_args, client_args):
  99         self.start_internal_echo_server(server_args)
 100         self.sleep(self.pre_test_sleep)
 101         self.start_internal_echo_client(client_args)
 102         self.sleep(self.post_test_sleep)
 103
 104
 105 class QUICInternalEchoIPv4TestCase(QUICTestCase):
 106     """ QUIC Internal Echo IPv4 Transfer Test Cases """
 107
 108     @classmethod
 109     def setUpClass(cls):
 110         super(QUICInternalEchoIPv4TestCase, cls).setUpClass()
 111
 112     @classmethod
 113     def tearDownClass(cls):
 114         super(QUICInternalEchoIPv4TestCase, cls).tearDownClass()
 115
 116     def setUp(self):
 117         super(QUICInternalEchoIPv4TestCase, self).setUp()
 118         self.thru_host_stack_ipv4_setup()
 119
 120     def tearDown(self):
 121         self.thru_host_stack_ipv4_tear_down()
 122         super(QUICInternalEchoIPv4TestCase, self).tearDown()
 123
 124     def show_commands_at_teardown(self):
 125         self.logger.debug(self.vapi.cli("show session verbose 2"))
 126
 127     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 128     def test_quic_internal_transfer(self):
 129         """ QUIC internal echo client/server transfer """
 130
 131         self.internal_ipv4_transfer_test(self.server_echo_test_args,
 132                                          self.client_echo_test_args)
 133
 134
 135 class QUICInternalEchoIPv4MultiStreamTestCase(QUICTestCase):
 136     """ QUIC Internal Echo IPv4 Transfer Test Cases """
 137
 138     @classmethod
 139     def setUpClass(cls):
 140         super(QUICInternalEchoIPv4MultiStreamTestCase, cls).setUpClass()
 141
 142     @classmethod
 143     def tearDownClass(cls):
 144         super(QUICInternalEchoIPv4MultiStreamTestCase, cls).tearDownClass()
 145
 146     def setUp(self):
 147         super(QUICInternalEchoIPv4MultiStreamTestCase, self).setUp()
 148         self.thru_host_stack_ipv4_setup()
 149
 150     def tearDown(self):
 151         self.thru_host_stack_ipv4_tear_down()
 152         super(QUICInternalEchoIPv4MultiStreamTestCase, self).tearDown()
 153
 154     def show_commands_at_teardown(self):
 155         self.logger.debug(self.vapi.cli("show session verbose 2"))
 156
 157     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 158     def test_quic_internal_multistream_transfer(self):
 159         """ QUIC internal echo client/server multi-stream transfer """
 160
 161         self.internal_ipv4_transfer_test(self.server_echo_test_args,
 162                                          self.client_echo_test_args +
 163                                          ["quic-streams", "10"])
 164
 165
 166 if __name__ == '__main__':
 167     unittest.main(testRunner=VppTestRunner)