+
+class QUICEchoInternalTransferTestCase(QUICEchoInternalTestCase):
+ """QUIC Echo Internal Transfer Test Case"""
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_quic_internal_transfer(self):
+ self.server()
+ self.client("no-output", "mbytes", "2")
+
+
+class QUICEchoInternalSerialTestCase(QUICEchoInternalTestCase):
+ """QUIC Echo Internal Serial Transfer Test Case"""
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_quic_serial_internal_transfer(self):
+ self.server()
+ self.client("no-output", "mbytes", "2")
+ self.client("no-output", "mbytes", "2")
+ self.client("no-output", "mbytes", "2")
+ self.client("no-output", "mbytes", "2")
+ self.client("no-output", "mbytes", "2")
+
+
+class QUICEchoInternalMStreamTestCase(QUICEchoInternalTestCase):
+ """QUIC Echo Internal MultiStream Test Case"""
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_quic_internal_multistream_transfer(self):
+ self.server()
+ self.client("nclients", "10", "mbytes", "1", "no-output")
+
+
+class QUICEchoExternalTestCase(QUICTestCase):
+ extra_vpp_punt_config = ["session", "{", "evt_qs_memfd_seg", "}"]
+ quic_setup = "default"
+
+ def setUp(self):
+ super(QUICEchoExternalTestCase, self).setUp()
+ common_args = [
+ "uri",
+ self.uri,
+ "fifo-size",
+ "64",
+ "test-bytes:assert",
+ "socket-name",
+ self.api_sock]
+ self.server_echo_test_args = common_args + \
+ ["server", "appns", "server", "quic-setup", self.quic_setup]
+ self.client_echo_test_args = common_args + \
+ ["client", "appns", "client", "quic-setup", self.quic_setup]
+
+ def server(self, *args):
+ _args = self.server_echo_test_args + list(args)
+ self.worker_server = QUICAppWorker(
+ self.build_dir,
+ "quic_echo",
+ _args,
+ self.logger)
+ self.worker_server.start()
+ self.sleep(self.pre_test_sleep)
+
+ def client(self, *args):
+ _args = self.client_echo_test_args + list(args)
+ # self.client_echo_test_args += "use-svm-api"
+ self.worker_client = QUICAppWorker(
+ self.build_dir,
+ "quic_echo",
+ _args,
+ self.logger)
+ self.worker_client.start()
+ self.worker_client.join(self.timeout)
+ self.worker_server.join(self.timeout)
+ self.sleep(self.post_test_sleep)
+
+ def validate_external_test_results(self):
+ self.logger.info(
+ "Client worker result is `%s'" %
+ self.worker_client.result)
+ server_result = self.worker_server.result
+ client_result = self.worker_client.result
+ server_kill_error = False
+ if self.worker_server.result is None:
+ server_kill_error = self.worker_server.teardown(
+ self.logger, self.timeout)
+ if self.worker_client.result is None:
+ self.worker_client.teardown(self.logger, self.timeout)
+ self.assertEqual(server_result, 0, "Wrong server worker return code")
+ self.assertIsNotNone(
+ client_result,
+ "Timeout! Client worker did not finish in %ss" %
+ self.timeout)
+ self.assertEqual(client_result, 0, "Wrong client worker return code")
+ self.assertFalse(server_kill_error, "Server kill errored")
+
+
+class QUICEchoExternalTransferTestCase(QUICEchoExternalTestCase):
+ """QUIC Echo External Transfer Test Case"""
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_quic_external_transfer(self):
+ self.server()
+ self.client()
+ self.validate_external_test_results()
+
+
+class QUICEchoExternalServerStreamTestCase(QUICEchoExternalTestCase):
+ """QUIC Echo External Transfer Server Stream Test Case"""
+ quic_setup = "serverstream"
+
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_quic_external_transfer_server_stream(self):
+ self.server("TX=1Kb", "RX=0")
+ self.client("TX=0", "RX=1Kb")
+ self.validate_external_test_results()
+
+
+class QUICEchoExternalServerStreamWorkersTestCase(QUICEchoExternalTestCase):
+ """QUIC Echo External Transfer Server Stream MultiWorker Test Case"""
+ quic_setup = "serverstream"
+
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_quic_external_transfer_server_stream_multi_workers(self):
+ self.server("nclients", "4/4", "TX=1Kb", "RX=0")
+ self.client("nclients", "4/4", "TX=0", "RX=1Kb")
+ self.validate_external_test_results()
+