2 # -*- coding: utf-8 -*-
5 from nepi.core.design import ExperimentDescription, FactoriesProvider
6 from nepi.core.execute import ExperimentController
7 from nepi.util import proxy
16 class PlanetLabIntegrationTestCase(unittest.TestCase):
17 testbed_id = "planetlab"
18 testbed_version = "01"
19 slicename = "inria_nepi"
20 plchost = "nepiplc.pl.sophia.inria.fr"
22 host1 = "nepi1.pl.sophia.inria.fr"
23 host2 = "nepi2.pl.sophia.inria.fr"
26 self.root_dir = tempfile.mkdtemp()
30 shutil.rmtree(self.root_dir)
34 shutil.rmtree(self.root_dir)
36 def make_experiment_desc(self):
37 testbed_id = self.testbed_id
38 testbed_version = self.testbed_version
39 slicename = self.slicename
40 plchost = self.plchost
41 pl_ssh_key = os.environ.get(
43 "%s/.ssh/id_rsa_planetlab" % (os.environ['HOME'],) )
44 pl_user, pl_pwd = test_util.pl_auth()
46 exp_desc = ExperimentDescription()
47 pl_provider = FactoriesProvider(testbed_id, testbed_version)
48 pl_desc = exp_desc.add_testbed_description(pl_provider)
49 pl_desc.set_attribute_value("homeDirectory", self.root_dir)
50 pl_desc.set_attribute_value("slice", slicename)
51 pl_desc.set_attribute_value("sliceSSHKey", pl_ssh_key)
52 pl_desc.set_attribute_value("authUser", pl_user)
53 pl_desc.set_attribute_value("authPass", pl_pwd)
54 pl_desc.set_attribute_value("plcHost", plchost)
56 return pl_desc, exp_desc
58 @test_util.skipUnless(test_util.pl_auth() is not None, "Test requires PlanetLab authentication info (PL_USER and PL_PASS environment variables)")
59 def test_simple(self):
60 pl, exp = self.make_experiment_desc()
62 node1 = pl.create("Node")
63 node2 = pl.create("Node")
64 node1.set_attribute_value("hostname", self.host1)
65 node2.set_attribute_value("hostname", self.host2)
66 iface1 = pl.create("NodeInterface")
67 iface2 = pl.create("NodeInterface")
68 iface2.set_attribute_value("label", "node2iface")
69 inet = pl.create("Internet")
70 node1.connector("devs").connect(iface1.connector("node"))
71 node2.connector("devs").connect(iface2.connector("node"))
72 iface1.connector("inet").connect(inet.connector("devs"))
73 iface2.connector("inet").connect(inet.connector("devs"))
74 app = pl.create("Application")
75 app.set_attribute_value("command", "ping -qc1 {#[node2iface].addr[0].[Address]#}")
76 app.enable_trace("stdout")
77 app.connector("node").connect(node1.connector("apps"))
81 controller = ExperimentController(xml, self.root_dir)
83 while not controller.is_finished(app.guid):
85 ping_result = controller.trace(app.guid, "stdout")
86 comp_result = r"""PING .* \(.*\) \d*\(\d*\) bytes of data.
88 --- .* ping statistics ---
89 1 packets transmitted, 1 received, 0% packet loss, time \d*ms.*
91 self.assertTrue(re.match(comp_result, ping_result, re.MULTILINE),
92 "Unexpected trace:\n" + ping_result)
96 if __name__ == '__main__':