8 from threading import Thread
9 from log import single_line_delim
10 from framework import VppTestCase, running_extended_tests, \
11 running_on_centos, VppTestRunner, Worker
14 class VCLTestCase(VppTestCase):
15 """ VPP Communications Library Test """
17 server_addr = "127.0.0.1"
22 super(VCLTestCase, cls).setUpClass()
24 cls.vapi.session_enable_disable(is_enabled=1)
27 super(VCLTestCase, self).setUp()
29 def test_vcl_cutthru(self):
30 """ run VCL cut-thru test """
32 var = "VPP_TEST_BUILD_DIR"
33 build_dir = os.getenv(var, None)
34 self.assertIsNotNone(build_dir,
35 "Environment variable `%s' not set" % var)
36 vcl_exe_dir = "%s/vpp/.libs" % build_dir
37 executable = "%s/vcl_test_server" % vcl_exe_dir
38 worker_server = Worker([executable, self.server_port], self.logger)
39 worker_server.env["VCL_API_PREFIX"] = self.shm_prefix
40 # worker_server.env["VCL_DEBUG"] = "2"
41 worker_server.env["VCL_APP_SCOPE_LOCAL"] = "true"
43 executable = "%s/vcl_test_client" % vcl_exe_dir
44 worker_client = Worker(
45 [executable, self.server_addr, self.server_port,
46 "-E", "Hello, world!", "-X"], self.logger)
47 worker_client.env["VCL_API_PREFIX"] = self.shm_prefix
48 # worker_client.env["VCL_DEBUG"] = "2"
49 worker_client.env["VCL_APP_SCOPE_LOCAL"] = "true"
51 worker_client.join(timeout)
52 self.logger.info("Client worker result is `%s'" % worker_client.result)
54 if worker_client.result is None:
58 "Timeout! Client worker did not finish in %ss" % timeout)
59 os.killpg(os.getpgid(worker_client.process.pid),
63 raise Exception("Couldn't kill client worker-spawned process")
65 os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM)
68 "Timeout! Client worker did not finish in %ss" % timeout)
69 self.assert_equal(worker_client.result, 0, "Binary test return code")
72 if __name__ == '__main__':
73 unittest.main(testRunner=VppTestRunner)