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 version 2 as
8 # published by the Free Software Foundation;
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
20 # Test based on netns test/test_core.py file test_run_ping_routing test
23 from __future__ import print_function
25 from nepi.resources.netns.netnsserver import run_server
26 from nepi.resources.linux.netns.netnsclient import LinuxNetNSClient
28 from test_utils import skipIf
36 class DummyEmulation(object):
37 def __init__(self, socket_name):
38 self.socket_name = socket_name
39 self.node = dict({'hostname': 'localhost'})
42 def remote_socket(self):
43 return self.socket_name
45 class LinuxNetNSClientTest(unittest.TestCase):
47 self.socket_name = os.path.join("/", "tmp", "NetNSWrapperServer.sock")
48 if os.path.exists(self.socket_name):
49 os.remove(self.socket_name)
52 os.remove(self.socket_name)
54 @skipIf(os.getuid() != 0, "Test requires root privileges")
55 def test_run_ping_routing(self):
56 thread = threading.Thread(target = run_server,
57 args = [self.socket_name], kwargs={"level":logging.DEBUG})
59 thread.setDaemon(True)
64 # Verify that the communication socket was created
65 self.assertTrue(os.path.exists(self.socket_name))
67 # Create a dummy simulation object
68 emulation = DummyEmulation(self.socket_name)
70 # Instantiate the NS3 client
71 client = LinuxNetNSClient(emulation)
77 n1 = client.create("Node")
78 n2 = client.create("Node")
79 n3 = client.create("Node")
81 ### add interfaces to nodes
86 i1 = client.invoke(n1, "add_if")
87 i2a = client.invoke(n2, "add_if")
88 i2b = client.invoke(n2, "add_if")
89 i3 = client.invoke(n3, "add_if")
92 # i1.up = i2a.up = i2b.up = i3.up = True
93 client.set(i1, "up", True)
94 client.set(i2a, "up", True)
95 client.set(i2b, "up", True)
96 client.set(i3, "up", True)
101 l1 = client.create("Switch")
102 l2 = client.create("Switch")
104 ### connect interfaces to switches
109 client.invoke(l1, "connect", i1)
110 client.invoke(l1, "connect", i2a)
111 client.invoke(l2, "connect", i2b)
112 client.invoke(l2, "connect", i3)
115 # l1.up = l2.up = True
116 client.set(l1, "up", True)
117 client.set(l2, "up", True)
119 ## add ip addresses to interfaces
120 #i1.add_v4_address('10.0.0.1', 24)
121 #i2a.add_v4_address('10.0.0.2', 24)
122 #i2b.add_v4_address('10.0.1.1', 24)
123 #i3.add_v4_address('10.0.1.2', 24)
124 client.invoke(i1, "add_v4_address", "10.0.0.1", 24)
125 client.invoke(i2a, "add_v4_address", "10.0.0.2", 24)
126 client.invoke(i2b, "add_v4_address", "10.0.1.1", 24)
127 client.invoke(i3, "add_v4_address", "10.0.1.2", 24)
129 ## add routes to nodes
130 #n1.add_route(prefix = '10.0.1.0', prefix_len = 24,
131 # nexthop = '10.0.0.2')
132 #n3.add_route(prefix = '10.0.0.0', prefix_len = 24,
133 # nexthop = '10.0.1.1')
134 client.invoke(n1, "add_route", prefix="10.0.1.0", prefix_len=24,
136 client.invoke(n3, "add_route", prefix="10.0.0.0", prefix_len=24,
140 #a1 = n1.Popen(['ping', '-qc1', '10.0.1.2'], stdout = null)
141 #a2 = n3.Popen(['ping', '-qc1', '10.0.0.1'], stdout = null)
142 path1 = "/tmp/netns_file1"
143 path2 = "/tmp/netns_file2"
144 file1 = client.create("open", path1, "w")
145 file2 = client.create("open", path2, "w")
146 a1 = client.invoke(n1, "Popen", ["ping", "-qc1", "10.0.1.2"], stdout=file1)
147 a2 = client.invoke(n3, "Popen", ["ping", "-qc1", "10.0.0.1"], stdout=file2)
152 while p1 is None or p2 is None:
153 p1 = client.invoke(a1, "poll")
154 p2 = client.invoke(a2, "poll")
156 with open(path1, "r") as stdout1:
157 with open(path2, "r") as stdout2:
163 expected = "1 packets transmitted, 1 received, 0% packet loss"
164 self.assertTrue(s1.find(expected) > -1)
165 self.assertTrue(s2.find(expected) > -1)
167 # wait until emulation is over
170 if __name__ == '__main__':