+class VCLThruHostStackTLS(VCLTestCase):
+ """VCL Thru Host Stack TLS"""
+
+ @classmethod
+ def setUpClass(cls):
+ cls.session_startup = ["poll-main", "use-app-socket-api"]
+ super(VCLThruHostStackTLS, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLThruHostStackTLS, cls).tearDownClass()
+
+ def setUp(self):
+ super(VCLThruHostStackTLS, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_uni_dir_tls_timeout = 20
+ self.server_tls_args = ["-L", self.server_port]
+ self.client_uni_dir_tls_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-L",
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+ self.sapi_server_sock = "1"
+ self.sapi_client_sock = "2"
+
+ def test_vcl_thru_host_stack_tls_uni_dir(self):
+ """run VCL thru host stack uni-directional TLS test"""
+
+ self.timeout = self.client_uni_dir_tls_timeout
+ self.thru_host_stack_test(
+ "vcl_test_server",
+ self.server_tls_args,
+ "vcl_test_client",
+ self.client_uni_dir_tls_test_args,
+ )
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(VCLThruHostStackTLS, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show app server"))
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+
+class VCLThruHostStackDTLS(VCLTestCase):
+ """VCL Thru Host Stack DTLS"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(VCLThruHostStackDTLS, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLThruHostStackDTLS, cls).tearDownClass()
+
+ def setUp(self):
+ super(VCLThruHostStackDTLS, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_uni_dir_dtls_timeout = 20
+ self.server_dtls_args = ["-p", "dtls", self.server_port]
+ self.client_uni_dir_dtls_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-p",
+ "dtls",
+ "-T 1400",
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+
+ def test_vcl_thru_host_stack_dtls_uni_dir(self):
+ """run VCL thru host stack uni-directional DTLS test"""
+
+ self.timeout = self.client_uni_dir_dtls_timeout
+ self.thru_host_stack_test(
+ "vcl_test_server",
+ self.server_dtls_args,
+ "vcl_test_client",
+ self.client_uni_dir_dtls_test_args,
+ )
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(VCLThruHostStackDTLS, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show app server"))
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+
+class VCLThruHostStackQUIC(VCLTestCase):
+ """VCL Thru Host Stack QUIC"""
+
+ @classmethod
+ def setUpClass(cls):
+ cls.extra_vpp_plugin_config.append("plugin quic_plugin.so { enable }")
+ super(VCLThruHostStackQUIC, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLThruHostStackQUIC, cls).tearDownClass()
+
+ def setUp(self):
+ super(VCLThruHostStackQUIC, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_uni_dir_quic_timeout = 20
+ self.server_quic_args = ["-p", "quic", self.server_port]
+ self.client_uni_dir_quic_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-p",
+ "quic",
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ def test_vcl_thru_host_stack_quic_uni_dir(self):
+ """run VCL thru host stack uni-directional QUIC test"""
+
+ self.timeout = self.client_uni_dir_quic_timeout
+ self.thru_host_stack_test(
+ "vcl_test_server",
+ self.server_quic_args,
+ "vcl_test_client",
+ self.client_uni_dir_quic_test_args,
+ )
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(VCLThruHostStackQUIC, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show app server"))
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+
+class VCLThruHostStackBidirNsock(VCLTestCase):
+ """VCL Thru Host Stack Bidir Nsock"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(VCLThruHostStackBidirNsock, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLThruHostStackBidirNsock, cls).tearDownClass()
+
+ def setUp(self):
+ super(VCLThruHostStackBidirNsock, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_bi_dir_nsock_timeout = 20
+ self.client_bi_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-B",
+ "-X",
+ "-I",
+ "2",
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+ self.client_echo_test_args = [
+ "-E",
+ self.echo_phrase,
+ "-X",
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(VCLThruHostStackBidirNsock, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ def test_vcl_thru_host_stack_bi_dir_nsock(self):
+ """run VCL thru host stack bi-directional (multiple sockets) test"""
+
+ self.timeout = self.client_bi_dir_nsock_timeout
+ self.thru_host_stack_test(
+ "vcl_test_server",
+ self.server_args,
+ "vcl_test_client",
+ self.client_bi_dir_nsock_test_args,
+ )
+
+
+class LDPThruHostStackBidirNsock(VCLTestCase):
+ """LDP Thru Host Stack Bidir Nsock"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(LDPThruHostStackBidirNsock, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(LDPThruHostStackBidirNsock, cls).tearDownClass()
+
+ def setUp(self):
+ super(LDPThruHostStackBidirNsock, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_bi_dir_nsock_timeout = 20
+ self.client_bi_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-B",
+ "-X",
+ # OUCH! Host Stack Bug?
+ # Only fails when running
+ # 'make test TEST_JOBS=auto'
+ # or TEST_JOBS > 1
+ # "-I", "2",
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(LDPThruHostStackBidirNsock, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ def test_ldp_thru_host_stack_bi_dir_nsock(self):
+ """run LDP thru host stack bi-directional (multiple sockets) test"""
+
+ self.timeout = self.client_bi_dir_nsock_timeout
+ self.thru_host_stack_test(
+ "sock_test_server",
+ self.server_args,
+ "sock_test_client",
+ self.client_bi_dir_nsock_test_args,
+ )
+
+
+class LDPThruHostStackNsock(VCLTestCase):
+ """LDP Thru Host Stack Nsock"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(LDPThruHostStackNsock, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(LDPThruHostStackNsock, cls).tearDownClass()
+
+ def setUp(self):
+ super(LDPThruHostStackNsock, self).setUp()
+
+ self.thru_host_stack_setup()
+ if self.vppDebug:
+ self.client_uni_dir_nsock_timeout = 20
+ self.numSockets = "2"
+ else:
+ self.client_uni_dir_nsock_timeout = 20
+ self.numSockets = "5"
+
+ self.client_uni_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-I",
+ self.numSockets,
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(LDPThruHostStackNsock, self).tearDown()
+
+ def test_ldp_thru_host_stack_uni_dir_nsock(self):
+ """run LDP thru host stack uni-directional (multiple sockets) test"""
+
+ self.timeout = self.client_uni_dir_nsock_timeout
+ self.thru_host_stack_test(
+ "sock_test_server",
+ self.server_args,
+ "sock_test_client",
+ self.client_uni_dir_nsock_test_args,
+ )
+
+
+class VCLThruHostStackNsock(VCLTestCase):
+ """VCL Thru Host Stack Nsock"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(VCLThruHostStackNsock, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLThruHostStackNsock, cls).tearDownClass()
+
+ def setUp(self):
+ super(VCLThruHostStackNsock, self).setUp()
+
+ self.thru_host_stack_setup()
+ if self.vppDebug:
+ self.client_uni_dir_nsock_timeout = 20
+ self.numSockets = "2"
+ else:
+ self.client_uni_dir_nsock_timeout = 20
+ self.numSockets = "5"
+
+ self.client_uni_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-I",
+ self.numSockets,
+ self.loop0.local_ip4,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(VCLThruHostStackNsock, self).tearDown()
+
+ def test_vcl_thru_host_stack_uni_dir_nsock(self):
+ """run VCL thru host stack uni-directional (multiple sockets) test"""
+
+ self.timeout = self.client_uni_dir_nsock_timeout
+ self.thru_host_stack_test(
+ "vcl_test_server",
+ self.server_args,
+ "vcl_test_client",
+ self.client_uni_dir_nsock_test_args,
+ )
+
+
+class LDPThruHostStackIperf(VCLTestCase):
+ """LDP Thru Host Stack Iperf"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(LDPThruHostStackIperf, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(LDPThruHostStackIperf, cls).tearDownClass()
+
+ def setUp(self):
+ super(LDPThruHostStackIperf, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_iperf3_timeout = 20
+ self.client_iperf3_args = ["-4", "-t 2", "-c", self.loop0.local_ip4]
+ self.server_iperf3_args = ["-4", "-s"]
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(LDPThruHostStackIperf, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
+ def test_ldp_thru_host_stack_iperf3(self):
+ """run LDP thru host stack iperf3 test"""
+
+ self.timeout = self.client_iperf3_timeout
+ self.thru_host_stack_test(
+ iperf3, self.server_iperf3_args, iperf3, self.client_iperf3_args
+ )
+
+
+class LDPThruHostStackIperfUdp(VCLTestCase):
+ """LDP Thru Host Stack Iperf UDP"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(LDPThruHostStackIperfUdp, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(LDPThruHostStackIperfUdp, cls).tearDownClass()
+
+ def setUp(self):
+ super(LDPThruHostStackIperfUdp, self).setUp()
+
+ self.thru_host_stack_setup()
+ self.client_iperf3_timeout = 20
+ self.client_iperf3_args = [
+ "-4",
+ "-t 2",
+ "-u",
+ "-l 1400",
+ "-c",
+ self.loop0.local_ip4,
+ ]
+ self.server_iperf3_args = ["-4", "-s"]
+
+ def tearDown(self):
+ self.thru_host_stack_tear_down()
+ super(LDPThruHostStackIperfUdp, self).tearDown()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
+ def test_ldp_thru_host_stack_iperf3_udp(self):
+ """run LDP thru host stack iperf3 UDP test"""
+
+ self.timeout = self.client_iperf3_timeout
+ self.thru_host_stack_test(
+ iperf3, self.server_iperf3_args, iperf3, self.client_iperf3_args
+ )
+
+
+class LDPIpv6CutThruTestCase(VCLTestCase):
+ """LDP IPv6 Cut Thru Tests"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(LDPIpv6CutThruTestCase, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(LDPIpv6CutThruTestCase, cls).tearDownClass()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ def setUp(self):
+ super(LDPIpv6CutThruTestCase, self).setUp()
+
+ self.cut_thru_setup()
+ self.client_iperf3_timeout = 20
+ self.client_uni_dir_nsock_timeout = 20
+ self.client_bi_dir_nsock_timeout = 20
+ self.client_ipv6_echo_test_args = [
+ "-6",
+ "-E",
+ self.echo_phrase,
+ "-X",
+ self.server_ipv6_addr,
+ self.server_port,
+ ]
+ self.client_ipv6_iperf3_args = ["-6", "-t 2", "-c", self.server_ipv6_addr]
+ self.server_ipv6_iperf3_args = ["-6", "-s"]
+ self.client_ipv6_uni_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-6",
+ "-I",
+ "2",
+ self.server_ipv6_addr,
+ self.server_port,
+ ]
+ self.client_ipv6_bi_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-B",
+ "-X",
+ "-6",
+ "-I",
+ "2",
+ self.server_ipv6_addr,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ super(LDPIpv6CutThruTestCase, self).tearDown()
+ self.cut_thru_tear_down()
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ def test_ldp_ipv6_cut_thru_echo(self):
+ """run LDP IPv6 cut thru echo test"""
+
+ self.cut_thru_test(
+ "sock_test_server",
+ self.server_ipv6_args,
+ "sock_test_client",
+ self.client_ipv6_echo_test_args,
+ )
+
+ @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
+ def test_ldp_ipv6_cut_thru_iperf3(self):
+ """run LDP IPv6 cut thru iperf3 test"""
+
+ self.timeout = self.client_iperf3_timeout
+ self.cut_thru_test(
+ iperf3, self.server_ipv6_iperf3_args, iperf3, self.client_ipv6_iperf3_args
+ )
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ def test_ldp_ipv6_cut_thru_uni_dir_nsock(self):
+ """run LDP IPv6 cut thru uni-directional (multiple sockets) test"""
+
+ self.timeout = self.client_uni_dir_nsock_timeout
+ self.cut_thru_test(
+ "sock_test_server",
+ self.server_ipv6_args,
+ "sock_test_client",
+ self.client_ipv6_uni_dir_nsock_test_args,
+ )
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ @unittest.skip("sock test apps need to be improved")
+ def test_ldp_ipv6_cut_thru_bi_dir_nsock(self):
+ """run LDP IPv6 cut thru bi-directional (multiple sockets) test"""
+
+ self.timeout = self.client_bi_dir_nsock_timeout
+ self.cut_thru_test(
+ "sock_test_server",
+ self.server_ipv6_args,
+ "sock_test_client",
+ self.client_ipv6_bi_dir_nsock_test_args,
+ )
+
+
+class VCLIpv6CutThruTestCase(VCLTestCase):
+ """VCL IPv6 Cut Thru Tests"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(VCLIpv6CutThruTestCase, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLIpv6CutThruTestCase, cls).tearDownClass()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ def setUp(self):
+ super(VCLIpv6CutThruTestCase, self).setUp()
+
+ self.cut_thru_setup()
+ self.client_uni_dir_nsock_timeout = 20
+ self.client_bi_dir_nsock_timeout = 20
+ self.client_ipv6_echo_test_args = [
+ "-6",
+ "-E",
+ self.echo_phrase,
+ "-X",
+ self.server_ipv6_addr,
+ self.server_port,
+ ]
+ self.client_ipv6_uni_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-U",
+ "-X",
+ "-6",
+ "-I",
+ "2",
+ self.server_ipv6_addr,
+ self.server_port,
+ ]
+ self.client_ipv6_bi_dir_nsock_test_args = [
+ "-N",
+ "1000",
+ "-B",
+ "-X",
+ "-6",
+ "-I",
+ "2",
+ self.server_ipv6_addr,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ super(VCLIpv6CutThruTestCase, self).tearDown()
+ self.cut_thru_tear_down()
+
+ def show_commands_at_teardown(self):
+ self.logger.debug(self.vapi.cli("show session verbose 2"))
+ self.logger.debug(self.vapi.cli("show app mq"))
+
+ def test_vcl_ipv6_cut_thru_echo(self):
+ """run VCL IPv6 cut thru echo test"""
+
+ self.cut_thru_test(
+ "vcl_test_server",
+ self.server_ipv6_args,
+ "vcl_test_client",
+ self.client_ipv6_echo_test_args,
+ )
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ def test_vcl_ipv6_cut_thru_uni_dir_nsock(self):
+ """run VCL IPv6 cut thru uni-directional (multiple sockets) test"""
+
+ self.timeout = self.client_uni_dir_nsock_timeout
+ self.cut_thru_test(
+ "vcl_test_server",
+ self.server_ipv6_args,
+ "vcl_test_client",
+ self.client_ipv6_uni_dir_nsock_test_args,
+ )
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ def test_vcl_ipv6_cut_thru_bi_dir_nsock(self):
+ """run VCL IPv6 cut thru bi-directional (multiple sockets) test"""
+
+ self.timeout = self.client_bi_dir_nsock_timeout
+ self.cut_thru_test(
+ "vcl_test_server",
+ self.server_ipv6_args,
+ "vcl_test_client",
+ self.client_ipv6_bi_dir_nsock_test_args,
+ )
+
+
+class VCLIpv6ThruHostStackEcho(VCLTestCase):
+ """VCL IPv6 Thru Host Stack Echo"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(VCLIpv6ThruHostStackEcho, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(VCLIpv6ThruHostStackEcho, cls).tearDownClass()
+
+ def setUp(self):
+ super(VCLIpv6ThruHostStackEcho, self).setUp()
+
+ self.thru_host_stack_ipv6_setup()
+ self.client_ipv6_echo_test_args = [
+ "-6",
+ "-E",
+ self.echo_phrase,
+ "-X",
+ self.loop0.local_ip6,
+ self.server_port,
+ ]
+
+ def tearDown(self):
+ self.thru_host_stack_ipv6_tear_down()
+ super(VCLIpv6ThruHostStackEcho, self).tearDown()
+
+ def test_vcl_ipv6_thru_host_stack_echo(self):
+ """run VCL IPv6 thru host stack echo test"""
+
+ self.thru_host_stack_test(
+ "vcl_test_server",
+ self.server_ipv6_args,
+ "vcl_test_client",
+ self.client_ipv6_echo_test_args,
+ )
+