diff options
Diffstat (limited to 'test/test_vcl.py')
| -rw-r--r-- | test/test_vcl.py | 165 |
1 files changed, 78 insertions, 87 deletions
diff --git a/test/test_vcl.py b/test/test_vcl.py index a68d0275968..a8a00f40f61 100644 --- a/test/test_vcl.py +++ b/test/test_vcl.py @@ -4,55 +4,27 @@ import unittest import os import signal -import subprocess -from threading import Thread -from log import single_line_delim -from framework import VppTestCase, running_extended_tests, \ - running_on_centos, VppTestRunner, Worker +from framework import VppTestCase, VppTestRunner, Worker from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath -class VCLCUTTHRUTestCase(VppTestCase): - """ VCL Cut Thru Test """ +class VclAppWorker(Worker): + """ VCL Test Application Worker """ - server_addr = "127.0.0.1" - server_port = "22000" - - @classmethod - def setUpClass(cls): - super(VCLCUTTHRUTestCase, cls).setUpClass() - - def setUp(self): - super(VCLCUTTHRUTestCase, self).setUp() - - self.vapi.session_enable_disable(is_enabled=1) + def __init__(self, appname, args, logger, env={}): + var = "VPP_TEST_BUILD_DIR" + build_dir = os.getenv(var, None) + if build_dir is None: + raise Exception("Environment variable `%s' not set" % var) + vcl_app_dir = "%s/vpp/.libs" % build_dir + self.args = ["%s/%s" % (vcl_app_dir, appname)] + args + super(VclAppWorker, self).__init__(self.args, logger, env) - def tearDown(self): - self.vapi.session_enable_disable(is_enabled=0) - super(VCLCUTTHRUTestCase, self).tearDown() +class VclTestCase(VppTestCase): + """ VCL Test Class """ - def test_vcl_cutthru(self): - """ run VCL cut-thru test """ - timeout = 5 - var = "VPP_TEST_BUILD_DIR" - build_dir = os.getenv(var, None) - self.assertIsNotNone(build_dir, - "Environment variable `%s' not set" % var) - vcl_exe_dir = "%s/vpp/.libs" % build_dir - executable = "%s/vcl_test_server" % vcl_exe_dir - worker_server = Worker([executable, self.server_port], self.logger) - worker_server.env["VCL_API_PREFIX"] = self.shm_prefix - worker_server.env["VCL_APP_SCOPE_LOCAL"] = "true" - worker_server.start() - executable = "%s/vcl_test_client" % vcl_exe_dir - worker_client = Worker( - [executable, self.server_addr, self.server_port, - "-E", "Hello, world!", "-X"], self.logger) - worker_client.env["VCL_API_PREFIX"] = self.shm_prefix - worker_client.env["VCL_APP_SCOPE_LOCAL"] = "true" - worker_client.start() - worker_client.join(timeout) + def validateResults(self, worker_client, worker_server, timeout): self.logger.info("Client worker result is `%s'" % worker_client.result) error = False if worker_client.result is None: @@ -64,7 +36,9 @@ class VCLCUTTHRUTestCase(VppTestCase): signal.SIGTERM) worker_client.join() except: - raise Exception("Couldn't kill client worker-spawned process") + self.logger.debug( + "Couldn't kill client worker-spawned process") + raise if error: os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM) worker_server.join() @@ -73,14 +47,49 @@ class VCLCUTTHRUTestCase(VppTestCase): self.assert_equal(worker_client.result, 0, "Binary test return code") -class VCLTHRUHSTestcase(VppTestCase): - """ VCL Thru Hoststack Test """ +class VCLCUTTHRUTestCase(VclTestCase): + """ VPP Communications Library Test """ + server_addr = "127.0.0.1" server_port = "22000" + timeout = 3 + echo_phrase = "Hello, world! Jenny is a friend of mine" + + def setUp(self): + super(VCLCUTTHRUTestCase, self).setUp() + + self.vapi.session_enable_disable(is_enabled=1) + + def tearDown(self): + self.vapi.session_enable_disable(is_enabled=0) + + super(VCLCUTTHRUTestCase, self).tearDown() - @classmethod - def setUpClass(cls): - super(VCLTHRUHSTestcase, cls).setUpClass() + def test_vcl_cutthru(self): + """ run VCL cut-thru test """ + self.env = {'VCL_API_PREFIX': self.shm_prefix, + 'VCL_APP_SCOPE_LOCAL': "true"} + + worker_server = VclAppWorker("vcl_test_server", + [self.server_port], + self.logger, self.env) + worker_server.start() + self.sleep(0.2) + worker_client = VclAppWorker("vcl_test_client", + [self.server_addr, self.server_port, + "-E", self.echo_phrase, "-X"], + self.logger, self.env) + worker_client.start() + worker_client.join(self.timeout) + self.validateResults(worker_client, worker_server, self.timeout) + + +class VCLTHRUHSTestcase(VclTestCase): + """ VCL Thru Hoststack Test """ + + server_port = "22000" + timeout = 3 + echo_phrase = "Hello, world! Jenny is a friend of mine" def setUp(self): super(VCLTHRUHSTestcase, self).setUp() @@ -102,9 +111,9 @@ class VCLTHRUHSTestcase(VppTestCase): table_id += 1 # Configure namespaces - self.vapi.app_namespace_add(namespace_id="0", + self.vapi.app_namespace_add(namespace_id="0", secret=1234, sw_if_index=self.loop0.sw_if_index) - self.vapi.app_namespace_add(namespace_id="1", + self.vapi.app_namespace_add(namespace_id="1", secret=5678, sw_if_index=self.loop1.sw_if_index) def tearDown(self): @@ -118,6 +127,9 @@ class VCLTHRUHSTestcase(VppTestCase): def test_vcl_thru_hoststack(self): """ run VCL thru hoststack test """ + self.env = {'VCL_API_PREFIX': self.shm_prefix, + 'VCL_APP_SCOPE_GLOBAL': "true"} + # Add inter-table routes ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32, [VppRoutePath("0.0.0.0", @@ -130,45 +142,24 @@ class VCLTHRUHSTestcase(VppTestCase): ip_t01.add_vpp_config() ip_t10.add_vpp_config() - timeout = 20 - var = "VPP_TEST_BUILD_DIR" - build_dir = os.getenv(var, None) - self.assertIsNotNone(build_dir, - "Environment variable `%s' not set" % var) - vcl_exe_dir = "%s/vpp/.libs" % build_dir - executable = "%s/vcl_test_server" % vcl_exe_dir - worker_server = Worker([executable, self.server_port], self.logger) - worker_server.env["VCL_API_PREFIX"] = self.shm_prefix - worker_server.env["VCL_APP_SCOPE_GLOBAL"] = "true" - worker_server.env["VCL_APP_NAMESPACE_ID"] = "0" + self.env.update({'VCL_APP_NAMESPACE_ID': "0", + 'VCL_APP_NAMESPACE_SECRET': "1234"}) + worker_server = VclAppWorker("vcl_test_server", + [self.server_port], + self.logger, self.env) worker_server.start() - executable = "%s/vcl_test_client" % vcl_exe_dir - worker_client = Worker( - [executable, self.loop0.local_ip4, self.server_port, - "-E", "Hello, world!", "-X"], self.logger) - worker_client.env["VCL_API_PREFIX"] = self.shm_prefix - worker_client.env["VCL_APP_SCOPE_GLOBAL"] = "true" - worker_client.env["VCL_APP_NAMESPACE_ID"] = "1" + self.sleep(0.2) + + self.env.update({'VCL_APP_NAMESPACE_ID': "1", + 'VCL_APP_NAMESPACE_SECRET': "5678"}) + worker_client = VclAppWorker("vcl_test_client", + [self.loop0.local_ip4, self.server_port, + "-E", self.echo_phrase, "-X"], + self.logger, self.env) worker_client.start() - worker_client.join(timeout) - self.logger.info("Client worker result is `%s'" % worker_client.result) - error = False - if worker_client.result is None: - try: - error = True - self.logger.error( - "Timeout! Client worker did not finish in %ss" % timeout) - os.killpg(os.getpgid(worker_client.process.pid), - signal.SIGTERM) - worker_client.join() - except: - raise Exception("Couldn't kill client worker-spawned process") - if error: - os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM) - worker_server.join() - raise Exception( - "Timeout! Client worker did not finish in %ss" % timeout) - self.assert_equal(worker_client.result, 0, "Binary test return code") + worker_client.join(self.timeout) + + self.validateResults(worker_client, worker_server, self.timeout) if __name__ == '__main__': unittest.main(testRunner=VppTestRunner) |