A Discrete-Event Network Simulator
API
Loading...
Searching...
No Matches
python-unit-tests.py
Go to the documentation of this file.
1#! /usr/bin/env python3
2
3# Copyright (C) 2008-2011 INESC Porto
4
5# SPDX-License-Identifier: GPL-2.0-or-later
6
7# Author: Gustavo J. A. M. Carneiro <gjc@inescporto.pt>
8
9import unittest
10
11try:
12 from ns import ns
13except ModuleNotFoundError:
14 raise SystemExit(
15 "Error: ns3 Python module not found;"
16 " Python bindings may not be enabled"
17 " or your PYTHONPATH might not be properly configured"
18 )
19import sys
20
21UINT32_MAX = 0xFFFFFFFF
22
23
24## TestSimulator class
25class TestSimulator(unittest.TestCase):
26 ## @var _received_packet
27 # received packet
28 ## @var _args_received
29 # args
30 ## @var _cb_time
31 # current time
32 ## @var _context_received
33 # context
34
35 def testScheduleNow(self):
36 """! Test schedule now
37 @param self this object
38 @return None
39 """
40
41 def callback(args: ns.cppyy.gbl.std.vector) -> None:
42 """! Callback function
43 @param args arguments
44 @return None
45 """
46 self._args_received = list(map(lambda x: x.decode("utf-8"), args))
47 self._cb_time = ns.Simulator.Now()
48
49 ns.Simulator.Destroy()
50 self._args_received = None
51 self._cb_time = None
52 ns.cppyy.cppdef(
53 """
54 EventImpl* pythonMakeEvent(void (*f)(std::vector<std::string>), std::vector<std::string> l)
55 {
56 return MakeEvent(f, l);
57 }
58 """
59 )
60 event = ns.cppyy.gbl.pythonMakeEvent(callback, sys.argv)
61 ns.Simulator.ScheduleNow(event)
62 ns.Simulator.Run()
63 self.assertListEqual(self._args_received, sys.argv)
64 self.assertEqual(self._cb_time.GetSeconds(), 0.0)
65
66 def testSchedule(self):
67 """! Test schedule
68 @param self this object
69 @return None
70 """
71
72 def callback(args: ns.cppyy.gbl.std.vector):
73 """! Callback function
74 @param args arguments
75 @return None
76 """
77 self._args_received = list(map(lambda x: x.decode("utf-8"), args))
78 self._cb_time = ns.Simulator.Now()
79
80 ns.Simulator.Destroy()
81 self._args_received = None
82 self._cb_time = None
83 ns.cppyy.cppdef(
84 """
85 EventImpl* pythonMakeEvent2(void (*f)(std::vector<std::string>), std::vector<std::string> l)
86 {
87 return MakeEvent(f, l);
88 }
89 """
90 )
91 event = ns.cppyy.gbl.pythonMakeEvent2(callback, sys.argv)
92 ns.Simulator.Schedule(ns.Seconds(123), event)
93 ns.Simulator.Run()
94 self.assertListEqual(self._args_received, sys.argv)
95 self.assertEqual(self._cb_time.GetSeconds(), 123.0)
96
98 """! Test schedule destroy
99 @param self this object
100 @return None
101 """
102
103 def callback(args: ns.cppyy.gbl.std.vector):
104 """! Callback function
105 @param args
106 @return None
107 """
108 self._args_received = list(map(lambda x: x.decode("utf-8"), args))
109 self._cb_time = ns.Simulator.Now()
110
111 ns.Simulator.Destroy()
112 self._args_received = None
113 self._cb_time = None
114 ns.cppyy.cppdef("void null(){ return; }")
115 ns.Simulator.Schedule(ns.Seconds(123), ns.cppyy.gbl.null)
116 ns.cppyy.cppdef(
117 """
118 EventImpl* pythonMakeEvent3(void (*f)(std::vector<std::string>), std::vector<std::string> l)
119 {
120 return MakeEvent(f, l);
121 }
122 """
123 )
124 event = ns.cppyy.gbl.pythonMakeEvent3(callback, sys.argv)
125 ns.Simulator.ScheduleDestroy(event)
126 ns.Simulator.Run()
127 ns.Simulator.Destroy()
128 self.assertListEqual(self._args_received, sys.argv)
129 self.assertEqual(self._cb_time.GetSeconds(), 123.0)
130
132 """! Test schedule with context
133 @param self this object
134 @return None
135 """
136
137 def callback(context, args: ns.cppyy.gbl.std.vector):
138 """! Callback
139 @param context the context
140 @param args the arguments
141 @return None
142 """
143 self._context_received = context
144 self._args_received = list(map(lambda x: x.decode("utf-8"), args))
145 self._cb_time = ns.Simulator.Now()
146
147 ns.Simulator.Destroy()
148 self._args_received = None
149 self._cb_time = None
150 self._context_received = None
151 ns.cppyy.cppdef(
152 """
153 EventImpl* pythonMakeEvent4(void (*f)(uint32_t, std::vector<std::string>), uint32_t context, std::vector<std::string> l)
154 {
155 return MakeEvent(f, context, l);
156 }
157 """
158 )
159 event = ns.cppyy.gbl.pythonMakeEvent4(callback, 54321, sys.argv)
160 ns.Simulator.ScheduleWithContext(54321, ns.Seconds(123), event)
161 ns.Simulator.Run()
162 self.assertEqual(self._context_received, 54321)
163 self.assertListEqual(self._args_received, sys.argv)
164 self.assertEqual(self._cb_time.GetSeconds(), 123.0)
165
167 """! Test time comparison
168 @param self this object
169 @return None
170 """
171 self.assertTrue(ns.Seconds(123) == ns.Seconds(123))
172 self.assertTrue(ns.Seconds(123) >= ns.Seconds(123))
173 self.assertTrue(ns.Seconds(123) <= ns.Seconds(123))
174 self.assertTrue(ns.Seconds(124) > ns.Seconds(123))
175 self.assertTrue(ns.Seconds(123) < ns.Seconds(124))
176
178 """! Test numeric operations
179 @param self this object
180 @return None
181 """
182 self.assertEqual(ns.Seconds(10) + ns.Seconds(5), ns.Seconds(15))
183 self.assertEqual(ns.Seconds(10) - ns.Seconds(5), ns.Seconds(5))
184
185 v1 = ns.int64x64_t(5.0) * ns.int64x64_t(10)
186 self.assertEqual(v1, ns.int64x64_t(50))
187
188 def testConfig(self):
189 """! Test configuration
190 @param self this object
191 @return None
192 """
193 ns.Config.SetDefault("ns3::OnOffApplication::PacketSize", ns.UintegerValue(123))
194 # hm.. no Config.Get?
195
196 def testSocket(self):
197 """! Test socket
198 @param self
199 @return None
200 """
201 nc = ns.NodeContainer(1)
202 node = nc.Get(0)
203 internet = ns.InternetStackHelper()
204 internet.Install(node)
206
207 def python_rx_callback(socket) -> None:
208 self._received_packet = socket.Recv(maxSize=UINT32_MAX, flags=0)
209
210 ns.cppyy.cppdef(
211 """
212 Callback<void,ns3::Ptr<ns3::Socket> > make_rx_callback_test_socket(void(*func)(Ptr<Socket>))
213 {
214 return MakeCallback(func);
215 }
216 """
217 )
218
219 sink = ns.Socket.CreateSocket(node, ns.TypeId.LookupByName("ns3::UdpSocketFactory"))
220 sink.Bind(ns.InetSocketAddress(ns.Ipv4Address.GetAny(), 80).ConvertTo())
221 sink.SetRecvCallback(ns.cppyy.gbl.make_rx_callback_test_socket(python_rx_callback))
222
223 source = ns.Socket.CreateSocket(node, ns.TypeId.LookupByName("ns3::UdpSocketFactory"))
224 source.SendTo(
225 ns.Packet(19),
226 0,
227 ns.InetSocketAddress(ns.Ipv4Address("127.0.0.1"), 80).ConvertTo(),
228 )
229
230 ns.Simulator.Run()
231 self.assertTrue(self._received_packet is not None)
232 self.assertEqual(self._received_packet.GetSize(), 19)
233
234 # Delete Ptr<>'s on the python side to let C++ clean them
235 del internet
236
237 def testAttributes(self):
238 """! Test attributes function
239 @param self this object
240 @return None
241 """
242 # Templated class DropTailQueue[ns.Packet] in C++
243 queue = ns.CreateObject[ns.DropTailQueue[ns.Packet]]()
244 queueSizeValue = ns.QueueSizeValue(ns.QueueSize("500p"))
245 queue.SetAttribute("MaxSize", queueSizeValue)
246
247 limit = ns.QueueSizeValue()
248 queue.GetAttribute("MaxSize", limit)
249 self.assertEqual(limit.Get(), ns.QueueSize("500p"))
250
251 ## -- object pointer values
252 mobility = ns.CreateObject[ns.RandomWaypointMobilityModel]()
253 ptr = ns.PointerValue()
254 mobility.GetAttribute("PositionAllocator", ptr)
255 self.assertEqual(ptr.GetObject(), ns.Ptr["Object"](ns.cppyy.nullptr))
256
257 pos = ns.ListPositionAllocator()
258 ptr.SetObject(pos)
259 mobility.SetAttribute("PositionAllocator", ptr)
260
261 ptr2 = ns.PointerValue()
262 mobility.GetAttribute("PositionAllocator", ptr2)
263 self.assertNotEqual(ptr.GetObject(), ns.Ptr["Object"](ns.cppyy.nullptr))
264
265 # Delete Ptr<>'s on the python side to let C++ clean them
266 del queue, mobility, ptr, ptr2
267
268 def testIdentity(self):
269 """! Test identify
270 @param self this object
271 @return None
272 """
273 csma = ns.CreateObject[ns.CsmaNetDevice]()
274 channel = ns.CreateObject[ns.CsmaChannel]()
275 csma.Attach(channel)
276
277 c1 = csma.GetChannel()
278 c2 = csma.GetChannel()
279
280 self.assertEqual(c1, c2)
281
282 # Delete Ptr<>'s on the python side to let C++ clean them
283 del csma, channel
284
285 def testTypeId(self):
286 """! Test type ID
287 @param self this object
288 @return None
289 """
290 ok, typeId1 = ns.LookupByNameFailSafe("ns3::UdpSocketFactory")
291 self.assertTrue(ok)
292 self.assertEqual(typeId1.GetName(), "ns3::UdpSocketFactory")
293
294 ok, typeId1 = ns.LookupByNameFailSafe("ns3::__InvalidTypeName__")
295 self.assertFalse(ok)
296
298 """! Test command line
299 @param self this object
300 @return None
301 """
302 from ctypes import c_bool, c_char_p, c_double, c_int, create_string_buffer
303
304 test1 = c_bool(True)
305 test2 = c_int(42)
306 test3 = c_double(3.1415)
307 BUFFLEN = 40 # noqa
308 test4Buffer = create_string_buffer(b"this is a test option", BUFFLEN)
309 test4 = c_char_p(test4Buffer.raw)
310
311 cmd = ns.CommandLine(__file__)
312 cmd.AddValue("Test1", "this is a test option", test1)
313 cmd.AddValue("Test2", "this is a test option", test2)
314 cmd.AddValue["double"]("Test3", "this is a test option", test3)
315 cmd.AddValue("Test4", "this is a test option", test4, BUFFLEN)
316
317 cmd.Parse(["python"])
318 self.assertEqual(test1.value, True)
319 self.assertEqual(test2.value, 42)
320 self.assertEqual(test3.value, 3.1415)
321 self.assertEqual(test4.value, b"this is a test option")
322
323 cmd.Parse(["python", "--Test1=false", "--Test2=0", "--Test3=0.0"])
324 self.assertEqual(test1.value, False)
325 self.assertEqual(test2.value, 0)
326 self.assertEqual(test3.value, 0.0)
327
328 cmd.Parse(["python", "--Test4=new_string"])
329 self.assertEqual(test4.value, b"new_string")
330
331 def testSubclass(self):
332 """! Test subclass
333 @param self this object
334 @return None
335 """
336
337 ## MyNode class
338 class MyNode(ns.Node):
339 def GetLocalTime(self) -> ns.Time:
340 return ns.Seconds(10)
341
342 node = MyNode()
343 forced_local_time = node.GetLocalTime()
344 self.assertEqual(forced_local_time, ns.Seconds(10))
345 del node
346
348 """! Test python-based application
349 @param self this object
350 @return None
351 """
352 ns.Simulator.Destroy()
353
354 nodes = ns.NodeContainer()
355 nodes.Create(2)
356
357 pointToPoint = ns.PointToPointHelper()
358 pointToPoint.SetDeviceAttribute("DataRate", ns.StringValue("5Mbps"))
359 pointToPoint.SetChannelAttribute("Delay", ns.StringValue("2ms"))
360
361 devices = pointToPoint.Install(nodes)
362
363 stack = ns.InternetStackHelper()
364 stack.Install(nodes)
365
366 address = ns.Ipv4AddressHelper()
367 address.SetBase(ns.Ipv4Address("10.1.1.0"), ns.Ipv4Mask("255.255.255.0"))
368
369 interfaces = address.Assign(devices)
370
371 ns.cppyy.cppdef(
372 """
373 namespace ns3
374 {
375 Callback<void,Ptr<Socket> > make_rx_callback(void(*func)(Ptr<Socket>))
376 {
377 return MakeCallback(func);
378 }
379 EventImpl* pythonMakeEventSend(void (*f)(Ptr<Socket>, Ptr<Packet>, Address&), Ptr<Socket> socket, Ptr<Packet> packet, Address address)
380 {
381 return MakeEvent(f, socket, packet, address);
382 }
383 }
384 """
385 )
386
387 ## EchoServer application class
388 class EchoServer(ns.Application):
389 LOGGING = False
390 ECHO_PORT = 1234
391 socketToInstanceDict = {}
392
393 def __init__(self, node: ns.Node, port=ECHO_PORT):
394 """! Constructor needs to call first the constructor to Application (super class)
395 @param self this object
396 @param node node where this application will be executed
397 @param port port to listen
398 return None
399 """
400 super().__init__()
401 ## __python_owns__ flag indicates that Cppyy should not manage the lifetime of this variable
402 self.__python_owns__ = False # Let C++ destroy this on Simulator::Destroy
403 ## Listen port for the server
404 self.port = port
405 ## Socket used by the server to listen to port
406 self.m_socket = ns.Socket.CreateSocket(
407 node, ns.TypeId.LookupByName("ns3::UdpSocketFactory")
408 )
409 self.m_socket.Bind(
410 ns.InetSocketAddress(ns.Ipv4Address.GetAny(), self.port).ConvertTo()
411 )
412 self.m_socket.SetRecvCallback(ns.make_rx_callback(EchoServer._Receive))
413 EchoServer.socketToInstanceDict[self.m_socket] = self
414
415 def __del__(self):
416 """! Destructor
417 @param self this object
418 return None
419 """
420 del EchoServer.socketToInstanceDict[self.m_socket]
421
422 def Send(self, packet: ns.Packet, address: ns.Address) -> None:
423 """! Function to send a packet to an address
424 @param self this object
425 @param packet packet to send
426 @param address destination address
427 return None
428 """
429 self.m_socket.SendTo(packet, 0, address)
430 if EchoServer.LOGGING:
431 inetAddress = ns.InetSocketAddress.ConvertFrom(address)
432 print(
433 "At time +{s}s server sent {b} bytes from {ip} port {port}".format(
434 s=ns.Simulator.Now().GetSeconds(),
435 b=packet.__deref__().GetSize(),
436 ip=inetAddress.GetIpv4(),
437 port=inetAddress.GetPort(),
438 ),
439 file=sys.stderr,
440 flush=True,
441 )
442
443 def Receive(self):
444 """! Function to receive a packet from an address
445 @param self this object
446 @return None
447 """
448 address = ns.Address()
449 packet = self.m_socket.RecvFrom(address)
450 if EchoServer.LOGGING:
451 inetAddress = ns.InetSocketAddress.ConvertFrom(address)
452 print(
453 "At time +{s}s server received {b} bytes from {ip} port {port}".format(
454 s=ns.Simulator.Now().GetSeconds(),
455 b=packet.__deref__().GetSize(),
456 ip=inetAddress.GetIpv4(),
457 port=inetAddress.GetPort(),
458 ),
459 file=sys.stderr,
460 flush=True,
461 )
462 event = ns.pythonMakeEventSend(EchoServer._Send, self.m_socket, packet, address)
463 ns.Simulator.Schedule(ns.Seconds(1), event)
464
465 @staticmethod
466 def _Send(socket: ns.Socket, packet: ns.Packet, address: ns.Address):
467 """! Static send function, which matches the output socket
468 to the EchoServer instance to call the instance Send function
469 @param socket socket from the instance that should send the packet
470 @param packet packet to send
471 @param address destination address
472 return None
473 """
474 instance = EchoServer.socketToInstanceDict[socket]
475 instance.Send(packet, address)
476
477 @staticmethod
478 def _Receive(socket: ns.Socket) -> None:
479 """! Static receive function, which matches the input socket
480 to the EchoServer instance to call the instance Receive function
481 @param socket socket from the instance that should receive the packet
482 return None
483 """
484 instance = EchoServer.socketToInstanceDict[socket]
485 instance.Receive()
486
487 echoServer = EchoServer(nodes.Get(1))
488 nodes.Get(1).AddApplication(echoServer)
489
490 serverApps = ns.ApplicationContainer()
491 serverApps.Add(echoServer)
492 serverApps.Start(ns.Seconds(1.0))
493 serverApps.Stop(ns.Seconds(10.0))
494
495 address = interfaces.GetAddress(1).ConvertTo()
496 echoClient = ns.UdpEchoClientHelper(address, EchoServer.ECHO_PORT)
497 echoClient.SetAttribute("MaxPackets", ns.UintegerValue(10))
498 echoClient.SetAttribute("Interval", ns.TimeValue(ns.Seconds(1.0)))
499 echoClient.SetAttribute("PacketSize", ns.UintegerValue(101))
500
501 clientApps = echoClient.Install(nodes.Get(0))
502 clientApps.Start(ns.Seconds(2.0))
503 clientApps.Stop(ns.Seconds(10.0))
504
505 ns.Simulator.Run()
506 ns.Simulator.Destroy()
507
508
509if __name__ == "__main__":
510 unittest.main(verbosity=1, failfast=True)
port
Listen port for the server.
__python_owns__
python_owns flag indicates that Cppyy should not manage the lifetime of this variable
testCommandLine(self)
Test command line.
testTimeComparison(self)
Test time comparison.
testScheduleWithContext(self)
Test schedule with context.
testScheduleNow(self)
Test schedule now.
testConfig(self)
Test configuration.
testEchoServerApplication(self)
Test python-based application.
testTimeNumericOperations(self)
Test numeric operations.
testScheduleDestroy(self)
Test schedule destroy.
m_socket
Socket used by the server to listen to port.
testAttributes(self)
Test attributes function.
static void Send(Ptr< NetDevice > dev, int level, std::string emuMode)