3 # NEPI, a framework to manage network experiments
4 # Copyright (C) 2013 INRIA
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
21 from nepi.execution.ec import ExperimentController
22 from nepi.execution.resource import ResourceState, ResourceAction
23 from nepi.execution.trace import TraceAttr
25 from test_utils import skipIfNotAlive, skipInteractive
32 class LinuxApplicationTestCase(unittest.TestCase):
34 self.fedora_host = "nepi2.pl.sophia.inria.fr"
35 self.fedora_user = "inria_nepi"
37 self.ubuntu_host = "roseval.pl.sophia.inria.fr"
38 self.ubuntu_user = "alina"
40 self.target = "nepi5.pl.sophia.inria.fr"
43 def t_stdout(self, host, user):
45 ec = ExperimentController(exp_id = "test-stdout")
47 node = ec.register_resource("LinuxNode")
48 ec.set(node, "hostname", host)
49 ec.set(node, "username", user)
50 ec.set(node, "cleanHome", True)
51 ec.set(node, "cleanProcesses", True)
53 app = ec.register_resource("LinuxApplication")
55 ec.set(app, "command", cmd)
56 ec.register_connection(app, node)
62 self.assertTrue(ec.state(node) == ResourceState.STARTED)
63 self.assertTrue(ec.state(app) == ResourceState.STOPPED)
65 stdout = ec.trace(app, "stdout")
66 self.assertTrue(stdout.strip() == "HOLA")
71 def t_ping(self, host, user):
73 ec = ExperimentController(exp_id = "test-ping")
75 node = ec.register_resource("LinuxNode")
76 ec.set(node, "hostname", host)
77 ec.set(node, "username", user)
78 ec.set(node, "cleanHome", True)
79 ec.set(node, "cleanProcesses", True)
81 app = ec.register_resource("LinuxApplication")
82 cmd = "ping -c5 %s" % self.target
83 ec.set(app, "command", cmd)
85 ec.register_connection(app, node)
91 self.assertTrue(ec.state(node) == ResourceState.STARTED)
92 self.assertTrue(ec.state(app) == ResourceState.STOPPED)
94 stdout = ec.trace(app, "stdout")
95 size = ec.trace(app, "stdout", attr = TraceAttr.SIZE)
96 self.assertEquals(len(stdout), size)
98 block = ec.trace(app, "stdout", attr = TraceAttr.STREAM, block = 5, offset = 1)
99 self.assertEquals(block, stdout[5:10])
101 path = ec.trace(app, "stdout", attr = TraceAttr.PATH)
102 rm = ec.get_resource(app)
103 p = os.path.join(rm.run_home, "stdout")
104 self.assertEquals(path, p)
109 def t_code(self, host, user):
111 ec = ExperimentController(exp_id = "tests-code")
113 node = ec.register_resource("LinuxNode")
114 ec.set(node, "hostname", host)
115 ec.set(node, "username", user)
116 ec.set(node, "cleanHome", True)
117 ec.set(node, "cleanProcesses", True)
119 prog = """#include <stdio.h>
124 printf ("Hello, world!\\n");
128 cmd = "${RUN_HOME}/hello"
129 build = "gcc -Wall -x c ${APP_HOME}/code -o hello"
131 app = ec.register_resource("LinuxApplication")
132 ec.set(app, "command", cmd)
133 ec.set(app, "code", prog)
134 ec.set(app, "depends", "gcc")
135 ec.set(app, "build", build)
136 ec.register_connection(app, node)
140 ec.wait_finished(app)
142 out = ec.trace(app, 'stdout')
143 self.assertEquals(out, "Hello, world!\n")
148 def t_concurrency(self, host, user):
150 ec = ExperimentController(exp_id="test-concurrency")
152 node = ec.register_resource("LinuxNode")
153 ec.set(node, "hostname", host)
154 ec.set(node, "username", user)
155 ec.set(node, "cleanHome", True)
156 ec.set(node, "cleanProcesses", True)
160 app = ec.register_resource("LinuxApplication")
161 cmd = "ping -c5 %s" % self.target
162 ec.set(app, "command", cmd)
163 ec.register_connection(app, node)
168 ec.wait_finished(apps)
170 self.assertTrue(ec.state(node) == ResourceState.STARTED)
172 all([ec.state(guid) == ResourceState.STOPPED \
177 stdout = ec.trace(app, 'stdout')
178 size = ec.trace(app, 'stdout', attr = TraceAttr.SIZE)
179 self.assertEquals(len(stdout), size)
181 block = ec.trace(app, 'stdout', attr = TraceAttr.STREAM, block = 5, offset = 1)
182 self.assertEquals(block, stdout[5:10])
184 path = ec.trace(app, 'stdout', attr = TraceAttr.PATH)
185 rm = ec.get_resource(app)
186 p = os.path.join(rm.run_home, 'stdout')
187 self.assertEquals(path, p)
192 def t_condition(self, host, user, depends):
194 ec = ExperimentController(exp_id="test-condition")
196 node = ec.register_resource("LinuxNode")
197 ec.set(node, "hostname", host)
198 ec.set(node, "username", user)
199 ec.set(node, "cleanHome", True)
200 ec.set(node, "cleanProcesses", True)
202 server = ec.register_resource("LinuxApplication")
203 cmd = "echo 'HOLA' | nc -l 3333"
204 ec.set(server, "command", cmd)
205 ec.set(server, "depends", depends)
206 ec.register_connection(server, node)
208 client = ec.register_resource("LinuxApplication")
209 cmd = "nc 127.0.0.1 3333"
210 ec.set(client, "command", cmd)
211 ec.register_connection(client, node)
213 ec.register_condition(client, ResourceAction.START, server, ResourceState.STARTED)
215 apps = [client, server]
219 ec.wait_finished(apps)
221 self.assertTrue(ec.state(node) == ResourceState.STARTED)
222 self.assertTrue(ec.state(server) == ResourceState.STOPPED)
223 self.assertTrue(ec.state(client) == ResourceState.STOPPED)
225 stdout = ec.trace(client, "stdout")
226 self.assertTrue(stdout.strip() == "HOLA")
231 def t_http_sources(self, host, user):
233 ec = ExperimentController(exp_id="test-http-sources")
235 node = ec.register_resource("LinuxNode")
236 ec.set(node, "hostname", host)
237 ec.set(node, "username", user)
238 ec.set(node, "cleanHome", True)
239 ec.set(node, "cleanProcesses", True)
241 sources = "http://yans.pl.sophia.inria.fr/code/nef/archive/tip.tar.gz " \
242 " http://yans.pl.sophia.inria.fr/code/nef/raw-file/8ace577d4079/src/nef/images/menu/connect.png"
244 app = ec.register_resource("LinuxApplication")
245 ec.set(app, "sources", sources)
247 ec.register_connection(app, node)
251 ec.wait_finished([app])
253 self.assertTrue(ec.state(node) == ResourceState.STARTED)
254 self.assertTrue(ec.state(app) == ResourceState.STOPPED)
256 exitcode = ec.trace(app, "deploy_exitcode")
257 self.assertTrue(exitcode.strip() == "0")
259 out = ec.trace(app, "deploy_stdout")
260 self.assertTrue(out.find("tip.tar.gz") > -1)
261 self.assertTrue(out.find("connect.png") > -1)
266 def t_xterm(self, host, user):
268 ec = ExperimentController(exp_id="test-xterm")
270 node = ec.register_resource("LinuxNode")
271 ec.set(node, "hostname", host)
272 ec.set(node, "username", user)
273 ec.set(node, "cleanHome", True)
274 ec.set(node, "cleanProcesses", True)
276 app = ec.register_resource("LinuxApplication")
277 ec.set(app, "command", "xterm")
278 ec.set(app, "depends", "xterm")
279 ec.set(app, "forwardX11", True)
281 ec.register_connection(app, node)
285 ec.wait_finished([app])
287 self.assertTrue(ec.state(app) == ResourceState.STOPPED)
291 def test_stdout_fedora(self):
292 self.t_stdout(self.fedora_host, self.fedora_user)
294 def test_stdout_ubuntu(self):
295 self.t_stdout(self.ubuntu_host, self.ubuntu_user)
297 def test_ping_fedora(self):
298 self.t_ping(self.fedora_host, self.fedora_user)
300 def test_ping_ubuntu(self):
301 self.t_ping(self.ubuntu_host, self.ubuntu_user)
303 def test_concurrency_fedora(self):
304 self.t_concurrency(self.fedora_host, self.fedora_user)
306 def test_concurrency_ubuntu(self):
307 self.t_concurrency(self.ubuntu_host, self.ubuntu_user)
309 def test_condition_fedora(self):
310 self.t_condition(self.fedora_host, self.fedora_user, "nc")
312 def test_condition_ubuntu(self):
313 self.t_condition(self.ubuntu_host, self.ubuntu_user, "netcat")
315 def test_http_sources_fedora(self):
316 self.t_http_sources(self.fedora_host, self.fedora_user)
318 def test_http_sources_ubuntu(self):
319 self.t_http_sources(self.ubuntu_host, self.ubuntu_user)
321 def test_code_fedora(self):
322 self.t_code(self.fedora_host, self.fedora_user)
324 def test_code_ubuntu(self):
325 self.t_code(self.ubuntu_host, self.ubuntu_user)
328 def test_xterm_ubuntu(self):
329 """ Interactive test. Should not run automatically """
330 self.t_xterm(self.ubuntu_host, self.ubuntu_user)
333 if __name__ == '__main__':