8 from framework import VppTestCase, VppTestRunner, running_extended_tests, \
10 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath, FibPathProto
12 iperf3 = '/usr/bin/iperf3'
17 subprocess.check_output([app, '-v'])
18 except (subprocess.CalledProcessError, OSError):
23 _have_iperf3 = have_app(iperf3)
24 _have_permissions = os.geteuid() == 0
27 class VCLAppWorker(Worker):
28 """ VCL Test Application Worker """
30 def __init__(self, build_dir, appname, args, logger, env={}):
31 vcl_lib_dir = "%s/vpp/lib" % build_dir
32 if "iperf" in appname:
34 env.update({'LD_PRELOAD':
35 "%s/libvcl_ldpreload.so" % vcl_lib_dir})
36 elif "sock" in appname:
37 app = "%s/vpp/bin/%s" % (build_dir, appname)
38 env.update({'LD_PRELOAD':
39 "%s/libvcl_ldpreload.so" % vcl_lib_dir})
41 app = "%s/vpp/bin/%s" % (build_dir, appname)
42 self.args = [app] + args
43 super(VCLAppWorker, self).__init__(self.args, logger, env)
46 class VCLTestCase(VppTestCase):
47 """ VCL Test Class """
51 super(VCLTestCase, cls).setUpClass()
54 def tearDownClass(cls):
55 super(VCLTestCase, cls).tearDownClass()
59 self.build_dir = os.getenv(var, None)
60 if self.build_dir is None:
61 raise EnvironmentError("Environment variable `%s' not set" % var)
62 self.vppDebug = 'vpp_debug' in self.build_dir
63 self.server_addr = "127.0.0.1"
64 self.server_port = "22000"
65 self.server_args = [self.server_port]
66 self.server_ipv6_addr = "::1"
67 self.server_ipv6_args = ["-6", self.server_port]
69 self.echo_phrase = "Hello, world! Jenny is a friend of mine."
70 self.pre_test_sleep = 0.3
71 self.post_test_sleep = 0.2
73 if os.path.isfile("/tmp/ldp_server_af_unix_socket"):
74 os.remove("/tmp/ldp_server_af_unix_socket")
76 super(VCLTestCase, self).setUp()
78 def cut_thru_setup(self):
79 self.vapi.session_enable_disable(is_enabled=1)
81 def cut_thru_tear_down(self):
82 self.vapi.session_enable_disable(is_enabled=0)
84 def cut_thru_test(self, server_app, server_args, client_app, client_args):
85 self.env = {'VCL_API_PREFIX': self.shm_prefix,
86 'VCL_APP_SCOPE_LOCAL': "true"}
87 worker_server = VCLAppWorker(self.build_dir, server_app, server_args,
88 self.logger, self.env)
90 self.sleep(self.pre_test_sleep)
91 worker_client = VCLAppWorker(self.build_dir, client_app, client_args,
92 self.logger, self.env)
94 worker_client.join(self.timeout)
96 self.validateResults(worker_client, worker_server, self.timeout)
97 except Exception as error:
98 self.fail("Failed with %s" % error)
99 self.sleep(self.post_test_sleep)
101 def thru_host_stack_setup(self):
102 self.vapi.session_enable_disable(is_enabled=1)
103 self.create_loopback_interfaces(2)
107 for i in self.lo_interfaces:
111 tbl = VppIpTable(self, table_id)
114 i.set_table_ip4(table_id)
118 # Configure namespaces
119 self.vapi.app_namespace_add_del(namespace_id=b"1", secret=1234,
120 sw_if_index=self.loop0.sw_if_index)
121 self.vapi.app_namespace_add_del(namespace_id=b"2", secret=5678,
122 sw_if_index=self.loop1.sw_if_index)
124 # Add inter-table routes
125 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
126 [VppRoutePath("0.0.0.0",
128 nh_table_id=2)], table_id=1)
129 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
130 [VppRoutePath("0.0.0.0",
132 nh_table_id=1)], table_id=2)
133 ip_t01.add_vpp_config()
134 ip_t10.add_vpp_config()
135 self.logger.debug(self.vapi.cli("show ip fib"))
137 def thru_host_stack_tear_down(self):
138 for i in self.lo_interfaces:
143 def thru_host_stack_ipv6_setup(self):
144 self.vapi.session_enable_disable(is_enabled=1)
145 self.create_loopback_interfaces(2)
149 for i in self.lo_interfaces:
152 tbl = VppIpTable(self, table_id, is_ip6=1)
155 i.set_table_ip6(table_id)
159 # Configure namespaces
160 self.vapi.app_namespace_add_del(namespace_id=b"1", secret=1234,
161 sw_if_index=self.loop0.sw_if_index)
162 self.vapi.app_namespace_add_del(namespace_id=b"2", secret=5678,
163 sw_if_index=self.loop1.sw_if_index)
165 # Add inter-table routes
166 ip_t01 = VppIpRoute(self, self.loop1.local_ip6, 128,
167 [VppRoutePath("::0", 0xffffffff,
170 ip_t10 = VppIpRoute(self, self.loop0.local_ip6, 128,
171 [VppRoutePath("::0", 0xffffffff,
174 ip_t01.add_vpp_config()
175 ip_t10.add_vpp_config()
176 self.logger.debug(self.vapi.cli("show interface addr"))
177 self.logger.debug(self.vapi.cli("show ip6 fib"))
179 def thru_host_stack_ipv6_tear_down(self):
180 for i in self.lo_interfaces:
185 self.vapi.session_enable_disable(is_enabled=0)
187 @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
188 @unittest.skipUnless(_have_permissions, 'Need to run as root. Skipping.')
189 def thru_host_stack_test(self, server_app, server_args,
190 client_app, client_args):
191 self.env = {'VCL_API_PREFIX': self.shm_prefix,
192 'VCL_APP_SCOPE_GLOBAL': "true",
193 'VCL_APP_NAMESPACE_ID': "1",
194 'VCL_APP_NAMESPACE_SECRET': "1234"}
196 worker_server = VCLAppWorker(self.build_dir, server_app, server_args,
197 self.logger, self.env)
198 worker_server.start()
199 self.sleep(self.pre_test_sleep)
201 self.env.update({'VCL_APP_NAMESPACE_ID': "2",
202 'VCL_APP_NAMESPACE_SECRET': "5678"})
203 worker_client = VCLAppWorker(self.build_dir, client_app, client_args,
204 self.logger, self.env)
205 worker_client.start()
206 worker_client.join(self.timeout)
209 self.validateResults(worker_client, worker_server, self.timeout)
210 except Exception as error:
211 self.fail("Failed with %s" % error)
212 self.sleep(self.post_test_sleep)
214 def validateResults(self, worker_client, worker_server, timeout):
215 if worker_server.process is None:
216 raise RuntimeError('worker_server is not running.')
217 if os.path.isdir('/proc/{}'.format(worker_server.process.pid)):
218 self.logger.info("Killing server worker process (pid %d)" %
219 worker_server.process.pid)
220 os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM)
222 self.logger.info("Client worker result is `%s'" % worker_client.result)
224 if worker_client.result is None:
228 "Timeout: %ss! Killing client worker process (pid %d)" %
229 (timeout, worker_client.process.pid))
230 os.killpg(os.getpgid(worker_client.process.pid),
235 "Couldn't kill client worker process")
239 "Timeout! Client worker did not finish in %ss" % timeout)
240 self.assert_equal(worker_client.result, 0, "Binary test return code")
243 class LDPCutThruTestCase(VCLTestCase):
244 """ LDP Cut Thru Tests """
248 super(LDPCutThruTestCase, cls).setUpClass()
251 def tearDownClass(cls):
252 super(LDPCutThruTestCase, cls).tearDownClass()
255 super(LDPCutThruTestCase, self).setUp()
257 self.cut_thru_setup()
258 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
259 self.server_addr, self.server_port]
260 self.client_iperf3_timeout = 20
261 self.client_iperf3_args = ["-V4d", "-t 2", "-c", self.server_addr]
262 self.server_iperf3_args = ["-V4d", "-s"]
263 self.client_uni_dir_nsock_timeout = 20
264 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
268 self.client_bi_dir_nsock_timeout = 20
269 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
275 super(LDPCutThruTestCase, self).tearDown()
276 self.cut_thru_tear_down()
278 def show_commands_at_teardown(self):
279 self.logger.debug(self.vapi.cli("show session verbose 2"))
281 @unittest.skipUnless(running_extended_tests, "part of extended tests")
282 def test_ldp_cut_thru_echo(self):
283 """ run LDP cut thru echo test """
285 self.cut_thru_test("sock_test_server", self.server_args,
286 "sock_test_client", self.client_echo_test_args)
288 @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
289 @unittest.skipUnless(_have_permissions, 'Need to run as root. Skipping.')
290 def test_ldp_cut_thru_iperf3(self):
291 """ run LDP cut thru iperf3 test """
293 self.timeout = self.client_iperf3_timeout
294 self.cut_thru_test(iperf3, self.server_iperf3_args,
295 iperf3, self.client_iperf3_args)
297 @unittest.skipUnless(running_extended_tests, "part of extended tests")
298 def test_ldp_cut_thru_uni_dir_nsock(self):
299 """ run LDP cut thru uni-directional (multiple sockets) test """
301 self.timeout = self.client_uni_dir_nsock_timeout
302 self.cut_thru_test("sock_test_server", self.server_args,
304 self.client_uni_dir_nsock_test_args)
306 @unittest.skipUnless(running_extended_tests, "part of extended tests")
307 def test_ldp_cut_thru_bi_dir_nsock(self):
308 """ run LDP cut thru bi-directional (multiple sockets) test """
310 self.timeout = self.client_bi_dir_nsock_timeout
311 self.cut_thru_test("sock_test_server", self.server_args,
313 self.client_bi_dir_nsock_test_args)
316 class VCLCutThruTestCase(VCLTestCase):
317 """ VCL Cut Thru Tests """
321 super(VCLCutThruTestCase, cls).setUpClass()
324 def tearDownClass(cls):
325 super(VCLCutThruTestCase, cls).tearDownClass()
328 super(VCLCutThruTestCase, self).setUp()
330 self.cut_thru_setup()
331 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
332 self.server_addr, self.server_port]
334 self.client_uni_dir_nsock_timeout = 20
335 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
339 self.client_bi_dir_nsock_timeout = 20
340 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
346 super(VCLCutThruTestCase, self).tearDown()
348 def show_commands_at_teardown(self):
349 self.logger.debug(self.vapi.cli("show session verbose 2"))
351 def test_vcl_cut_thru_echo(self):
352 """ run VCL cut thru echo test """
354 self.cut_thru_test("vcl_test_server", self.server_args,
355 "vcl_test_client", self.client_echo_test_args)
357 def test_vcl_cut_thru_uni_dir_nsock(self):
358 """ run VCL cut thru uni-directional (multiple sockets) test """
360 self.timeout = self.client_uni_dir_nsock_timeout
361 self.cut_thru_test("vcl_test_server", self.server_args,
363 self.client_uni_dir_nsock_test_args)
365 def test_vcl_cut_thru_bi_dir_nsock(self):
366 """ run VCL cut thru bi-directional (multiple sockets) test """
368 self.timeout = self.client_bi_dir_nsock_timeout
369 self.cut_thru_test("vcl_test_server", self.server_args,
371 self.client_bi_dir_nsock_test_args)
374 class VCLThruHostStackEcho(VCLTestCase):
375 """ VCL Thru Host Stack Echo """
379 super(VCLThruHostStackEcho, cls).setUpClass()
382 def tearDownClass(cls):
383 super(VCLThruHostStackEcho, cls).tearDownClass()
386 super(VCLThruHostStackEcho, self).setUp()
388 self.thru_host_stack_setup()
389 self.client_bi_dir_nsock_timeout = 20
390 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
392 self.loop0.local_ip4,
394 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
395 self.loop0.local_ip4,
399 self.thru_host_stack_tear_down()
400 super(VCLThruHostStackEcho, self).tearDown()
402 def show_commands_at_teardown(self):
403 self.logger.debug(self.vapi.cli("show app server"))
404 self.logger.debug(self.vapi.cli("show session verbose"))
407 class VCLThruHostStackTLS(VCLTestCase):
408 """ VCL Thru Host Stack TLS """
412 super(VCLThruHostStackTLS, cls).setUpClass()
415 def tearDownClass(cls):
416 super(VCLThruHostStackTLS, cls).tearDownClass()
419 super(VCLThruHostStackTLS, self).setUp()
421 self.thru_host_stack_setup()
422 self.client_uni_dir_tls_timeout = 20
423 self.server_tls_args = ["-L", self.server_port]
424 self.client_uni_dir_tls_test_args = ["-N", "1000", "-U", "-X", "-L",
425 self.loop0.local_ip4,
428 def test_vcl_thru_host_stack_tls_uni_dir(self):
429 """ run VCL thru host stack uni-directional TLS test """
431 self.timeout = self.client_uni_dir_tls_timeout
432 self.thru_host_stack_test("vcl_test_server", self.server_tls_args,
434 self.client_uni_dir_tls_test_args)
437 self.thru_host_stack_tear_down()
438 super(VCLThruHostStackTLS, self).tearDown()
440 def show_commands_at_teardown(self):
441 self.logger.debug(self.vapi.cli("show app server"))
442 self.logger.debug(self.vapi.cli("show session verbose 2"))
445 class VCLThruHostStackBidirNsock(VCLTestCase):
446 """ VCL Thru Host Stack Bidir Nsock """
450 super(VCLThruHostStackBidirNsock, cls).setUpClass()
453 def tearDownClass(cls):
454 super(VCLThruHostStackBidirNsock, cls).tearDownClass()
457 super(VCLThruHostStackBidirNsock, self).setUp()
459 self.thru_host_stack_setup()
460 self.client_bi_dir_nsock_timeout = 20
461 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
463 self.loop0.local_ip4,
465 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
466 self.loop0.local_ip4,
470 self.thru_host_stack_tear_down()
471 super(VCLThruHostStackBidirNsock, self).tearDown()
473 def show_commands_at_teardown(self):
474 self.logger.debug(self.vapi.cli("show session verbose 2"))
476 def test_vcl_thru_host_stack_bi_dir_nsock(self):
477 """ run VCL thru host stack bi-directional (multiple sockets) test """
479 self.timeout = self.client_bi_dir_nsock_timeout
480 self.thru_host_stack_test("vcl_test_server", self.server_args,
482 self.client_bi_dir_nsock_test_args)
485 class LDPThruHostStackBidirNsock(VCLTestCase):
486 """ LDP Thru Host Stack Bidir Nsock """
490 super(LDPThruHostStackBidirNsock, cls).setUpClass()
493 def tearDownClass(cls):
494 super(LDPThruHostStackBidirNsock, cls).tearDownClass()
497 super(LDPThruHostStackBidirNsock, self).setUp()
499 self.thru_host_stack_setup()
501 self.client_bi_dir_nsock_timeout = 20
502 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
503 # OUCH! Host Stack Bug?
505 self.loop0.local_ip4,
508 self.client_bi_dir_nsock_timeout = 20
509 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
510 # OUCH! Host Stack Bug?
512 self.loop0.local_ip4,
516 self.thru_host_stack_tear_down()
517 super(LDPThruHostStackBidirNsock, self).tearDown()
519 def show_commands_at_teardown(self):
520 self.logger.debug(self.vapi.cli("show session verbose 2"))
522 def test_ldp_thru_host_stack_bi_dir_nsock(self):
523 """ run LDP thru host stack bi-directional (multiple sockets) test """
525 self.timeout = self.client_bi_dir_nsock_timeout
526 self.thru_host_stack_test("sock_test_server", self.server_args,
528 self.client_bi_dir_nsock_test_args)
531 class LDPThruHostStackNsock(VCLTestCase):
532 """ LDP Thru Host Stack Nsock """
536 super(LDPThruHostStackNsock, cls).setUpClass()
539 def tearDownClass(cls):
540 super(LDPThruHostStackNsock, cls).tearDownClass()
543 super(LDPThruHostStackNsock, self).setUp()
545 self.thru_host_stack_setup()
547 self.client_uni_dir_nsock_timeout = 20
548 self.numSockets = "2"
550 self.client_uni_dir_nsock_timeout = 20
551 self.numSockets = "5"
553 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
554 "-I", self.numSockets,
555 self.loop0.local_ip4,
559 self.thru_host_stack_tear_down()
560 super(LDPThruHostStackNsock, self).tearDown()
562 def test_ldp_thru_host_stack_uni_dir_nsock(self):
563 """ run LDP thru host stack uni-directional (multiple sockets) test """
565 self.timeout = self.client_uni_dir_nsock_timeout
566 self.thru_host_stack_test("sock_test_server", self.server_args,
568 self.client_uni_dir_nsock_test_args)
571 class VCLThruHostStackNsock(VCLTestCase):
572 """ VCL Thru Host Stack Nsock """
576 super(VCLThruHostStackNsock, cls).setUpClass()
579 def tearDownClass(cls):
580 super(VCLThruHostStackNsock, cls).tearDownClass()
583 super(VCLThruHostStackNsock, self).setUp()
585 self.thru_host_stack_setup()
587 self.client_uni_dir_nsock_timeout = 20
588 self.numSockets = "2"
590 self.client_uni_dir_nsock_timeout = 20
591 self.numSockets = "5"
593 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
594 "-I", self.numSockets,
595 self.loop0.local_ip4,
599 self.thru_host_stack_tear_down()
600 super(VCLThruHostStackNsock, self).tearDown()
602 def test_vcl_thru_host_stack_uni_dir_nsock(self):
603 """ run VCL thru host stack uni-directional (multiple sockets) test """
605 self.timeout = self.client_uni_dir_nsock_timeout
606 self.thru_host_stack_test("vcl_test_server", self.server_args,
608 self.client_uni_dir_nsock_test_args)
611 class LDPThruHostStackIperf(VCLTestCase):
612 """ LDP Thru Host Stack Iperf """
616 super(LDPThruHostStackIperf, cls).setUpClass()
619 def tearDownClass(cls):
620 super(LDPThruHostStackIperf, cls).tearDownClass()
623 super(LDPThruHostStackIperf, self).setUp()
625 self.thru_host_stack_setup()
626 self.client_iperf3_timeout = 20
627 self.client_iperf3_args = ["-V4d", "-t 2", "-c", self.loop0.local_ip4]
628 self.server_iperf3_args = ["-V4d", "-s"]
631 self.thru_host_stack_tear_down()
632 super(LDPThruHostStackIperf, self).tearDown()
634 def show_commands_at_teardown(self):
635 self.logger.debug(self.vapi.cli("show session verbose 2"))
637 @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
638 @unittest.skipUnless(_have_permissions, 'Need to run as root. Skipping.')
639 def test_ldp_thru_host_stack_iperf3(self):
640 """ run LDP thru host stack iperf3 test """
642 self.timeout = self.client_iperf3_timeout
643 self.thru_host_stack_test(iperf3, self.server_iperf3_args,
644 iperf3, self.client_iperf3_args)
647 class LDPIpv6CutThruTestCase(VCLTestCase):
648 """ LDP IPv6 Cut Thru Tests """
652 super(LDPIpv6CutThruTestCase, cls).setUpClass()
655 def tearDownClass(cls):
656 super(LDPIpv6CutThruTestCase, cls).tearDownClass()
659 super(LDPIpv6CutThruTestCase, self).setUp()
661 self.cut_thru_setup()
662 self.client_iperf3_timeout = 20
663 self.client_uni_dir_nsock_timeout = 20
664 self.client_bi_dir_nsock_timeout = 20
665 self.client_ipv6_echo_test_args = ["-6", "-E", self.echo_phrase, "-X",
666 self.server_ipv6_addr,
668 self.client_ipv6_iperf3_args = ["-V6d", "-t 2", "-c",
669 self.server_ipv6_addr]
670 self.server_ipv6_iperf3_args = ["-V6d", "-s"]
671 self.client_ipv6_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
674 self.server_ipv6_addr,
676 self.client_ipv6_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
679 self.server_ipv6_addr,
683 super(LDPIpv6CutThruTestCase, self).tearDown()
684 self.cut_thru_tear_down()
686 def test_ldp_ipv6_cut_thru_echo(self):
687 """ run LDP IPv6 cut thru echo test """
689 self.cut_thru_test("sock_test_server",
690 self.server_ipv6_args,
692 self.client_ipv6_echo_test_args)
694 @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
695 @unittest.skipUnless(_have_permissions, 'Need to run as root. Skipping.')
696 @unittest.skipUnless(running_extended_tests, "part of extended tests")
697 def test_ldp_ipv6_cut_thru_iperf3(self):
698 """ run LDP IPv6 cut thru iperf3 test """
700 self.timeout = self.client_iperf3_timeout
701 self.cut_thru_test(iperf3, self.server_ipv6_iperf3_args,
702 iperf3, self.client_ipv6_iperf3_args)
704 @unittest.skipUnless(running_extended_tests, "part of extended tests")
705 def test_ldp_ipv6_cut_thru_uni_dir_nsock(self):
706 """ run LDP IPv6 cut thru uni-directional (multiple sockets) test """
708 self.timeout = self.client_uni_dir_nsock_timeout
709 self.cut_thru_test("sock_test_server", self.server_ipv6_args,
711 self.client_ipv6_uni_dir_nsock_test_args)
713 @unittest.skipUnless(running_extended_tests, "part of extended tests")
714 def test_ldp_ipv6_cut_thru_bi_dir_nsock(self):
715 """ run LDP IPv6 cut thru bi-directional (multiple sockets) test """
717 self.timeout = self.client_bi_dir_nsock_timeout
718 self.cut_thru_test("sock_test_server", self.server_ipv6_args,
720 self.client_ipv6_bi_dir_nsock_test_args)
723 class VCLIpv6CutThruTestCase(VCLTestCase):
724 """ VCL IPv6 Cut Thru Tests """
728 super(VCLIpv6CutThruTestCase, cls).setUpClass()
731 def tearDownClass(cls):
732 super(VCLIpv6CutThruTestCase, cls).tearDownClass()
735 super(VCLIpv6CutThruTestCase, self).setUp()
737 self.cut_thru_setup()
738 self.client_uni_dir_nsock_timeout = 20
739 self.client_bi_dir_nsock_timeout = 20
740 self.client_ipv6_echo_test_args = ["-6", "-E", self.echo_phrase, "-X",
741 self.server_ipv6_addr,
743 self.client_ipv6_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
746 self.server_ipv6_addr,
748 self.client_ipv6_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
751 self.server_ipv6_addr,
755 super(VCLIpv6CutThruTestCase, self).tearDown()
756 self.cut_thru_tear_down()
758 def test_vcl_ipv6_cut_thru_echo(self):
759 """ run VCL IPv6 cut thru echo test """
761 self.cut_thru_test("vcl_test_server",
762 self.server_ipv6_args,
764 self.client_ipv6_echo_test_args)
766 @unittest.skipUnless(running_extended_tests, "part of extended tests")
767 def test_vcl_ipv6_cut_thru_uni_dir_nsock(self):
768 """ run VCL IPv6 cut thru uni-directional (multiple sockets) test """
770 self.timeout = self.client_uni_dir_nsock_timeout
771 self.cut_thru_test("vcl_test_server", self.server_ipv6_args,
773 self.client_ipv6_uni_dir_nsock_test_args)
775 @unittest.skipUnless(running_extended_tests, "part of extended tests")
776 def test_vcl_ipv6_cut_thru_bi_dir_nsock(self):
777 """ run VCL IPv6 cut thru bi-directional (multiple sockets) test """
779 self.timeout = self.client_bi_dir_nsock_timeout
780 self.cut_thru_test("vcl_test_server", self.server_ipv6_args,
782 self.client_ipv6_bi_dir_nsock_test_args)
785 class VCLIpv6ThruHostStackEcho(VCLTestCase):
786 """ VCL IPv6 Thru Host Stack Echo """
790 super(VCLIpv6ThruHostStackEcho, cls).setUpClass()
793 def tearDownClass(cls):
794 super(VCLIpv6ThruHostStackEcho, cls).tearDownClass()
797 super(VCLIpv6ThruHostStackEcho, self).setUp()
799 self.thru_host_stack_ipv6_setup()
800 self.client_ipv6_echo_test_args = ["-6", "-E", self.echo_phrase, "-X",
801 self.loop0.local_ip6,
805 self.thru_host_stack_ipv6_tear_down()
806 super(VCLIpv6ThruHostStackEcho, self).tearDown()
808 def test_vcl_ipv6_thru_host_stack_echo(self):
809 """ run VCL IPv6 thru host stack echo test """
811 self.thru_host_stack_test("vcl_test_server",
812 self.server_ipv6_args,
814 self.client_ipv6_echo_test_args)
817 if __name__ == '__main__':
818 unittest.main(testRunner=VppTestRunner)