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))
493 serverApps.Stop(ns.Seconds(10))
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)))
499 echoClient.SetAttribute("PacketSize", ns.UintegerValue(101))
500
501 clientApps = echoClient.Install(nodes.Get(0))
502 clientApps.Start(ns.Seconds(2))
503 clientApps.Stop(ns.Seconds(10))
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.
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.
bool __python_owns__
python_owns flag indicates that Cppyy should not manage the lifetime of this variable
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)