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>
21 # Test based on netns test/test_core.py file test_run_ping_routing test
24 from nepi.resources.netns.netnswrapper import NetNSWrapper
26 from test_utils import skipIf
34 class NetNSWrapperTest(unittest.TestCase):
38 @skipIf(os.getuid() != 0, "Test requires root privileges")
39 def test_run_ping_routing(self):
40 wrapper = NetNSWrapper()
46 n1 = wrapper.create("Node")
47 n2 = wrapper.create("Node")
48 n3 = wrapper.create("Node")
50 ### add interfaces to nodes
55 i1 = wrapper.invoke(n1, "add_if")
56 i2a = wrapper.invoke(n2, "add_if")
57 i2b = wrapper.invoke(n2, "add_if")
58 i3 = wrapper.invoke(n3, "add_if")
61 # i1.up = i2a.up = i2b.up = i3.up = True
62 wrapper.set(i1, "up", True)
63 wrapper.set(i2a, "up", True)
64 wrapper.set(i2b, "up", True)
65 wrapper.set(i3, "up", True)
70 l1 = wrapper.create("Switch")
71 l2 = wrapper.create("Switch")
73 ### connect interfaces to switches
78 wrapper.invoke(l1, "connect", i1)
79 wrapper.invoke(l1, "connect", i2a)
80 wrapper.invoke(l2, "connect", i2b)
81 wrapper.invoke(l2, "connect", i3)
84 # l1.up = l2.up = True
85 wrapper.set(l1, "up", True)
86 wrapper.set(l2, "up", True)
88 ## add ip addresses to interfaces
89 #i1.add_v4_address('10.0.0.1', 24)
90 #i2a.add_v4_address('10.0.0.2', 24)
91 #i2b.add_v4_address('10.0.1.1', 24)
92 #i3.add_v4_address('10.0.1.2', 24)
93 wrapper.invoke(i1, "add_v4_address", "10.0.0.1", 24)
94 wrapper.invoke(i2a, "add_v4_address", "10.0.0.2", 24)
95 wrapper.invoke(i2b, "add_v4_address", "10.0.1.1", 24)
96 wrapper.invoke(i3, "add_v4_address", "10.0.1.2", 24)
98 ## add routes to nodes
99 #n1.add_route(prefix = '10.0.1.0', prefix_len = 24,
100 # nexthop = '10.0.0.2')
101 #n3.add_route(prefix = '10.0.0.0', prefix_len = 24,
102 # nexthop = '10.0.1.1')
103 wrapper.invoke(n1, "add_route", prefix = "10.0.1.0", prefix_len = 24,
104 nexthop = "10.0.0.2")
105 wrapper.invoke(n3, "add_route", prefix = "10.0.0.0", prefix_len = 24,
106 nexthop = "10.0.1.1")
109 #a1 = n1.Popen(['ping', '-qc1', '10.0.1.2'], stdout = null)
110 #a2 = n3.Popen(['ping', '-qc1', '10.0.0.1'], stdout = null)
111 path1 = "/tmp/netns_file1"
112 path2 = "/tmp/netns_file2"
113 file1 = wrapper.create("open", path1, "w")
114 file2 = wrapper.create("open", path2, "w")
115 a1 = wrapper.invoke(n1, "Popen", ["ping", "-qc1", "10.0.1.2"], stdout = file1)
116 a2 = wrapper.invoke(n3, "Popen", ["ping", "-qc1", "10.0.0.1"], stdout = file2)
121 while p1 is None or p2 is None:
122 p1 = wrapper.invoke(a1, "poll")
123 p2 = wrapper.invoke(a2, "poll")
125 stdout1 = open(path1, "r")
126 stdout2 = open(path2, "r")
131 expected = "1 packets transmitted, 1 received, 0% packet loss"
132 self.assertTrue(s1.find(expected) > -1)
133 self.assertTrue(s2.find(expected) > -1)
137 if __name__ == '__main__':