test/test_quic.py

   1 #!/usr/bin/env python
   2 """ Vpp QUIC tests """
   3
   4 import unittest
   5 import os
   6 import subprocess
   7 import signal
   8 from framework import VppTestCase, VppTestRunner, running_extended_tests, \
   9     Worker
  10 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
  11
  12
  13 class QUICAppWorker(Worker):
  14     """ QUIC Test Application Worker """
  15
  16     def __init__(self, build_dir, appname, args, logger, env={}):
  17         app = "%s/vpp/bin/%s" % (build_dir, appname)
  18         self.args = [app] + args
  19         super(QUICAppWorker, self).__init__(self.args, logger, env)
  20
  21
  22 class QUICTestCase(VppTestCase):
  23     """ QUIC Test Case """
  24
  25     @classmethod
  26     def setUpClass(cls):
  27         super(QUICTestCase, cls).setUpClass()
  28
  29     @classmethod
  30     def tearDownClass(cls):
  31         super(QUICTestCase, cls).tearDownClass()
  32
  33     def setUp(self):
  34         var = "VPP_BUILD_DIR"
  35         self.build_dir = os.getenv(var, None)
  36         if self.build_dir is None:
  37             raise Exception("Environment variable `%s' not set" % var)
  38         self.vppDebug = 'vpp_debug' in self.build_dir
  39         self.timeout = 20
  40         self.pre_test_sleep = 0.3
  41         self.post_test_sleep = 0.3
  42         self.vapi.session_enable_disable(is_enabled=1)
  43
  44     def tearDown(self):
  45         self.vapi.session_enable_disable(is_enabled=0)
  46
  47     def thru_host_stack_ipv4_setup(self):
  48         super(QUICTestCase, self).setUp()
  49
  50         self.create_loopback_interfaces(2)
  51         self.uri = "quic://%s/1234" % self.loop0.local_ip4
  52         common_args = ["uri", self.uri, "fifo-size", "64"]
  53         self.server_echo_test_args = common_args + ["appns", "server"]
  54         self.client_echo_test_args = common_args + ["appns", "client",
  55                                                     "test-bytes"]
  56         table_id = 1
  57         for i in self.lo_interfaces:
  58             i.admin_up()
  59
  60             if table_id != 0:
  61                 tbl = VppIpTable(self, table_id)
  62                 tbl.add_vpp_config()
  63
  64             i.set_table_ip4(table_id)
  65             i.config_ip4()
  66             table_id += 1
  67
  68         # Configure namespaces
  69         self.vapi.app_namespace_add_del(namespace_id=b"server",
  70                                         sw_if_index=self.loop0.sw_if_index)
  71         self.vapi.app_namespace_add_del(namespace_id=b"client",
  72                                         sw_if_index=self.loop1.sw_if_index)
  73
  74         # Add inter-table routes
  75         self.ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
  76                                  [VppRoutePath("0.0.0.0",
  77                                                0xffffffff,
  78                                                nh_table_id=2)], table_id=1)
  79         self.ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
  80                                  [VppRoutePath("0.0.0.0",
  81                                                0xffffffff,
  82                                                nh_table_id=1)], table_id=2)
  83         self.ip_t01.add_vpp_config()
  84         self.ip_t10.add_vpp_config()
  85         self.logger.debug(self.vapi.cli("show ip fib"))
  86
  87     def thru_host_stack_ipv4_tear_down(self):
  88         # Delete inter-table routes
  89         self.ip_t01.remove_vpp_config()
  90         self.ip_t10.remove_vpp_config()
  91
  92         for i in self.lo_interfaces:
  93             i.unconfig_ip4()
  94             i.set_table_ip4(0)
  95             i.admin_down()
  96
  97     def start_internal_echo_server(self, args):
  98         error = self.vapi.cli("test echo server %s" % ' '.join(args))
  99         if error:
 100             self.logger.critical(error)
 101             self.assertNotIn("failed", error)
 102
 103     def start_internal_echo_client(self, args):
 104         error = self.vapi.cli("test echo client %s" % ' '.join(args))
 105         if error:
 106             self.logger.critical(error)
 107             self.assertNotIn("failed", error)
 108
 109     def internal_ipv4_transfer_test(self, server_args, client_args):
 110         self.start_internal_echo_server(server_args)
 111         self.start_internal_echo_client(client_args)
 112
 113     def start_external_echo_server(self, args):
 114         self.worker_server = QUICAppWorker(self.build_dir, "quic_echo",
 115                                            args, self.logger)
 116         self.worker_server.start()
 117
 118     def start_external_echo_client(self, args):
 119         self.client_echo_test_args += "use-svm-api"
 120         self.worker_client = QUICAppWorker(self.build_dir, "quic_echo",
 121                                            args, self.logger)
 122         self.worker_client.start()
 123         self.worker_client.join(self.timeout)
 124         try:
 125             self.validateExternalTestResults()
 126         except Exception as error:
 127             self.fail("Failed with %s" % error)
 128
 129     def external_ipv4_transfer_test(self, server_args, client_args):
 130         self.start_external_echo_server(server_args)
 131         self.sleep(self.pre_test_sleep)
 132         self.start_external_echo_client(client_args)
 133         self.sleep(self.post_test_sleep)
 134
 135     def validateExternalTestResults(self):
 136         if os.path.isdir('/proc/{}'.format(self.worker_server.process.pid)):
 137             self.logger.info("Killing server worker process (pid %d)" %
 138                              self.worker_server.process.pid)
 139             os.killpg(os.getpgid(self.worker_server.process.pid),
 140                       signal.SIGTERM)
 141             self.worker_server.join()
 142         self.logger.info("Client worker result is `%s'" %
 143                          self.worker_client.result)
 144         error = False
 145         if self.worker_client.result is None:
 146             try:
 147                 error = True
 148                 self.logger.error(
 149                     "Timeout: %ss! Killing client worker process (pid %d)" %
 150                     (self.timeout, self.worker_client.process.pid))
 151                 os.killpg(os.getpgid(self.worker_client.process.pid),
 152                           signal.SIGKILL)
 153                 self.worker_client.join()
 154             except OSError:
 155                 self.logger.debug(
 156                     "Couldn't kill client worker process")
 157                 raise
 158         if error:
 159             raise Exception(
 160                 "Timeout! Client worker did not finish in %ss" % timeout)
 161         self.assert_equal(self.worker_client.result, 0,
 162                           "Binary test return code")
 163
 164
 165 class QUICInternalEchoIPv4TestCase(QUICTestCase):
 166     """ QUIC Internal Echo IPv4 Transfer Test Cases """
 167
 168     @classmethod
 169     def setUpClass(cls):
 170         super(QUICInternalEchoIPv4TestCase, cls).setUpClass()
 171
 172     @classmethod
 173     def tearDownClass(cls):
 174         super(QUICInternalEchoIPv4TestCase, cls).tearDownClass()
 175
 176     def setUp(self):
 177         super(QUICInternalEchoIPv4TestCase, self).setUp()
 178         self.thru_host_stack_ipv4_setup()
 179
 180     def tearDown(self):
 181         self.thru_host_stack_ipv4_tear_down()
 182         super(QUICInternalEchoIPv4TestCase, self).tearDown()
 183
 184     def show_commands_at_teardown(self):
 185         self.logger.debug(self.vapi.cli("show session verbose 2"))
 186
 187     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 188     def test_quic_internal_transfer(self):
 189         """ QUIC internal echo client/server transfer """
 190
 191         self.internal_ipv4_transfer_test(self.server_echo_test_args,
 192                                          self.client_echo_test_args +
 193                                          ["no-output", "mbytes", "10"])
 194
 195
 196 class QUICInternalEchoIPv4MultiStreamTestCase(QUICTestCase):
 197     """ QUIC Internal Echo IPv4 Transfer Test Cases """
 198
 199     @classmethod
 200     def setUpClass(cls):
 201         super(QUICInternalEchoIPv4MultiStreamTestCase, cls).setUpClass()
 202
 203     @classmethod
 204     def tearDownClass(cls):
 205         super(QUICInternalEchoIPv4MultiStreamTestCase, cls).tearDownClass()
 206
 207     def setUp(self):
 208         super(QUICInternalEchoIPv4MultiStreamTestCase, self).setUp()
 209         self.thru_host_stack_ipv4_setup()
 210
 211     def tearDown(self):
 212         self.thru_host_stack_ipv4_tear_down()
 213         super(QUICInternalEchoIPv4MultiStreamTestCase, self).tearDown()
 214
 215     def show_commands_at_teardown(self):
 216         self.logger.debug(self.vapi.cli("show session verbose 2"))
 217
 218     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 219     def test_quic_internal_multistream_transfer(self):
 220         """ QUIC internal echo client/server multi-stream transfer """
 221
 222         self.internal_ipv4_transfer_test(self.server_echo_test_args,
 223                                          self.client_echo_test_args +
 224                                          ["quic-streams", "10",
 225                                           "mbytes", "1",
 226                                           "no-output"])
 227
 228
 229 class QUICExternalEchoIPv4TestCase(QUICTestCase):
 230     """ QUIC External Echo IPv4 Transfer Test Cases """
 231
 232     @classmethod
 233     def setUpClass(cls):
 234         super(QUICExternalEchoIPv4TestCase, cls).setUpClass()
 235
 236     @classmethod
 237     def tearDownClass(cls):
 238         super(QUICExternalEchoIPv4TestCase, cls).tearDownClass()
 239
 240     def setUp(self):
 241         super(QUICExternalEchoIPv4TestCase, self).setUp()
 242         self.thru_host_stack_ipv4_setup()
 243
 244     def tearDown(self):
 245         super(QUICExternalEchoIPv4TestCase, self).tearDown()
 246         self.thru_host_stack_ipv4_tear_down()
 247
 248     def show_commands_at_teardown(self):
 249         self.logger.debug(self.vapi.cli("show session verbose 2"))
 250
 251     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 252     def test_quic_external_transfer(self):
 253         """ QUIC external echo client/server transfer """
 254
 255         self.external_ipv4_transfer_test(self.server_echo_test_args +
 256                                          ["socket-name", self.api_sock],
 257                                          self.client_echo_test_args +
 258                                          ["socket-name", self.api_sock,
 259                                           "mbytes", "10"])
 260
 261
 262 if __name__ == '__main__':
 263     unittest.main(testRunner=VppTestRunner)