7 from framework import VppTestCase, VppTestRunner, Worker
8 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
11 class VclAppWorker(Worker):
12 """ VCL Test Application Worker """
14 def __init__(self, appname, args, logger, env={}):
15 var = "VPP_TEST_BUILD_DIR"
16 build_dir = os.getenv(var, None)
18 raise Exception("Environment variable `%s' not set" % var)
19 vcl_lib_dir = "%s/vpp/.libs" % build_dir
20 app = "%s/%s" % (vcl_lib_dir, appname)
21 if not os.path.isfile(app):
22 app = "%s/vpp/%s" % (build_dir, appname)
23 env.update({'LD_PRELOAD':
24 "%s/libvcl_ldpreload.so.0.0.0" % vcl_lib_dir})
25 self.args = [app] + args
26 super(VclAppWorker, self).__init__(self.args, logger, env)
29 class VclTestCase(VppTestCase):
30 """ VCL Test Class """
32 def __init__(self, methodName):
33 self.server_addr = "127.0.0.1"
34 self.server_port = "22000"
36 self.echo_phrase = "Hello, world! Jenny is a friend of mine."
38 super(VclTestCase, self).__init__(methodName)
40 def cut_thru_setup(self):
41 self.vapi.session_enable_disable(is_enabled=1)
43 def cut_thru_tear_down(self):
44 self.vapi.session_enable_disable(is_enabled=0)
46 def cut_thru_test(self, server_app, client_app, client_args):
47 self.env = {'VCL_API_PREFIX': self.shm_prefix,
48 'VCL_APP_SCOPE_LOCAL': "true"}
50 worker_server = VclAppWorker(server_app, [self.server_port],
51 self.logger, self.env)
54 worker_client = VclAppWorker(client_app, client_args,
55 self.logger, self.env)
57 worker_client.join(self.timeout)
58 self.validateResults(worker_client, worker_server, self.timeout)
60 def thru_host_stack_setup(self):
61 self.vapi.session_enable_disable(is_enabled=1)
62 self.create_loopback_interfaces(range(2))
66 for i in self.lo_interfaces:
70 tbl = VppIpTable(self, table_id)
73 i.set_table_ip4(table_id)
77 # Configure namespaces
78 self.vapi.app_namespace_add(namespace_id="0", secret=1234,
79 sw_if_index=self.loop0.sw_if_index)
80 self.vapi.app_namespace_add(namespace_id="1", secret=5678,
81 sw_if_index=self.loop1.sw_if_index)
83 # Add inter-table routes
84 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
85 [VppRoutePath("0.0.0.0",
88 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
89 [VppRoutePath("0.0.0.0",
91 nh_table_id=0)], table_id=1)
92 ip_t01.add_vpp_config()
93 ip_t10.add_vpp_config()
95 def thru_host_stack_tear_down(self):
96 for i in self.lo_interfaces:
101 self.vapi.session_enable_disable(is_enabled=0)
103 def thru_host_stack_test(self, server_app, client_app, client_args):
104 self.env = {'VCL_API_PREFIX': self.shm_prefix,
105 'VCL_APP_SCOPE_GLOBAL': "true",
106 'VCL_APP_NAMESPACE_ID': "0",
107 'VCL_APP_NAMESPACE_SECRET': "1234"}
109 worker_server = VclAppWorker(server_app, [self.server_port],
110 self.logger, self.env)
111 worker_server.start()
114 self.env.update({'VCL_APP_NAMESPACE_ID': "1",
115 'VCL_APP_NAMESPACE_SECRET': "5678"})
116 worker_client = VclAppWorker(client_app, client_args,
117 self.logger, self.env)
118 worker_client.start()
119 worker_client.join(self.timeout)
121 self.validateResults(worker_client, worker_server, self.timeout)
123 def validateResults(self, worker_client, worker_server, timeout):
124 self.logger.info("Client worker result is `%s'" % worker_client.result)
126 if worker_client.result is None:
130 "Timeout (%ss)! Killing client worker process" % timeout)
131 os.killpg(os.getpgid(worker_client.process.pid),
136 "Couldn't kill client worker process")
139 os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM)
142 "Timeout! Client worker did not finish in %ss" % timeout)
143 self.assert_equal(worker_client.result, 0, "Binary test return code")
146 class VCLCutThruTestCase(VclTestCase):
147 """ VCL Cut Thru Tests """
150 super(VCLCutThruTestCase, self).setUp()
152 self.cut_thru_setup()
153 self.client_echo_test_args = [self.server_addr, self.server_port,
154 "-E", self.echo_phrase, "-X"]
157 self.cut_thru_tear_down()
159 super(VCLCutThruTestCase, self).tearDown()
161 def test_ldp_cut_thru_echo(self):
162 """ run LDP cut thru echo test """
164 self.cut_thru_test("sock_test_server", "sock_test_client",
165 self.client_echo_test_args)
167 def test_vcl_cut_thru_echo(self):
168 """ run VCL cut thru echo test """
170 self.cut_thru_test("vcl_test_server", "vcl_test_client",
171 self.client_echo_test_args)
174 class VCLThruHostStackTestCase(VclTestCase):
175 """ VCL Thru Host Stack Tests """
178 super(VCLThruHostStackTestCase, self).setUp()
180 self.thru_host_stack_setup()
181 self.client_echo_test_args = [self.loop0.local_ip4, self.server_port,
182 "-E", self.echo_phrase, "-X"]
185 self.thru_host_stack_tear_down()
187 super(VCLThruHostStackTestCase, self).tearDown()
189 def test_ldp_thru_host_stack_echo(self):
190 """ run LDP thru host stack echo test """
192 self.thru_host_stack_test("sock_test_server", "sock_test_client",
193 self.client_echo_test_args)
194 # TBD: Remove this when VPP crash is fixed.
195 self.thru_host_stack_test("vcl_test_server", "vcl_test_client",
196 self.client_echo_test_args)
198 def test_vcl_thru_host_stack_echo(self):
199 """ run VCL thru host stack echo test """
201 # TBD: Enable this when VPP crash is fixed.
202 # self.thru_host_stack_test("vcl_test_server", "vcl_test_client",
203 # self.client_echo_test_args)
205 if __name__ == '__main__':
206 unittest.main(testRunner=VppTestRunner)