8 from framework import VppTestCase, VppTestRunner, running_extended_tests, \
10 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath, DpoProto
13 class VCLAppWorker(Worker):
14 """ VCL Test Application Worker """
16 def __init__(self, build_dir, appname, args, logger, env={}):
17 vcl_lib_dir = "%s/vpp/lib" % build_dir
18 if "iperf" in appname:
20 env.update({'LD_PRELOAD':
21 "%s/libvcl_ldpreload.so" % vcl_lib_dir})
22 elif "sock" in appname:
23 app = "%s/vpp/bin/%s" % (build_dir, appname)
24 env.update({'LD_PRELOAD':
25 "%s/libvcl_ldpreload.so" % vcl_lib_dir})
27 app = "%s/vpp/bin/%s" % (build_dir, appname)
28 self.args = [app] + args
29 super(VCLAppWorker, self).__init__(self.args, logger, env)
32 class VCLTestCase(VppTestCase):
33 """ VCL Test Class """
37 super(VCLTestCase, cls).setUpClass()
40 def tearDownClass(cls):
41 super(VCLTestCase, cls).tearDownClass()
45 self.build_dir = os.getenv(var, None)
46 if self.build_dir is None:
47 raise Exception("Environment variable `%s' not set" % var)
48 self.vppDebug = 'vpp_debug' in self.build_dir
49 self.server_addr = "127.0.0.1"
50 self.server_port = "22000"
51 self.server_args = [self.server_port]
52 self.server_ipv6_addr = "::1"
53 self.server_ipv6_args = ["-6", self.server_port]
55 self.echo_phrase = "Hello, world! Jenny is a friend of mine."
56 self.pre_test_sleep = 0.3
57 self.post_test_sleep = 0.2
59 if os.path.isfile("/tmp/ldp_server_af_unix_socket"):
60 os.remove("/tmp/ldp_server_af_unix_socket")
62 super(VCLTestCase, self).setUp()
64 def cut_thru_setup(self):
65 self.vapi.session_enable_disable(is_enabled=1)
67 def cut_thru_tear_down(self):
68 self.vapi.session_enable_disable(is_enabled=0)
70 def cut_thru_test(self, server_app, server_args, client_app, client_args):
71 self.env = {'VCL_API_PREFIX': self.shm_prefix,
72 'VCL_APP_SCOPE_LOCAL': "true"}
74 worker_server = VCLAppWorker(self.build_dir, server_app, server_args,
75 self.logger, self.env)
77 self.sleep(self.pre_test_sleep)
78 worker_client = VCLAppWorker(self.build_dir, client_app, client_args,
79 self.logger, self.env)
81 worker_client.join(self.timeout)
83 self.validateResults(worker_client, worker_server, self.timeout)
84 except Exception as error:
85 self.fail("Failed with %s" % error)
86 self.sleep(self.post_test_sleep)
88 def thru_host_stack_setup(self):
89 self.vapi.session_enable_disable(is_enabled=1)
90 self.create_loopback_interfaces(2)
94 for i in self.lo_interfaces:
98 tbl = VppIpTable(self, table_id)
101 i.set_table_ip4(table_id)
105 # Configure namespaces
106 self.vapi.app_namespace_add_del(namespace_id=b"1", secret=1234,
107 sw_if_index=self.loop0.sw_if_index)
108 self.vapi.app_namespace_add_del(namespace_id=b"2", secret=5678,
109 sw_if_index=self.loop1.sw_if_index)
111 # Add inter-table routes
112 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
113 [VppRoutePath("0.0.0.0",
115 nh_table_id=2)], table_id=1)
116 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
117 [VppRoutePath("0.0.0.0",
119 nh_table_id=1)], table_id=2)
120 ip_t01.add_vpp_config()
121 ip_t10.add_vpp_config()
122 self.logger.debug(self.vapi.cli("show ip fib"))
124 def thru_host_stack_tear_down(self):
125 for i in self.lo_interfaces:
130 self.vapi.session_enable_disable(is_enabled=0)
132 def thru_host_stack_ipv6_setup(self):
133 self.vapi.session_enable_disable(is_enabled=1)
134 self.create_loopback_interfaces(2)
138 for i in self.lo_interfaces:
141 tbl = VppIpTable(self, table_id, is_ip6=1)
144 i.set_table_ip6(table_id)
148 # Configure namespaces
149 self.vapi.app_namespace_add_del(namespace_id=b"1", secret=1234,
150 sw_if_index=self.loop0.sw_if_index)
151 self.vapi.app_namespace_add_del(namespace_id=b"2", secret=5678,
152 sw_if_index=self.loop1.sw_if_index)
154 # Add inter-table routes
155 ip_t01 = VppIpRoute(self, self.loop1.local_ip6, 128,
156 [VppRoutePath("::0", 0xffffffff,
158 proto=DpoProto.DPO_PROTO_IP6)],
159 table_id=1, is_ip6=1)
160 ip_t10 = VppIpRoute(self, self.loop0.local_ip6, 128,
161 [VppRoutePath("::0", 0xffffffff,
163 proto=DpoProto.DPO_PROTO_IP6)],
164 table_id=2, is_ip6=1)
165 ip_t01.add_vpp_config()
166 ip_t10.add_vpp_config()
167 self.logger.debug(self.vapi.cli("show interface addr"))
168 self.logger.debug(self.vapi.cli("show ip6 fib"))
170 def thru_host_stack_ipv6_tear_down(self):
171 for i in self.lo_interfaces:
176 self.vapi.session_enable_disable(is_enabled=0)
178 def thru_host_stack_test(self, server_app, server_args,
179 client_app, client_args):
180 self.env = {'VCL_API_PREFIX': self.shm_prefix,
181 'VCL_APP_SCOPE_GLOBAL': "true",
182 'VCL_APP_NAMESPACE_ID': "1",
183 'VCL_APP_NAMESPACE_SECRET': "1234"}
185 worker_server = VCLAppWorker(self.build_dir, server_app, server_args,
186 self.logger, self.env)
187 worker_server.start()
188 self.sleep(self.pre_test_sleep)
190 self.env.update({'VCL_APP_NAMESPACE_ID': "2",
191 'VCL_APP_NAMESPACE_SECRET': "5678"})
192 worker_client = VCLAppWorker(self.build_dir, client_app, client_args,
193 self.logger, self.env)
194 worker_client.start()
195 worker_client.join(self.timeout)
198 self.validateResults(worker_client, worker_server, self.timeout)
199 except Exception as error:
200 self.fail("Failed with %s" % error)
201 self.sleep(self.post_test_sleep)
203 def validateResults(self, worker_client, worker_server, timeout):
204 if os.path.isdir('/proc/{}'.format(worker_server.process.pid)):
205 self.logger.info("Killing server worker process (pid %d)" %
206 worker_server.process.pid)
207 os.killpg(os.getpgid(worker_server.process.pid), signal.SIGKILL)
209 self.logger.info("Client worker result is `%s'" % worker_client.result)
211 if worker_client.result is None:
215 "Timeout: %ss! Killing client worker process (pid %d)" %
216 (timeout, worker_client.process.pid))
217 os.killpg(os.getpgid(worker_client.process.pid),
222 "Couldn't kill client worker process")
226 "Timeout! Client worker did not finish in %ss" % timeout)
227 self.assert_equal(worker_client.result, 0, "Binary test return code")
230 class LDPCutThruTestCase(VCLTestCase):
231 """ LDP Cut Thru Tests """
235 super(LDPCutThruTestCase, cls).setUpClass()
238 def tearDownClass(cls):
239 super(LDPCutThruTestCase, cls).tearDownClass()
242 super(LDPCutThruTestCase, self).setUp()
244 self.cut_thru_setup()
245 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
246 self.server_addr, self.server_port]
247 self.client_iperf3_timeout = 20
248 self.client_iperf3_args = ["-V4d", "-t 2", "-c", self.server_addr]
249 self.server_iperf3_args = ["-V4d", "-s"]
250 self.client_uni_dir_nsock_timeout = 20
251 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
255 self.client_bi_dir_nsock_timeout = 20
256 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
262 self.cut_thru_tear_down()
263 super(LDPCutThruTestCase, self).tearDown()
265 def show_commands_at_teardown(self):
266 self.logger.debug(self.vapi.cli("show session verbose 2"))
268 @unittest.skipUnless(running_extended_tests, "part of extended tests")
269 def test_ldp_cut_thru_echo(self):
270 """ run LDP cut thru echo test """
272 self.cut_thru_test("sock_test_server", self.server_args,
273 "sock_test_client", self.client_echo_test_args)
275 def test_ldp_cut_thru_iperf3(self):
276 """ run LDP cut thru iperf3 test """
279 subprocess.check_output(['iperf3', '-v'])
280 except subprocess.CalledProcessError:
282 "WARNING: Subprocess returned non-0 running 'iperf3 -v")
283 self.logger.error(" 'test_ldp_cut_thru_iperf3' not run!")
287 "WARNING: Subprocess returned with OS error (%s) %s\n"
288 " 'iperf3' is likely not installed,",
290 self.logger.error(" 'test_ldp_cut_thru_iperf3' not run!")
293 self.logger.exception(
294 "Subprocess returned non-0 running 'iperf3 -v")
296 self.timeout = self.client_iperf3_timeout
297 self.cut_thru_test("iperf3", self.server_iperf3_args,
298 "iperf3", self.client_iperf3_args)
300 @unittest.skipUnless(running_extended_tests, "part of extended tests")
301 def test_ldp_cut_thru_uni_dir_nsock(self):
302 """ run LDP cut thru uni-directional (multiple sockets) test """
304 self.timeout = self.client_uni_dir_nsock_timeout
305 self.cut_thru_test("sock_test_server", self.server_args,
307 self.client_uni_dir_nsock_test_args)
309 @unittest.skipUnless(running_extended_tests, "part of extended tests")
310 def test_ldp_cut_thru_bi_dir_nsock(self):
311 """ run LDP cut thru bi-directional (multiple sockets) test """
313 self.timeout = self.client_bi_dir_nsock_timeout
314 self.cut_thru_test("sock_test_server", self.server_args,
316 self.client_bi_dir_nsock_test_args)
319 class VCLCutThruTestCase(VCLTestCase):
320 """ VCL Cut Thru Tests """
324 super(VCLCutThruTestCase, cls).setUpClass()
327 def tearDownClass(cls):
328 super(VCLCutThruTestCase, cls).tearDownClass()
331 super(VCLCutThruTestCase, self).setUp()
333 self.cut_thru_setup()
334 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
335 self.server_addr, self.server_port]
337 self.client_uni_dir_nsock_timeout = 20
338 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
342 self.client_bi_dir_nsock_timeout = 20
343 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
349 super(VCLCutThruTestCase, self).tearDown()
351 def show_commands_at_teardown(self):
352 self.logger.debug(self.vapi.cli("show session verbose 2"))
354 def test_vcl_cut_thru_echo(self):
355 """ run VCL cut thru echo test """
357 self.cut_thru_test("vcl_test_server", self.server_args,
358 "vcl_test_client", self.client_echo_test_args)
360 def test_vcl_cut_thru_uni_dir_nsock(self):
361 """ run VCL cut thru uni-directional (multiple sockets) test """
363 self.timeout = self.client_uni_dir_nsock_timeout
364 self.cut_thru_test("vcl_test_server", self.server_args,
366 self.client_uni_dir_nsock_test_args)
368 def test_vcl_cut_thru_bi_dir_nsock(self):
369 """ run VCL cut thru bi-directional (multiple sockets) test """
371 self.timeout = self.client_bi_dir_nsock_timeout
372 self.cut_thru_test("vcl_test_server", self.server_args,
374 self.client_bi_dir_nsock_test_args)
377 class VCLThruHostStackEcho(VCLTestCase):
378 """ VCL Thru Host Stack Echo """
382 super(VCLThruHostStackEcho, cls).setUpClass()
385 def tearDownClass(cls):
386 super(VCLThruHostStackEcho, cls).tearDownClass()
389 super(VCLThruHostStackEcho, self).setUp()
391 self.thru_host_stack_setup()
392 self.client_bi_dir_nsock_timeout = 20
393 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
395 self.loop0.local_ip4,
397 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
398 self.loop0.local_ip4,
402 self.thru_host_stack_tear_down()
403 super(VCLThruHostStackEcho, self).tearDown()
405 def show_commands_at_teardown(self):
406 self.logger.debug(self.vapi.cli("show app server"))
407 self.logger.debug(self.vapi.cli("show session verbose"))
410 class VCLThruHostStackTLS(VCLTestCase):
411 """ VCL Thru Host Stack TLS """
415 super(VCLThruHostStackTLS, cls).setUpClass()
418 def tearDownClass(cls):
419 super(VCLThruHostStackTLS, cls).tearDownClass()
422 super(VCLThruHostStackTLS, self).setUp()
424 self.thru_host_stack_setup()
425 self.client_uni_dir_tls_timeout = 20
426 self.server_tls_args = ["-L", self.server_port]
427 self.client_uni_dir_tls_test_args = ["-N", "1000", "-U", "-X", "-L",
428 self.loop0.local_ip4,
431 def test_vcl_thru_host_stack_tls_uni_dir(self):
432 """ run VCL thru host stack uni-directional TLS test """
434 self.timeout = self.client_uni_dir_tls_timeout
435 self.thru_host_stack_test("vcl_test_server", self.server_tls_args,
437 self.client_uni_dir_tls_test_args)
440 self.thru_host_stack_tear_down()
441 super(VCLThruHostStackTLS, self).tearDown()
443 def show_commands_at_teardown(self):
444 self.logger.debug(self.vapi.cli("show app server"))
445 self.logger.debug(self.vapi.cli("show session verbose 2"))
448 class VCLThruHostStackBidirNsock(VCLTestCase):
449 """ VCL Thru Host Stack Bidir Nsock """
453 super(VCLThruHostStackBidirNsock, cls).setUpClass()
456 def tearDownClass(cls):
457 super(VCLThruHostStackBidirNsock, cls).tearDownClass()
460 super(VCLThruHostStackBidirNsock, self).setUp()
462 self.thru_host_stack_setup()
463 self.client_bi_dir_nsock_timeout = 20
464 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
466 self.loop0.local_ip4,
468 self.client_echo_test_args = ["-E", self.echo_phrase, "-X",
469 self.loop0.local_ip4,
473 self.thru_host_stack_tear_down()
474 super(VCLThruHostStackBidirNsock, self).tearDown()
476 def show_commands_at_teardown(self):
477 self.logger.debug(self.vapi.cli("show session verbose 2"))
479 def test_vcl_thru_host_stack_bi_dir_nsock(self):
480 """ run VCL thru host stack bi-directional (multiple sockets) test """
482 self.timeout = self.client_bi_dir_nsock_timeout
483 self.thru_host_stack_test("vcl_test_server", self.server_args,
485 self.client_bi_dir_nsock_test_args)
488 class LDPThruHostStackBidirNsock(VCLTestCase):
489 """ LDP Thru Host Stack Bidir Nsock """
493 super(LDPThruHostStackBidirNsock, cls).setUpClass()
496 def tearDownClass(cls):
497 super(LDPThruHostStackBidirNsock, cls).tearDownClass()
500 super(LDPThruHostStackBidirNsock, self).setUp()
502 self.thru_host_stack_setup()
504 self.client_bi_dir_nsock_timeout = 20
505 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
506 # OUCH! Host Stack Bug?
508 self.loop0.local_ip4,
511 self.client_bi_dir_nsock_timeout = 20
512 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
513 # OUCH! Host Stack Bug?
515 self.loop0.local_ip4,
519 self.thru_host_stack_tear_down()
520 super(LDPThruHostStackBidirNsock, self).tearDown()
522 def show_commands_at_teardown(self):
523 self.logger.debug(self.vapi.cli("show session verbose 2"))
525 def test_ldp_thru_host_stack_bi_dir_nsock(self):
526 """ run LDP thru host stack bi-directional (multiple sockets) test """
528 self.timeout = self.client_bi_dir_nsock_timeout
529 self.thru_host_stack_test("sock_test_server", self.server_args,
531 self.client_bi_dir_nsock_test_args)
534 class LDPThruHostStackNsock(VCLTestCase):
535 """ LDP Thru Host Stack Nsock """
539 super(LDPThruHostStackNsock, cls).setUpClass()
542 def tearDownClass(cls):
543 super(LDPThruHostStackNsock, cls).tearDownClass()
546 super(LDPThruHostStackNsock, self).setUp()
548 self.thru_host_stack_setup()
550 self.client_uni_dir_nsock_timeout = 20
551 self.numSockets = "2"
553 self.client_uni_dir_nsock_timeout = 20
554 self.numSockets = "5"
556 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
557 "-I", self.numSockets,
558 self.loop0.local_ip4,
562 self.thru_host_stack_tear_down()
563 super(LDPThruHostStackNsock, self).tearDown()
565 def test_ldp_thru_host_stack_uni_dir_nsock(self):
566 """ run LDP thru host stack uni-directional (multiple sockets) test """
568 self.timeout = self.client_uni_dir_nsock_timeout
569 self.thru_host_stack_test("sock_test_server", self.server_args,
571 self.client_uni_dir_nsock_test_args)
574 class VCLThruHostStackNsock(VCLTestCase):
575 """ VCL Thru Host Stack Nsock """
579 super(VCLThruHostStackNsock, cls).setUpClass()
582 def tearDownClass(cls):
583 super(VCLThruHostStackNsock, cls).tearDownClass()
586 super(VCLThruHostStackNsock, self).setUp()
588 self.thru_host_stack_setup()
590 self.client_uni_dir_nsock_timeout = 20
591 self.numSockets = "2"
593 self.client_uni_dir_nsock_timeout = 20
594 self.numSockets = "5"
596 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
597 "-I", self.numSockets,
598 self.loop0.local_ip4,
602 self.thru_host_stack_tear_down()
603 super(VCLThruHostStackNsock, self).tearDown()
605 def test_vcl_thru_host_stack_uni_dir_nsock(self):
606 """ run VCL thru host stack uni-directional (multiple sockets) test """
608 self.timeout = self.client_uni_dir_nsock_timeout
609 self.thru_host_stack_test("vcl_test_server", self.server_args,
611 self.client_uni_dir_nsock_test_args)
614 class LDPThruHostStackIperf(VCLTestCase):
615 """ LDP Thru Host Stack Iperf """
619 super(LDPThruHostStackIperf, cls).setUpClass()
622 def tearDownClass(cls):
623 super(LDPThruHostStackIperf, cls).tearDownClass()
626 super(LDPThruHostStackIperf, self).setUp()
628 self.thru_host_stack_setup()
629 self.client_iperf3_timeout = 20
630 self.client_iperf3_args = ["-V4d", "-t 2", "-c", self.loop0.local_ip4]
631 self.server_iperf3_args = ["-V4d", "-s"]
634 self.thru_host_stack_tear_down()
635 super(LDPThruHostStackIperf, self).tearDown()
637 def show_commands_at_teardown(self):
638 self.logger.debug(self.vapi.cli("show session verbose 2"))
640 def test_ldp_thru_host_stack_iperf3(self):
641 """ run LDP thru host stack iperf3 test """
644 subprocess.check_output(['iperf3', '-v'])
645 except subprocess.CalledProcessError:
646 self.logger.error("WARNING: 'iperf3' is not installed,")
648 " 'test_ldp_thru_host_stack_iperf3' not run!")
651 self.logger.error("WARNING: 'iperf3' is not installed,")
652 self.logger.error(" 'test' not run!")
654 except Exception as e:
655 self.logger.error("WARNING: 'iperf3' unexpected error,")
656 self.logger.error(" 'test' not run!")
659 self.timeout = self.client_iperf3_timeout
660 self.thru_host_stack_test("iperf3", self.server_iperf3_args,
661 "iperf3", self.client_iperf3_args)
664 class LDPIpv6CutThruTestCase(VCLTestCase):
665 """ LDP IPv6 Cut Thru Tests """
669 super(LDPIpv6CutThruTestCase, cls).setUpClass()
672 def tearDownClass(cls):
673 super(LDPIpv6CutThruTestCase, cls).tearDownClass()
676 super(LDPIpv6CutThruTestCase, self).setUp()
678 self.cut_thru_setup()
679 self.client_iperf3_timeout = 20
680 self.client_uni_dir_nsock_timeout = 20
681 self.client_bi_dir_nsock_timeout = 20
682 self.client_ipv6_echo_test_args = ["-6", "-E", self.echo_phrase, "-X",
683 self.server_ipv6_addr,
685 self.client_ipv6_iperf3_args = ["-V6d", "-t 2", "-c",
686 self.server_ipv6_addr]
687 self.server_ipv6_iperf3_args = ["-V6d", "-s"]
688 self.client_ipv6_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
691 self.server_ipv6_addr,
693 self.client_ipv6_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
696 self.server_ipv6_addr,
700 self.cut_thru_tear_down()
702 super(LDPIpv6CutThruTestCase, self).tearDown()
704 def test_ldp_ipv6_cut_thru_echo(self):
705 """ run LDP IPv6 cut thru echo test """
707 self.cut_thru_test("sock_test_server",
708 self.server_ipv6_args,
710 self.client_ipv6_echo_test_args)
712 @unittest.skipUnless(running_extended_tests, "part of extended tests")
713 def test_ldp_ipv6_cut_thru_iperf3(self):
714 """ run LDP IPv6 cut thru iperf3 test """
717 subprocess.check_output(['iperf3', '-v'])
719 self.logger.error("WARNING: 'iperf3' is not installed,")
721 " 'test_ldp_ipv6_cut_thru_iperf3' not run!")
724 self.timeout = self.client_iperf3_timeout
725 self.cut_thru_test("iperf3", self.server_ipv6_iperf3_args,
726 "iperf3", self.client_ipv6_iperf3_args)
728 @unittest.skipUnless(running_extended_tests, "part of extended tests")
729 def test_ldp_ipv6_cut_thru_uni_dir_nsock(self):
730 """ run LDP IPv6 cut thru uni-directional (multiple sockets) test """
732 self.timeout = self.client_uni_dir_nsock_timeout
733 self.cut_thru_test("sock_test_server", self.server_ipv6_args,
735 self.client_ipv6_uni_dir_nsock_test_args)
737 @unittest.skipUnless(running_extended_tests, "part of extended tests")
738 def test_ldp_ipv6_cut_thru_bi_dir_nsock(self):
739 """ run LDP IPv6 cut thru bi-directional (multiple sockets) test """
741 self.timeout = self.client_bi_dir_nsock_timeout
742 self.cut_thru_test("sock_test_server", self.server_ipv6_args,
744 self.client_ipv6_bi_dir_nsock_test_args)
747 class VCLIpv6CutThruTestCase(VCLTestCase):
748 """ VCL IPv6 Cut Thru Tests """
752 super(VCLIpv6CutThruTestCase, cls).setUpClass()
755 def tearDownClass(cls):
756 super(VCLIpv6CutThruTestCase, cls).tearDownClass()
759 super(VCLIpv6CutThruTestCase, self).setUp()
761 self.cut_thru_setup()
762 self.client_uni_dir_nsock_timeout = 20
763 self.client_bi_dir_nsock_timeout = 20
764 self.client_ipv6_echo_test_args = ["-6", "-E", self.echo_phrase, "-X",
765 self.server_ipv6_addr,
767 self.client_ipv6_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X",
770 self.server_ipv6_addr,
772 self.client_ipv6_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X",
775 self.server_ipv6_addr,
779 self.cut_thru_tear_down()
781 super(VCLIpv6CutThruTestCase, self).tearDown()
783 def test_vcl_ipv6_cut_thru_echo(self):
784 """ run VCL IPv6 cut thru echo test """
786 self.cut_thru_test("vcl_test_server",
787 self.server_ipv6_args,
789 self.client_ipv6_echo_test_args)
791 @unittest.skipUnless(running_extended_tests, "part of extended tests")
792 def test_vcl_ipv6_cut_thru_uni_dir_nsock(self):
793 """ run VCL IPv6 cut thru uni-directional (multiple sockets) test """
795 self.timeout = self.client_uni_dir_nsock_timeout
796 self.cut_thru_test("vcl_test_server", self.server_ipv6_args,
798 self.client_ipv6_uni_dir_nsock_test_args)
800 @unittest.skipUnless(running_extended_tests, "part of extended tests")
801 def test_vcl_ipv6_cut_thru_bi_dir_nsock(self):
802 """ run VCL IPv6 cut thru bi-directional (multiple sockets) test """
804 self.timeout = self.client_bi_dir_nsock_timeout
805 self.cut_thru_test("vcl_test_server", self.server_ipv6_args,
807 self.client_ipv6_bi_dir_nsock_test_args)
810 class VCLIpv6ThruHostStackEcho(VCLTestCase):
811 """ VCL IPv6 Thru Host Stack Echo """
815 super(VCLIpv6ThruHostStackEcho, cls).setUpClass()
818 def tearDownClass(cls):
819 super(VCLIpv6ThruHostStackEcho, cls).tearDownClass()
822 super(VCLIpv6ThruHostStackEcho, self).setUp()
824 self.thru_host_stack_ipv6_setup()
825 self.client_ipv6_echo_test_args = ["-6", "-E", self.echo_phrase, "-X",
826 self.loop0.local_ip6,
830 self.thru_host_stack_ipv6_tear_down()
831 super(VCLIpv6ThruHostStackEcho, self).tearDown()
833 def test_vcl_ipv6_thru_host_stack_echo(self):
834 """ run VCL IPv6 thru host stack echo test """
836 self.thru_host_stack_test("vcl_test_server",
837 self.server_ipv6_args,
839 self.client_ipv6_echo_test_args)
842 if __name__ == '__main__':
843 unittest.main(testRunner=VppTestRunner)