The Assimilation Monitoring Project
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
tests/cclass_wrappers_test.py
Go to the documentation of this file.
1 # vim: smartindent tabstop=4 shiftwidth=4 expandtab
2 #
3 #
4 # This file is part of the Assimilation Project.
5 #
6 # Copyright (C) 2011, 2012 - Alan Robertson <alanr@unix.sh>
7 #
8 # The Assimilation software is free software: you can redistribute it and/or modify
9 # it under the terms of the GNU General Public License as published by
10 # the Free Software Foundation, either version 3 of the License, or
11 # (at your option) any later version.
12 #
13 # The Assimilation software is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 # GNU General Public License for more details.
17 #
18 # You should have received a copy of the GNU General Public License
19 # along with the Assimilation Project software. If not, see http://www.gnu.org/licenses/
20 #
21 #
22 _suites = ['all', 'cclass']
23 import sys
24 sys.path.append("../pyclasswrappers")
25 sys.path.append("pyclasswrappers")
26 from testify import *
27 
28 from frameinfo import *
29 from AssimCclasses import *
30 import gc
31 import re
32 
33 
34 CheckForDanglingClasses = True
35 WorstDanglingCount = 0
36 DEBUG=False
37 
38 def assert_no_dangling_Cclasses():
39  global CheckForDanglingClasses
40  global WorstDanglingCount
41  gc.collect()
42  count = proj_class_live_object_count()
43  # Avoid cluttering the output up with redundant messages...
44  if count > WorstDanglingCount and CheckForDanglingClasses:
45  WorstDanglingCount = count
46  proj_class_dump_live_objects()
47  raise AssertionError, "Dangling C-class objects - %d still around" % count
48 
49 class pyNetAddrTest(TestCase):
50  "A pyNetAddr is a network address of some kind... - let's test it"
51  def test_constructor(self):
52  if DEBUG: print >>sys.stderr, "===============test_constructor(pyNetAddrTest)"
53  if DEBUG: print >>sys.stderr, "===============test_constructor(pyNetAddrTest)"
54  if DEBUG: print >>sys.stderr, "===============test_constructor(pyNetAddrTest)"
55  if DEBUG: print >>sys.stderr, "===============test_constructor(pyNetAddrTest)"
56  if DEBUG: print >>sys.stderr, "===============test_constructor(pyNetAddrTest)"
57  ipv4 = pyNetAddr((1,2,3,4),)
58  ipv4b = pyNetAddr((1,2,3,5),)
59  mac48 = pyNetAddr((1,2,3,4,5,6),)
60  mac64 = pyNetAddr( (1,2,3,4,5,6,7,8),)
61  ipv6 = pyNetAddr((1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16),)
62  self.assertEqual(str(ipv4), "1.2.3.4")
63  self.assertEqual(str(ipv4b), "1.2.3.5")
64  self.assertEqual(str(mac48), "01:02:03:04:05:06")
65  self.assertEqual(str(mac64), "01:02:03:04:05:06:07:08")
66  self.assertFalse(ipv4 != ipv4)
67  self.assertTrue(ipv4 == ipv4)
68  self.assertTrue(mac48 == mac48)
69  self.assertTrue(mac64 == mac64)
70  self.assertFalse(ipv4 == ipv4b)
71  self.assertFalse(ipv4 == mac48)
72  self.assertFalse(mac48 == ipv4)
73  self.assertFalse(ipv4 == mac64)
74  self.assertFalse(mac64 == ipv4)
75  self.assertFalse(mac48 == mac64)
76  self.assertFalse(mac64 == mac48)
77  self.assertRaises(ValueError, pyNetAddr, (1,))
78  self.assertRaises(ValueError, pyNetAddr, (1,2,))
79  self.assertRaises(ValueError, pyNetAddr, (1,2,3))
80  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5))
81  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7))
82  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9))
83  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10))
84  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11))
85  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12))
86  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13))
87  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13,14))
88  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15))
89  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17))
90 
91  def test_ipv6_str(self):
92  'Test the str() function for ipv6 - worth a separate test.'
93  if DEBUG: print >>sys.stderr, "===============test_ipv6_str(pyNetAddrTest)"
94  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0),)
95  self.assertEqual(str(ipv6),"::")
96  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,01),)
97  self.assertEqual(str(ipv6),"::1")
98  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,0,0,0,0,01,02),)
99  self.assertEqual(str(ipv6),"::102")
100  ipv6 = pyNetAddr((0,0,0,1,0,2,0,3,0,4,0,5,0,6,0,7),)
101  self.assertEqual(str(ipv6),"0:1:2:3:4:5:6:7")
102  ipv6 = pyNetAddr((0,0,0,0,0,2,0,3,0,4,0,5,0,6,0,7),)
103  self.assertEqual(str(ipv6),"::2:3:4:5:6:7")
104  # Example below is from http://en.wikipedia.org/wiki/IPv6_address
105  # Note that we now convert it into the equivalent IPv4 address as
106  # suggested: ::ffff:192.0.2.128
107  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,255,255,192,0,2,128),)
108  self.assertEqual(str(ipv6),"::ffff:192.0.2.128")
109 
110  @class_teardown
111  def tearDown(self):
112  assert_no_dangling_Cclasses()
113 
114 class pyFrameTest(TestCase):
115  '''Frames are our basic superclass for things we put on the wire.
116  This base class just has a generic binary blob with no special
117  properties. They are all valid (if they have a value)'''
118  def test_constructor(self):
119  if DEBUG: print >>sys.stderr, "===============test_constructor(pyFrameTest)"
120  pyf = pyFrame(100)
121  self.assertEqual(pyf.frametype(), 100)
122  self.assertTrue(pyf.isvalid())
123 
124  def test_setvalue(self):
125  if DEBUG: print >>sys.stderr, "===============test_setvalue(pyFrameTest)"
126  pyf = pyFrame(101)
127  pyf.setvalue('fred')
128  self.assertTrue(pyf.isvalid(), "PyFrame('fred') failed isvalid())")
129  self.assertEqual(pyf.framelen(), 5)
130  self.assertEqual(pyf.dataspace(), 9) # Total space for this Frame on the wire
131  self.assertEqual(string_at(pyf.framevalue()), 'fred') # Raw from 'C'
132 
133  @class_teardown
134  def tearDown(self):
135  assert_no_dangling_Cclasses()
136 
137 class pyAddrFrameTest(TestCase):
138  'An AddrFrame wraps a NetAddr for sending on the wire'
139  def test_constructor(self):
140  if DEBUG: print >>sys.stderr, "===============test_constructor(pyAddrFrameTest)"
141  pyf = pyAddrFrame(200, addrstring=(1,2,3,4))
142  self.assertEqual(pyf.frametype(), 200)
143  self.assertEqual(pyf.framelen(), 6)
144  self.assertEqual(str(pyf), 'pyAddrFrame(200, (1.2.3.4))')
145  self.assertEqual(pyf.addrtype(), 1)
146  self.assertTrue(pyf.isvalid(), "AddrFrame(200, (1,2,3,4)) failed isvalid()")
147  self.assertRaises(ValueError, pyAddrFrame, 201, addrstring=(1,2,3))
148 
149  @class_teardown
150  def tearDown(self):
151  assert_no_dangling_Cclasses()
152 
153 
154 class pyIpPortFrameTest(TestCase):
155  'An AddrFrame wraps a NetAddr *with a port* for sending on the wire'
156  def test_constructor(self):
157  if DEBUG: print >>sys.stderr, "===============test_constructor(pyIpAddrFrameTest)"
158  pyf = pyIpPortFrame(200, (1,2,3,4), 1984)
159  self.assertEqual(pyf.frametype(), 200)
160  self.assertEqual(pyf.framelen(), 8)
161  self.assertEqual(str(pyf), 'pyIpPortFrame(200, (1.2.3.4:1984))')
162  self.assertEqual(pyf.getnetaddr(), pyNetAddr('1.2.3.4:1984'))
163  self.assertEqual(pyf.addrtype(), 1)
164  self.assertTrue(pyf.isvalid(), "pyIpPortFrame(200, (1,2,3,4:1984)) failed isvalid()")
165  self.assertRaises(ValueError, pyIpPortFrame, 201, (1,2,3),80)
166  pyf = pyIpPortFrame(202, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16), 1984)
167  self.assertEqual(pyf.frametype(), 202)
168  self.assertEqual(pyf.framelen(), 20)
169  self.assertEqual(pyf.addrtype(), 2)
170  self.assertTrue(pyf.isvalid(), 'pyIpPortFrame(202, ([102:304:506:708:90a:b0c:d0e:f10]:1984))')
171  self.assertEqual(str(pyf), 'pyIpPortFrame(202, ([102:304:506:708:90a:b0c:d0e:f10]:1984))')
172  sameaddr = pyNetAddr([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10], port=1984)
173  self.assertEqual(pyf.getnetaddr(), sameaddr)
174  pyf = pyIpPortFrame(202, sameaddr, None)
175  self.assertEqual(str(pyf), 'pyIpPortFrame(202, ([102:304:506:708:90a:b0c:d0e:f10]:1984))')
176  self.assertEqual(pyf.getnetaddr(), sameaddr)
177 
178 
179  @class_teardown
180  def tearDown(self):
181  assert_no_dangling_Cclasses()
182 
183 class pyIntFrameTest(TestCase):
184  'An IntFrame wraps various sizes of unsigned integers for sending on the wire'
185  def test_constructor(self):
186  if DEBUG: print >>sys.stderr, "========================test_constructor(pyIntFrameTest)"
187  # Test a variety of illegal/unsupported integer sizes
188  for size in (5, 6, 7, 9, 10):
189  self.assertRaises(ValueError, pyIntFrame, 300+size-1, intbytes=size)
190  # Some network protocols have 3-byte integers. So I implemented them.
191  for size in (1, 2, 3, 4, 8):
192  pyf = pyIntFrame(310+size, initval=42, intbytes=size)
193  self.assertTrue(pyf.isvalid())
194  self.assertEqual(pyf.intlength(), size)
195  self.assertEqual(int(pyf), 42)
196  self.assertEqual(str(pyf), 'pyIntFrame(%d, (42))' % (310+size))
197 
198  def test_set(self):
199  'Test setting integer values for all the size integers'
200  if DEBUG: print >>sys.stderr, "========================test_set(pyIntFrameTest)"
201  for size in (1, 2, 3, 4, 8):
202  pyf = pyIntFrame(320, initval=0, intbytes=size)
203  val = 42 + size
204  pyf.setint(val)
205  self.assertEqual(int(pyf), val)
206  self.assertEqual(str(pyf), ('pyIntFrame(320, (%d))' % val))
207 
208 
209  @class_teardown
210  def tearDown(self):
211  assert_no_dangling_Cclasses()
212 
213 class pyUnknownFrameTest(TestCase):
214  "An unknown frame is one we don't recognize the type of."
215  def test_constructor(self):
216  if DEBUG: print >>sys.stderr, "========================test_constructor(pyUnknownFrameTest)"
217  pyf = pyUnknownFrame(400)
218  self.assertEqual(pyf.frametype(), 400)
219  # All Unknown frames are invalid...
220  self.assertFalse(pyf.isvalid(), "pyUnkownFrame(400) should not have passed isvalid()")
221 
222  @class_teardown
223  def tearDown(self):
224  assert_no_dangling_Cclasses()
225 
226 
227 class pySeqnoFrameTest(TestCase):
228  'A SeqnoFrame is a frame wrapping an ordered pair for a sequence number'
229  def test_constructor(self):
230  if DEBUG: print >>sys.stderr, "========================test_constructor(pySeqnoFrameTest)"
231  pyf = pySeqnoFrame(500)
232  self.assertEqual(pyf.frametype(), 500)
233  self.assertTrue(pyf.isvalid(), 'pySeqnoFrame(500) did not pass isvalid()')
234  pyf = pySeqnoFrame(501,(1,2))
235  self.assertEqual(pyf.frametype(), 501)
236  self.assertTrue(pyf.isvalid(), 'pySeqnoFrame(501) did not pass isvalid()')
237 
238  def test_reqid(self):
239  'reqid is the request id of a sequence number'
240  if DEBUG: print >>sys.stderr, "========================test_reqid(pySeqnoFrameTest)"
241  pyf = pySeqnoFrame(502)
242  pyf.setreqid(42)
243  self.assertTrue(pyf.getreqid, 42)
244  pyf.setreqid(43)
245  self.assertTrue(pyf.getreqid, 43)
246 
247  def test_qid(self):
248  'qid is analogous to a port - it is the id of a queue on the other side'
249  if DEBUG: print >>sys.stderr, "========================test_qid(pySeqnoFrameTest)"
250  pyf = pySeqnoFrame(503)
251  pyf.setqid(6)
252  self.assertTrue(pyf.getqid, 6)
253  pyf.setqid(7)
254  self.assertTrue(pyf.getqid, 7)
255 
256  def test_equal(self):
257  'A bit of overkill, but nothing really wrong with it'
258  if DEBUG: print >>sys.stderr, "========================test_equal(pySeqnoFrameTest)"
259  seqFrame1 = pySeqnoFrame( 504, (1,1))
260  seqFrame1b = pySeqnoFrame(505, (1,1))
261  seqFrame2 = pySeqnoFrame( 506, (1,2))
262  seqFrame3 = pySeqnoFrame( 507, (2,1))
263  seqFrame4 = pySeqnoFrame( 508, (2,2))
264  seqFrame4b = pySeqnoFrame(509, (2,2))
265  self.assertTrue(seqFrame1 == seqFrame1)
266  self.assertTrue(seqFrame1 == seqFrame1b)
267  self.assertTrue(seqFrame1b == seqFrame1)
268  self.assertFalse(seqFrame1 == seqFrame2)
269  self.assertFalse(seqFrame1 == seqFrame3)
270  self.assertFalse(seqFrame1 == seqFrame4)
271  self.assertFalse(seqFrame1 == seqFrame4b)
272  self.assertFalse(seqFrame2 == seqFrame1)
273  self.assertFalse(seqFrame2 == seqFrame1b)
274  self.assertFalse(seqFrame2 == seqFrame1)
275  self.assertTrue (seqFrame2 == seqFrame2)
276  self.assertFalse(seqFrame2 == seqFrame3)
277  self.assertFalse(seqFrame2 == seqFrame4)
278  self.assertFalse(seqFrame1 == seqFrame4b)
279  self.assertFalse(seqFrame3 == seqFrame1)
280  self.assertFalse(seqFrame3 == seqFrame1b)
281  self.assertFalse(seqFrame3 == seqFrame1)
282  self.assertFalse(seqFrame3 == seqFrame2)
283  self.assertTrue (seqFrame3 == seqFrame3)
284  self.assertFalse(seqFrame3 == seqFrame4)
285  self.assertFalse(seqFrame3 == seqFrame4b)
286  self.assertFalse(seqFrame4 == seqFrame1)
287  self.assertFalse(seqFrame4 == seqFrame1b)
288  self.assertFalse(seqFrame4 == seqFrame1)
289  self.assertFalse(seqFrame4 == seqFrame2)
290  self.assertFalse(seqFrame4 == seqFrame3)
291  self.assertTrue (seqFrame4 == seqFrame4)
292  self.assertTrue (seqFrame4 == seqFrame4b)
293  self.assertTrue(seqFrame4b == seqFrame4)
294 
295  @class_teardown
296  def tearDown(self):
297  assert_no_dangling_Cclasses()
298 
299 
300 class pyCstringFrameTest(TestCase):
301  '''A CstringFrame is a frame which can only hold NUL-terminated C strings.
302  The last byte must be the one and only NUL character in a CstringFrame value.'''
303  def test_constructor(self):
304  if DEBUG: print >>sys.stderr, "========================test_constructor(pyCstringFrameTest)"
305  pyf = pyCstringFrame(600, "Hello, World.")
306  self.assertTrue(pyf.isvalid())
307  self.assertEqual(str(pyf), '600: CstringFrame(600, "Hello, World.")')
308  pyf2 = pyCstringFrame(601)
309  self.assertFalse(pyf2.isvalid())
310  pyf2.setvalue("42")
311  self.assertTrue(pyf2.isvalid())
312  self.assertEqual(str(pyf2), '601: CstringFrame(601, "42")')
313 
314  @class_teardown
315  def tearDown(self):
316  assert_no_dangling_Cclasses()
317 
318 class pySignFrameTest(TestCase):
319  'A SignFrame is a digital signature frame.'
320  def test_constructor(self):
321  if DEBUG: print >>sys.stderr, "========================test_constructor(pySignFrameTest)"
322  pyf = pySignFrame(1) # the 1 determines the type of digital signature
323  self.assertTrue(pyf.isvalid())
324  self.assertRaises(ValueError, pySignFrame, 935) # Just a random invalid signature type
325 
326  @class_teardown
327  def tearDown(self):
328  assert_no_dangling_Cclasses()
329 
330 
331 class pyFrameSetTest(TestCase):
332  'A FrameSet is a collection of frames - typically to be sent over the wire'
333 
334  @staticmethod
335  def cmpstring(frame):
336  s=str(frame)
337  s = re.sub(' at 0x[^{}]*', ' at 0xsomewhere', s)
338  return s
339 
340  def test_constructor(self):
341  if DEBUG: print >>sys.stderr, "========================test_constructor(pyFrameSetTest)"
342  pyf = pyFrameSet(700) # The 700 is the frameset (message) type
343  self.assertEqual(pyf.get_framesettype(), 700)
344 
345  def test_flags(self):
346  if DEBUG: print >>sys.stderr, "========================test_flags(pyFrameSetTest)"
347  'Flags are bit masks, to be turned on or off. They are 16-bits only.'
348  pyf = pyFrameSet(701)
349  self.assertEqual(pyf.get_flags(), 0x00)
350  pyf.set_flags(0x01)
351  self.assertEqual(pyf.get_flags(), 0x01)
352  pyf.set_flags(0x01)
353  self.assertEqual(pyf.get_flags(), 0x01)
354  pyf.set_flags(0x02)
355  self.assertEqual(pyf.get_flags(), 0x03)
356  pyf.clear_flags(0x01)
357  self.assertEqual(pyf.get_flags(), 0x02)
358  pyf.set_flags(0x0fffffffffffffffff)
359  self.assertEqual(pyf.get_flags(), 0x0ffff)
360  pyf.clear_flags(0x5555)
361  self.assertEqual(pyf.get_flags(), 0x0AAAA)
362 
363  def test_buildlistforward(self):
364  'Build a FrameSet using append and verify that it gets built right'
365  if DEBUG: print >>sys.stderr, "========================test_buildlistforward(pyFrameSetTest)"
366  pyfs = pyFrameSet(702)
367  sign = pySignFrame(1) # digital signature frame
368  flist = (pyFrame(703), pyAddrFrame(704, (42,42,42,42)), pyIntFrame(705,42),
369  pyCstringFrame(706, "HhGttG"),
370  pySeqnoFrame(707, (42, 424242424242)),
371  pyIpPortFrame(200, (1,2,3,4), 1984),
372  pyIpPortFrame(202, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16), 1984)
373  )
374  for frame in flist:
375  pyfs.append(frame)
376  self.assertEqual(len(pyfs), 7)
377  #pyfs.dump()
378  ylist = []
379  # The iter member function is a generator. I love it.
380  # It goes to a lot of trouble to wrap the underlying C Classes with Python classes.
381  for frame in pyfs.iter():
382  ylist.append(frame)
383  for i in range(0,len(ylist)):
384  f=flist[i]
385  y=ylist[i]
386  # This isn't exhaustive, but it isn't bad.
387  self.assertEqual(f.frametype(), y.frametype())
388  self.assertEqual(type(f), type(y))
389  # Constructing the packet will add a signature frame at the beginning
390  # and an END (type 0) frame at the end
391  pyfs.construct_packet(sign)
392  # So we do it over again to make sure everything still looks OK
393  ylist = []
394  for frame in pyfs.iter():
395  ylist.append(frame)
396  self.assertEqual(len(pyfs), 9)
397  self.assertEqual(len(ylist), len(pyfs)) # len(pyfs) traverses the linked list
398  for i in range(0,len(flist)):
399  f=flist[i]
400  y=ylist[i+1]
401  # This isn't exhaustive, but it isn't bad.
402  self.assertEqual(f.frametype(), y.frametype())
403  self.assertEqual(type(f), type(y))
404  # Check on our automatically added frames.
405  self.assertEqual(ylist[0].frametype(), 1)
406  self.assertEqual(ylist[8].frametype(), 0)
407 
408  def test_buildlistbackwards(self):
409  '''Build a FrameSet using prepend and verify that it gets built right.
410  Similar to the append testing above, but backwards ;-)'''
411  if DEBUG: print >>sys.stderr, "========================test_buildlistbackwards(pyFrameSetTest)"
412  pyfs = pyFrameSet(707)
413  sign = pySignFrame(1)
414  flist = (pyFrame(708), pyAddrFrame(709, (42,42,42,42)), pyIntFrame(710,42), pyCstringFrame(711, "HhGttG"),
415  pySeqnoFrame(712, (42, 424242424242)))
416  for frame in flist:
417  pyfs.prepend(frame)
418  self.assertEqual(len(pyfs), 5)
419  #pyfs.dump()
420  ylist = []
421  for frame in pyfs.iter():
422  ylist.append(frame)
423  for i in range(0,len(flist)):
424  f=flist[i]
425  y=ylist[4-i]
426  # This isn't exhaustive, but it isn't bad.
427  self.assertEqual(f.frametype(), y.frametype())
428  self.assertEqual(type(f), type(y))
429  self.assertEqual(f.__class__, y.__class__)
430  if DEBUG: print >>sys.stderr, "Classes are", f.__class__, "lens are", f.framelen(), y.framelen()
431  self.assertEqual(f.framelen(), y.framelen())
432 
433  def test_buildpacket(self):
434  'Build a FrameSet, then make it into a packet, and make a frameset list out of the packet'
435  if DEBUG: print >>sys.stderr, "========================test_buildpacket(pyFrameSetTest)"
436  pyfs = pyFrameSet(801)
437  sign = pySignFrame(1) # digital signature frame
438  flist = (pyAddrFrame(FrameTypes.IPADDR, (42,42,42,42)), pyIntFrame(FrameTypes.WALLCLOCK,42), pyCstringFrame(FrameTypes.INTERFACE, "HhGttG"),
439  pyIntFrame(FrameTypes.CINTVAL,3000000, intbytes=4),
440  pyIntFrame(FrameTypes.CINTVAL,3000000000000, intbytes=8),
441  pySeqnoFrame(FrameTypes.REQID, (42, 424242424242)),
442  pyIntFrame(FrameTypes.CINTVAL,4242, intbytes=3))
443  if DEBUG: print >>sys.stderr, "flist:", flist
444  decoder = pyPacketDecoder(0)
445  for frame in flist:
446  pyfs.append(frame)
447  pyfs.construct_packet(sign)
448  if DEBUG: print >>sys.stderr, "packet constructed"
449  xlist=[]
450  for frame in pyfs.iter():
451  xlist.append(frame)
452  if DEBUG: print >>sys.stderr, "xlist constructed"
453  pktdata = pyfs.getpacket()
454  if DEBUG: print >>sys.stderr, "getpacket done", pktdata
455  cp_pyfs = decoder.fslist_from_pktdata(pktdata)
456  if DEBUG: print >>sys.stderr, "decoder done", cp_pyfs
457  fs0 = cp_pyfs[0]
458  ylist=[]
459  for frame in fs0.iter():
460  ylist.append(frame)
461  for i in range(0,len(xlist)):
462  x=xlist[i]
463  y=ylist[i]
464  self.assertEqual(x.frametype(), y.frametype())
465  self.assertEqual(x.framelen(), y.framelen())
466  self.assertEqual(x.dataspace(), y.dataspace())
467  self.assertEqual(type(x), type(y))
468  self.assertEqual(x.__class__, y.__class__)
469  # Not all our classes have a __str__ method defined.
470  strx = re.sub(str(x), ' instance at .*>', ' instance at -somewhere- >')
471  stry = re.sub(str(y), ' instance at .*>', ' instance at -somewhere- >')
472  self.assertEqual(strx, stry)
473  self.assertEqual(pyFrameSetTest.cmpstring(x), pyFrameSetTest.cmpstring(y))
474 
475 
476  @class_teardown
477  def tearDown(self):
478  assert_no_dangling_Cclasses()
479 
480 class pyConfigContextTest(TestCase):
481 
482  def test_constructor(self):
483  pyConfigContext()
484  if DEBUG: print >>sys.stderr, "===============test_constructor(pyConfigContextTest)"
485  foo = pyConfigContext(init={'int1': 42, 'str1': 'forty-two', 'bar': pyNetAddr((1,2,3,4),) })
486  foo = pyConfigContext(init={'int1': 42, 'str1': 'forty-two', 'bar': pyNetAddr((1,2,3,4),), 'csf': pyCstringFrame(42, '41+1')})
487  self.assertEqual(foo.getint('int1'), 42)
488  self.assertEqual(foo.getstring('str1'), 'forty-two')
489  self.assertRaises(IndexError, foo.getaddr, ('int1'))
490  self.assertRaises(IndexError, foo.getstring, ('int1'))
491  self.assertRaises(IndexError, foo.getaddr, ('str1'))
492  self.assertRaises(IndexError, foo.getframe, ('str1'))
493  self.assertRaises(IndexError, foo.getframe, ('int1'))
494  self.assertEqual(foo['int1'], 42)
495  self.assertEqual(foo['str1'], 'forty-two')
496  self.assertEqual(foo.getint('fred'), -1)
497  foo['bar']
498  self.assertEqual(foo['bar'], pyNetAddr((1,2,3,4),))
499  self.assertEqual(str(foo['bar']), '1.2.3.4')
500  self.assertEqual(str(foo['csf']), '42: CstringFrame(42, "41+1")')
501  self.assertEqual(str(foo), '{"str1":"forty-two","csf":"CstringFrame(42, \\"41+1\\")","int1":42,"bar":"1.2.3.4"}')
502 
503  foo['isf'] = pyIntFrame(310, initval=42, intbytes=3)
504  if DEBUG: print >>sys.stderr, "test_constructor.18(pyConfigContextTest)"
505  self.assertEqual(str(foo),
506  '{"isf":"IntFrame(310, 3, 42)","str1":"forty-two","csf":"CstringFrame(42, \\"41+1\\")","int1":42,"bar":"1.2.3.4"}')
507  if DEBUG: print >>sys.stderr, "test_constructor.19(pyConfigContextTest)"
508 
509  def test_string(self):
510  if DEBUG: print >>sys.stderr, "test_string(pyConfigContextTest)"
511  foo = pyConfigContext()
512  foo['arthur'] = 'dent'
513  foo['seven'] = 'ofnine'
514  foo['JeanLuc'] = 'Picard'
515  foo['important'] = 'towel'
516  foo['integer'] = 42
517  self.assertEqual(foo['arthur'], 'dent')
518  self.assertEqual(foo['seven'], 'ofnine')
519  self.assertEqual(foo['JeanLuc'], 'Picard')
520  self.assertEqual(foo['important'], 'towel')
521  self.assertRaises(IndexError, foo.getstring, ('towel'))
522  self.assertEqual(str(foo), '{"arthur":"dent","JeanLuc":"Picard","important":"towel","integer":42,"seven":"ofnine"}')
523  foo['seven'] = '7'
524  self.assertEqual(foo['seven'], '7')
525  self.assertEqual(type(foo['seven']), str)
526  foo['JeanLuc'] = 'Locutus'
527  self.assertEqual(foo['JeanLuc'], 'Locutus')
528  self.assertEqual(str(foo), '{"arthur":"dent","JeanLuc":"Locutus","important":"towel","integer":42,"seven":"7"}')
529 
530  def test_int(self):
531  if DEBUG: print >>sys.stderr, "test_int(pyConfigContextTest)"
532  foo = pyConfigContext()
533  foo['arthur'] = 42
534  foo['seven'] = 9
535  self.assertEqual(foo['arthur'], 42)
536  self.assertEqual(foo['seven'], 9)
537  self.assertEqual(str(foo), '{"arthur":42,"seven":9}')
538  foo['seven'] = 7
539  self.assertEqual(type(foo['seven']), int)
540  self.assertEqual(foo["seven"], 7)
541  self.assertEqual(str(foo), '{"arthur":42,"seven":7}')
542 
543  def test_child_ConfigContext(self):
544  if DEBUG: print >>sys.stderr, "========================test_child_ConfigContext(pyConfigContextTest)"
545  foo = pyConfigContext()
546  foo['ford'] = 'prefect'
547  baz = pyConfigContext()
548  baz['Kathryn'] = 'Janeway'
549  baz['there\'s no place like'] = pyNetAddr((127,0,0,1),)
550  bar = pyConfigContext()
551  bar['hhgttg'] = foo
552  bar['voyager'] = baz
553  if DEBUG: print >>sys.stderr, "EQUAL TEST"
554  self.assertEqual(str(bar), '{"hhgttg":{"ford":"prefect"},"voyager":{"there\'s no place like":"127.0.0.1","Kathryn":"Janeway"}}')
555  # We make a new pyConfigContext object from the str() of another one. Cool!
556  if DEBUG: print >>sys.stderr, "JSON TEST"
557  bar2 = pyConfigContext(str(bar))
558  if DEBUG: print >>sys.stderr, "JSON COMPARE"
559  self.assertEqual(str(bar), str(bar2))
560  self.assertEqual(bar["voyager"]["Kathryn"], "Janeway")
561  self.assertEqual(bar["hhgttg"]["ford"], "prefect")
562  self.assertEqual(bar2["voyager"]["Kathryn"], "Janeway")
563  self.assertEqual(bar2["hhgttg"]["ford"], "prefect")
564  if DEBUG: print >>sys.stderr, "COMPARE DONE"
565  # However... The pyNetAddr() was turned into a mere string :-( - at least for the moment... Sigh...
566  if DEBUG: print >>sys.stderr, "END OF ========================test_child_ConfigContext(pyConfigContextTest)"
567 
568 
569  def test_keys(self):
570  if DEBUG: print >>sys.stderr, "===============test_keys(pyConfigContextTest)"
571  foo = pyConfigContext()
572  foo['arthur'] = 'dent'
573  foo['seven'] = 'ofnine'
574  foo['JeanLuc'] = 'Picard'
575  foo['important'] = 'towel'
576  foo['integer'] = 42
577  self.assertEqual(str(foo.keys()), "['JeanLuc', 'arthur', 'important', 'integer', 'seven']")
578 
579  def test_ConfigContext_array(self):
580  array1str = '{"a":[1,2,3,4,"a",{"b":true},[5,6,7,8,3.14]]}'
581  array1config = pyConfigContext(array1str)
582  self.assertEqual(array1str, str(array1config))
583 
584 
585  @class_teardown
586  def tearDown(self):
587  assert_no_dangling_Cclasses()
588 
589 class pyNetIOudpTest(TestCase):
590 
591  def test_constructor(self):
592  if DEBUG: print >>sys.stderr, "========================test_constructor(pyNetIOudpTest)"
593  config = pyConfigContext(init={'outsig': pySignFrame(1)})
594  io = pyNetIOudp(config, pyPacketDecoder(0))
595  self.assertTrue(io.getfd() > 2)
596 
597  def test_members(self):
598  if DEBUG: print >>sys.stderr, "========================test_members(pyNetIOudpTest)"
599  io = pyNetIOudp(pyConfigContext(init={'outsig': pySignFrame(1)}), pyPacketDecoder(0))
600  self.assertTrue(io.getmaxpktsize() > 65000)
601  self.assertTrue(io.getmaxpktsize() < 65535)
602  io.setmaxpktsize(1500)
603  self.assertEqual(io.getmaxpktsize(), 1500)
604  # Does signframe really work? Next statement seems to crash things
605  #self.assertEqual(type(io.signframe()), type(pySignFrame(1)))
606 
607  def test_send(self):
608  if DEBUG: print >>sys.stderr, "========================test_send(pyNetIOudpTest)"
609  home = pyNetAddr((127,0,0,1),1984)
610  fs = pyFrameSet(801)
611  flist = (pyAddrFrame(FrameTypes.IPADDR, (42,42,42,42)), pyIntFrame(FrameTypes.HBWARNTIME,42), pyCstringFrame(FrameTypes.HOSTNAME, "HhGttG"),
612  pyIntFrame(FrameTypes.PORTNUM,3000000, intbytes=4),
613  pyIntFrame(FrameTypes.HBINTERVAL,3000000000000, intbytes=8),
614  pySeqnoFrame(FrameTypes.REPLYID, (42, 424242424242)),
615  pyIntFrame(FrameTypes.CINTVAL,4242, intbytes=3))
616  for frame in flist:
617  fs.append(frame)
618  io = pyNetIOudp(pyConfigContext(init={'outsig': pySignFrame(1)}), pyPacketDecoder(0))
619  io.sendframesets(home, fs)
620  io.sendframesets(home, (fs,fs,fs))
621 
622  def test_receiveandsend(self):
623  if DEBUG: print >>sys.stderr, "========================test_receiveandsend(pyNetIOudpTest)"
624  home = pyNetAddr((127,0,0,1),1984)
625  anyaddr = pyNetAddr((0,0,0,0),1984)
626  fs = pyFrameSet(801)
627  #flist = (pyIntFrame(7,42), pyCstringFrame(8, "HhGttG"),
628  flist = (pyAddrFrame(FrameTypes.IPADDR, (42,42,42,42)), pyIntFrame(7,42), pyCstringFrame(8, "HhGttG"),
629  pyAddrFrame(FrameTypes.IPADDR,(1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16)),
630  pyIntFrame(FrameTypes.HBINTERVAL,3000000, intbytes=4),
631  pyIntFrame(FrameTypes.HBDEADTIME,3000000000000, intbytes=8),
632  pySeqnoFrame(FrameTypes.REPLYID, (42, 424242424242)),
633  pyIntFrame(FrameTypes.PORTNUM,4242, intbytes=3))
634  for frame in flist:
635  fs.append(frame)
636  io = pyNetIOudp(pyConfigContext(init={'outsig': pySignFrame(1)}), pyPacketDecoder(0))
637  io.bindaddr(anyaddr)
638  io.sendframesets(home, fs) # Send a packet with a single frameset containing a bunch of frames
639  (addr, framesetlist) = io.recvframesets() # Receive a packet - with some framesets in it
640  #print >>sys.stderr, 'ADDR: [%s] HOME: [%s]' % (addr, home)
641  self.assertEqual(addr, home)
642  self.assertEqual(len(framesetlist), 1)
643  ylist = []
644  for frame in framesetlist[0].iter():
645  ylist.append(frame)
646  self.assertEqual(len(flist), len(ylist)-2)
647  for i in range(0,len(flist)):
648  x=flist[i]
649  y=ylist[i+1]
650  self.assertEqual(x.frametype(), y.frametype())
651  self.assertEqual(x.framelen(), y.framelen())
652  self.assertEqual(x.dataspace(), y.dataspace())
653  self.assertEqual(type(x), type(y))
654  self.assertEqual(x.__class__, y.__class__)
655  self.assertEqual(pyFrameSetTest.cmpstring(x), pyFrameSetTest.cmpstring(y))
656 
657  @class_teardown
658  def tearDown(self):
659  assert_no_dangling_Cclasses()
660 
661 if __name__ == "__main__":
662  run()