A Discrete-Event Network Simulator
API
python-unit-tests.py
Go to the documentation of this file.
1 import unittest
2 from ns.core import Simulator, Seconds, Config, int64x64_t
3 import ns.core
4 import ns.network
5 import ns.internet
6 import ns.mobility
7 import ns.csma
8 import ns.applications
9 
10 
11 
12 class TestSimulator(unittest.TestCase):
13 
21 
22  def testScheduleNow(self):
23  """! Test schedule now
24  @param self this object
25  @return none
26  """
27  def callback(args):
28  """! Callback function
29  @param args arguments
30  return none
31  """
32  self._args_received = args
33  self._cb_time = Simulator.Now()
34  Simulator.Destroy()
35  self._args_received = None
36  self._cb_time = None
37  Simulator.ScheduleNow(callback, "args")
38  Simulator.Run()
39  self.assertEqual(self._args_received, "args")
40  self.assertEqual(self._cb_time.GetSeconds(), 0.0)
41 
42  def testSchedule(self):
43  """! Test schedule
44  @param self this object
45  @return none
46  """
47  def callback(args):
48  """! Callback function
49  @param args arguments
50  @return none
51  """
52  self._args_received = args
53  self._cb_time = Simulator.Now()
54  Simulator.Destroy()
55  self._args_received = None
56  self._cb_time = None
57  Simulator.Schedule(Seconds(123), callback, "args")
58  Simulator.Run()
59  self.assertEqual(self._args_received, "args")
60  self.assertEqual(self._cb_time.GetSeconds(), 123.0)
61 
63  """! Test schedule destroy
64  @param self this object
65  @return none
66  """
67  def callback(args):
68  """! Callback function
69  @param args
70  @return none
71  """
72  self._args_received = args
73  self._cb_time = Simulator.Now()
74  Simulator.Destroy()
75  self._args_received = None
76  self._cb_time = None
77  def null(): pass
78  Simulator.Schedule(Seconds(123), null)
79  Simulator.ScheduleDestroy(callback, "args")
80  Simulator.Run()
81  Simulator.Destroy()
82  self.assertEqual(self._args_received, "args")
83  self.assertEqual(self._cb_time.GetSeconds(), 123.0)
84 
86  """! Test schedule with context
87  @param self this object
88  @return none
89  """
90  def callback(context, args):
91  """! Callback
92  @param context the cntet
93  @param args the arguments
94  @return none
95  """
96  self._context_received = context
97  self._args_received = args
98  self._cb_time = Simulator.Now()
99  Simulator.Destroy()
100  self._args_received = None
101  self._cb_time = None
102  self._context_received = None
103  Simulator.ScheduleWithContext(54321, Seconds(123), callback, "args")
104  Simulator.Run()
105  self.assertEqual(self._context_received, 54321)
106  self.assertEqual(self._args_received, "args")
107  self.assertEqual(self._cb_time.GetSeconds(), 123.0)
108 
110  """! Test time comparison
111  @param self this object
112  @return none
113  """
114  self.assert_(Seconds(123) == Seconds(123))
115  self.assert_(Seconds(123) >= Seconds(123))
116  self.assert_(Seconds(123) <= Seconds(123))
117  self.assert_(Seconds(124) > Seconds(123))
118  self.assert_(Seconds(123) < Seconds(124))
119 
121  """! Test numeric operations
122  @param self this object
123  @return none
124  """
125  self.assertEqual(Seconds(10) + Seconds(5), Seconds(15))
126  self.assertEqual(Seconds(10) - Seconds(5), Seconds(5))
127 
128  v1 = int64x64_t(5.0)*int64x64_t(10)
129  self.assertEqual(v1, int64x64_t(50))
130 
131  def testConfig(self):
132  """! Test configuration
133  @param self this object
134  @return none
135  """
136  Config.SetDefault("ns3::OnOffApplication::PacketSize", ns.core.UintegerValue(123))
137  # hm.. no Config.Get?
138 
139  def testSocket(self):
140  """! Test socket
141  @param self
142  @return none
143  """
144  node = ns.network.Node()
145  internet = ns.internet.InternetStackHelper()
146  internet.Install(node)
147  self._received_packet = None
148 
149  def rx_callback(socket):
150  """! Receive Callback
151  @param socket the socket to receive
152  @return none
153  """
154  assert self._received_packet is None
155  self._received_packet = socket.Recv()
156 
157  sink = ns.network.Socket.CreateSocket(node, ns.core.TypeId.LookupByName("ns3::UdpSocketFactory"))
158  sink.Bind(ns.network.InetSocketAddress(ns.network.Ipv4Address.GetAny(), 80))
159  sink.SetRecvCallback(rx_callback)
160 
161  source = ns.network.Socket.CreateSocket(node, ns.core.TypeId.LookupByName("ns3::UdpSocketFactory"))
162  source.SendTo(ns.network.Packet(19), 0, ns.network.InetSocketAddress(ns.network.Ipv4Address("127.0.0.1"), 80))
163 
164  Simulator.Run()
165  self.assert_(self._received_packet is not None)
166  self.assertEqual(self._received_packet.GetSize(), 19)
167 
168 
169  def testAttributes(self):
170  """! Test attributes function
171  @param self this object
172  @return none
173  """
174 
178  queue = ns.network.DropTailQueue()
179 
180  queue.SetAttribute("MaxPackets", ns.core.UintegerValue(123456))
181 
182  limit = ns.core.UintegerValue()
183  queue.GetAttribute("MaxPackets", limit)
184  self.assertEqual(limit.Get(), 123456)
185 
186 
187  mobility = ns.mobility.RandomWaypointMobilityModel()
188  ptr = ns.core.PointerValue()
189  mobility.GetAttribute("PositionAllocator", ptr)
190  self.assertEqual(ptr.GetObject(), None)
191 
192  pos = ns.mobility.ListPositionAllocator()
193  mobility.SetAttribute("PositionAllocator", ns.core.PointerValue(pos))
194 
195  ptr = ns.core.PointerValue()
196  mobility.GetAttribute("PositionAllocator", ptr)
197  self.assert_(ptr.GetObject() is not None)
198 
199  def testIdentity(self):
200  """! Test identify
201  @param self this object
202  @return none
203  """
204  csma = ns.csma.CsmaNetDevice()
205  channel = ns.csma.CsmaChannel()
206  csma.Attach(channel)
207 
208  c1 = csma.GetChannel()
209  c2 = csma.GetChannel()
210 
211  self.assert_(c1 is c2)
212 
213  def testTypeId(self):
214  """! Test type ID
215  @param self this object
216  @return none
217  """
218  typeId1 = ns.core.TypeId.LookupByNameFailSafe("ns3::UdpSocketFactory")
219  self.assertEqual(typeId1.GetName (), "ns3::UdpSocketFactory")
220 
221  self.assertRaises(KeyError, ns.core.TypeId.LookupByNameFailSafe, "__InvalidTypeName__")
222 
223  def testCommandLine(self):
224  """! Test command line
225  @param self this object
226  @return none
227  """
228  cmd = ns.core.CommandLine()
229  cmd.AddValue("Test1", "this is a test option")
230  cmd.AddValue("Test2", "this is a test option")
231  cmd.AddValue("Test3", "this is a test option", variable="test_xxx")
232  cmd.Test1 = None
233  cmd.Test2 = None
234  cmd.test_xxx = None
235  class Foo:
236  pass
237  foo = Foo()
238  foo.test_foo = None
239  cmd.AddValue("Test4", "this is a test option", variable="test_foo", namespace=foo)
240 
241  cmd.Parse(["python", "--Test1=value1", "--Test2=value2", "--Test3=123", "--Test4=xpto"])
242 
243  self.assertEqual(cmd.Test1, "value1")
244  self.assertEqual(cmd.Test2, "value2")
245  self.assertEqual(cmd.test_xxx, "123")
246  self.assertEqual(foo.test_foo, "xpto")
247 
248  def testSubclass(self):
249  """! Test subclass
250  @param self this object
251  @return none
252  """
253 
254  class MyNode(ns.network.Node):
255  def __init__(self):
256  """! Initializer
257  @param self this object
258  @return none
259  """
260  super(MyNode, self).__init__()
261 
262  node = MyNode()
263 
264 
265 if __name__ == '__main__':
266  unittest.main()
def testSubclass(self)
Test subclass.
def testTypeId(self)
Test type ID.
def testTimeNumericOperations(self)
Test numeric operations.
uint32_t GetSize(Ptr< const Packet > packet, const WifiMacHeader *hdr, bool isAmpdu)
Return the total size of the packet after WifiMacHeader and FCS trailer have been added...
Definition: wifi-utils.cc:238
def testConfig(self)
Test configuration.
def testAttributes(self)
Test attributes function.
def testCommandLine(self)
Test command line.
def testSchedule(self)
Test schedule.
def testTimeComparison(self)
Test time comparison.
def testIdentity(self)
Test identify.
def testScheduleWithContext(self)
Test schedule with context.
def testScheduleNow(self)
Test schedule now.
Time Seconds(double value)
Construct a Time in the indicated unit.
Definition: nstime.h:1070
def testScheduleDestroy(self)
Test schedule destroy.
def testSocket(self)
Test socket.