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
15 class NetnsIntegrationTestCase(unittest.TestCase):
17 self._root_dir = os.path.join(os.getenv("HOME"), ".nepi",
19 os.makedirs(self._root_dir)
21 @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
22 def test_local_if(self):
23 exp_desc = ExperimentDescription()
24 testbed_version = "01"
26 user = getpass.getuser()
27 netns_provider = FactoriesProvider(testbed_id, testbed_version)
28 netns_desc = exp_desc.add_testbed_description(netns_provider)
29 netns_desc.set_attribute_value("homeDirectory", self._root_dir)
30 #netns_desc.set_attribute_value("enableDebug", True)
31 node1 = netns_desc.create("Node")
32 node2 = netns_desc.create("Node")
33 iface1 = netns_desc.create("NodeInterface")
34 iface1.set_attribute_value("up", True)
35 node1.connector("devs").connect(iface1.connector("node"))
36 ip1 = iface1.add_address()
37 ip1.set_attribute_value("Address", "10.0.0.1")
38 iface2 = netns_desc.create("NodeInterface")
39 iface2.set_attribute_value("up", True)
40 node2.connector("devs").connect(iface2.connector("node"))
41 ip2 = iface2.add_address()
42 ip2.set_attribute_value("Address", "10.0.0.2")
43 switch = netns_desc.create("Switch")
44 switch.set_attribute_value("up", True)
45 iface1.connector("switch").connect(switch.connector("devs"))
46 iface2.connector("switch").connect(switch.connector("devs"))
47 app = netns_desc.create("Application")
48 app.set_attribute_value("command", "ping -qc1 10.0.0.2")
49 app.set_attribute_value("user", user)
50 app.connector("node").connect(node1.connector("apps"))
51 app.enable_trace("stdout")
52 xml = exp_desc.to_xml()
54 controller = ExperimentController(xml)
56 while not controller.is_finished(app.guid):
58 ping_result = controller.trace(netns_desc.guid, app.guid, "stdout")
59 comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
61 --- 10.0.0.2 ping statistics ---
62 1 packets transmitted, 1 received, 0% packet loss, time 0ms
64 self.assertTrue(ping_result.startswith(comp_result))
68 @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
69 def test_all_daemonized_if(self):
70 exp_desc = ExperimentDescription()
71 testbed_version = "01"
73 user = getpass.getuser()
74 netns_provider = FactoriesProvider(testbed_id, testbed_version)
75 netns_desc = exp_desc.add_testbed_description(netns_provider)
76 netns_desc.set_attribute_value("homeDirectory", self._root_dir)
77 #netns_desc.set_attribute_value("enableDebug", True)
78 node1 = netns_desc.create("Node")
79 node2 = netns_desc.create("Node")
80 iface1 = netns_desc.create("NodeInterface")
81 iface1.set_attribute_value("up", True)
82 node1.connector("devs").connect(iface1.connector("node"))
83 ip1 = iface1.add_address()
84 ip1.set_attribute_value("Address", "10.0.0.1")
85 iface2 = netns_desc.create("NodeInterface")
86 iface2.set_attribute_value("up", True)
87 node2.connector("devs").connect(iface2.connector("node"))
88 ip2 = iface2.add_address()
89 ip2.set_attribute_value("Address", "10.0.0.2")
90 switch = netns_desc.create("Switch")
91 switch.set_attribute_value("up", True)
92 iface1.connector("switch").connect(switch.connector("devs"))
93 iface2.connector("switch").connect(switch.connector("devs"))
94 app = netns_desc.create("Application")
95 app.set_attribute_value("command", "ping -qc1 10.0.0.2")
96 app.set_attribute_value("user", user)
97 app.connector("node").connect(node1.connector("apps"))
98 app.enable_trace("stdout")
99 xml = exp_desc.to_xml()
101 access_config = proxy.AccessConfiguration()
102 access_config.set_attribute_value("mode",
103 proxy.AccessConfiguration.MODE_DAEMON)
104 access_config.set_attribute_value("rootDirectory", self._root_dir)
105 access_config.set_attribute_value("logLevel",
106 proxy.AccessConfiguration.DEBUG_LEVEL)
107 controller = proxy.create_controller(xml, access_config)
109 access_config2 = proxy.AccessConfiguration()
110 access_config2.set_attribute_value("mode",
111 proxy.AccessConfiguration.MODE_DAEMON)
112 inst_root_dir = os.path.join(self._root_dir, "instance")
113 os.mkdir(inst_root_dir)
114 access_config2.set_attribute_value("rootDirectory", inst_root_dir)
115 access_config2.set_attribute_value("logLevel",
116 proxy.AccessConfiguration.DEBUG_LEVEL)
117 controller.set_access_configuration(netns_desc.guid, access_config2)
120 while not controller.is_finished(app.guid):
122 ping_result = controller.trace(netns_desc.guid, app.guid, "stdout")
123 comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
125 --- 10.0.0.2 ping statistics ---
126 1 packets transmitted, 1 received, 0% packet loss, time 0ms
128 self.assertTrue(ping_result.startswith(comp_result))
130 controller.shutdown()
133 shutil.rmtree(self._root_dir)
135 if __name__ == '__main__':