The Assimilation Monitoring Project
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
cma/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 import traceback
25 #traceback.print_exc()
26 sys.path.append("../cma")
27 from testify import *
28 
29 from frameinfo import *
30 from AssimCclasses import *
31 import gc
32 import re
33 from AssimCtypes import proj_class_incr_debug, proj_class_decr_debug
34 
35 
36 CheckForDanglingClasses = True
37 WorstDanglingCount = 0
38 DEBUG=False
39 
40 def assert_no_dangling_Cclasses():
41  global CheckForDanglingClasses
42  global WorstDanglingCount
43  gc.collect()
44  count = proj_class_live_object_count()
45  # Avoid cluttering the output up with redundant messages...
46  if count > WorstDanglingCount and CheckForDanglingClasses:
47  WorstDanglingCount = count
48  proj_class_dump_live_objects()
49  raise AssertionError, "Dangling C-class objects - %d still around" % count
50 
51 class pyNetAddrTest(TestCase):
52  "A pyNetAddr is a network address of some kind... - let's test it"
53  def test_constructor(self):
54  if DEBUG: print >>sys.stderr, "===============test_constructor(pyNetAddrTest)"
55  ipv4 = pyNetAddr((1,2,3,4),)
56  ipv4b = pyNetAddr((1,2,3,5),)
57  mac48 = pyNetAddr((1,2,3,4,5,6),)
58  mac64 = pyNetAddr( (1,2,3,4,5,6,7,8),)
59  ipv6 = pyNetAddr((1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16),)
60  self.assertEqual(str(ipv4), "1.2.3.4")
61  self.assertEqual(str(ipv4b), "1.2.3.5")
62  self.assertEqual(str(mac48), "01:02:03:04:05:06")
63  self.assertEqual(str(mac64), "01:02:03:04:05:06:07:08")
64  self.assertFalse(ipv4 != ipv4)
65  self.assertTrue(ipv4 == ipv4)
66  self.assertTrue(mac48 == mac48)
67  self.assertTrue(mac64 == mac64)
68  self.assertFalse(ipv4 == ipv4b)
69  self.assertFalse(ipv4 == mac48)
70  self.assertFalse(mac48 == ipv4)
71  self.assertFalse(ipv4 == mac64)
72  self.assertFalse(mac64 == ipv4)
73  self.assertFalse(mac48 == mac64)
74  self.assertFalse(mac64 == mac48)
75  self.assertRaises(ValueError, pyNetAddr, (1,))
76  self.assertRaises(ValueError, pyNetAddr, (1,2,))
77  self.assertRaises(ValueError, pyNetAddr, (1,2,3))
78  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5))
79  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7))
80  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9))
81  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10))
82  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11))
83  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12))
84  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13))
85  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13,14))
86  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15))
87  self.assertRaises(ValueError, pyNetAddr, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17))
88 
89  def test_ipv4_eq(self):
90  'Test if various ipv4 addresses are equal'
91  self.assertEqual(pyNetAddr((1,2,3,4),80), pyNetAddr('1.2.3.4:80'))
92  self.assertEqual(pyNetAddr((1,2,3,4),), pyNetAddr('1.2.3.4'),)
93  self.assertEqual(pyNetAddr('1.2.3.4:80'), pyNetAddr((1,2,3,4),80))
94  self.assertNotEqual(pyNetAddr('1.2.3.4:80'), pyNetAddr((1,2,3,4),81))
95  self.assertNotEqual(pyNetAddr('1.2.3.4:80'), pyNetAddr((1,2,3,4),))
96 
97  def test_ipv6_eq(self):
98  'Test if various ipv6 addresses are equal'
99  self.assertEqual(pyNetAddr('::1'), pyNetAddr('::1'))
100  self.assertEqual(pyNetAddr('::'), pyNetAddr('::'))
101  self.assertNotEqual(pyNetAddr('::'), pyNetAddr('::1'))
102  self.assertNotEqual(pyNetAddr('::1'), pyNetAddr('[::1]:80'))
103  self.assertEqual(pyNetAddr('[::1]:1984'), pyNetAddr('[::1]:1984'))
104  self.assertNotEqual(pyNetAddr('::1:2:3:4:1'), pyNetAddr('::1'))
105 
106  def test_mixed_eq(self):
107  'Test if various ipv4/v6 addresses are equal to each other'
108  self.assertEqual(pyNetAddr('::1'), pyNetAddr('127.0.0.1'))
109  self.assertEqual(pyNetAddr('127.0.0.1'), pyNetAddr('::1'))
110  self.assertEqual(pyNetAddr('::ffff:7f00:1'), pyNetAddr('::1'))
111  self.assertEqual(pyNetAddr('::ffff:7f00:1'), pyNetAddr('127.0.0.1'))
112  self.assertEqual(pyNetAddr((0,0,0,0,0,0,0,0,0,0, 0xff, 0xff, 127, 0, 0, 1),), pyNetAddr('::1'))
113 
114  def test_mixed_hash(self):
115  'Test if various ipv4/v6 addresses hash to the same values - or not'
116  self.assertEqual(pyNetAddr('::1').__hash__(), pyNetAddr('::1').__hash__())
117  self.assertEqual(pyNetAddr('::1').__hash__(), pyNetAddr('127.0.0.1').__hash__())
118  self.assertEqual(pyNetAddr('[::1]:80').__hash__(), pyNetAddr('127.0.0.1:80').__hash__())
119  self.assertNotEqual(pyNetAddr('[::1]:80').__hash__(), pyNetAddr('127.0.0.1:1984').__hash__())
120  self.assertEqual(pyNetAddr('::ffff:7f00:1').__hash__(), pyNetAddr('::1').__hash__())
121  self.assertEqual(pyNetAddr('[::ffff:7f00:1]:25').__hash__(), pyNetAddr('[::1]:25').__hash__())
122  self.assertNotEqual(pyNetAddr('[::ffff:7f00:1]:25').__hash__(), pyNetAddr('[::1]:26').__hash__())
123  self.assertNotEqual(pyNetAddr('::').__hash__(), pyNetAddr('::1').__hash__())
124  self.assertEqual(pyNetAddr('::ffff:7f00:1').__hash__(), pyNetAddr('127.0.0.1').__hash__())
125  self.assertEqual(pyNetAddr((0,0,0,0,0,0,0,0,0,0, 0xff, 0xff, 127, 0, 0, 1),).__hash__(), pyNetAddr('::1').__hash__())
126 
127  def test_ipv6_str(self):
128  'Test the str() function for ipv6 - worth a separate test.'
129  if DEBUG: print >>sys.stderr, "===============test_ipv6_str(pyNetAddrTest)"
130  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0),)
131  self.assertEqual(str(ipv6),"::")
132  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,01),)
133  self.assertEqual(str(ipv6),"::1")
134  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,0,0,0,0,01,02),)
135  self.assertEqual(str(ipv6),"::102")
136  ipv6 = pyNetAddr((0,0,0,1,0,2,0,3,0,4,0,5,0,6,0,7),)
137  self.assertEqual(str(ipv6),"0:1:2:3:4:5:6:7")
138  ipv6 = pyNetAddr((0,0,0,0,0,2,0,3,0,4,0,5,0,6,0,7),)
139  self.assertEqual(str(ipv6),"::2:3:4:5:6:7")
140  # Example below is from http://en.wikipedia.org/wiki/IPv6_address
141  # Note that we now convert it into the equivalent IPv4 address as
142  # suggested: ::ffff:192.0.2.128
143  ipv6 = pyNetAddr((0,0,0,0,0,0,0,0,0,0,255,255,192,0,2,128),)
144  self.assertEqual(str(ipv6),"::ffff:192.0.2.128")
145 
146  def test_ipv4_strinit(self):
147  'Test constructing ipv4 addresses from strings.'
148  if DEBUG: print >>sys.stderr, "===============test_ipv4_strinit(pyNetAddrTest)"
149  ipv4 = pyNetAddr('1.2.3.4')
150  self.assertEqual(str(ipv4),'1.2.3.4')
151  ipv4 = pyNetAddr('1.2.3.5')
152  self.assertEqual(str(ipv4),'1.2.3.5')
153  ipv4 = pyNetAddr('1.2.3.4:80')
154  self.assertEqual(str(ipv4),'1.2.3.4:80')
155  ipv4 = pyNetAddr('1.2.3.5:80')
156  self.assertEqual(str(ipv4),'1.2.3.5:80')
157  self.assertRaises(ValueError, pyNetAddr, '1.2.ff.5')
158  self.assertRaises(ValueError, pyNetAddr, '1.2.3.4:')
159  self.assertRaises(ValueError, pyNetAddr, '1.2.3.4:ff')
160 
161  def test_toIPv6(self):
162  'Test converting IPv4 addresses to IPv6'
163  ipstr= '213.138.100.204:57225'
164  ipv4 = pyNetAddr(ipstr)
165  self.assertEqual(str(ipv4),ipstr)
166  ipv6 = ipv4.toIPv6()
167  self.assertEqual(str(ipv6),'[::ffff:213.138.100.204]:57225')
168  ipv6 = ipv6.toIPv6()
169  self.assertEqual(str(ipv6),'[::ffff:213.138.100.204]:57225')
170  ipv6 = pyNetAddr(str(ipv6))
171  self.assertEqual(str(ipv6),'[::ffff:213.138.100.204]:57225')
172  #
173  ipstr= '213.138.100.255:65535'
174  ipv4 = pyNetAddr(ipstr)
175  self.assertEqual(str(ipv4),ipstr)
176  ipv6 = ipv4.toIPv6()
177  self.assertEqual(str(ipv6),'[::ffff:213.138.100.255]:65535')
178  ipv6 = ipv6.toIPv6()
179  self.assertEqual(str(ipv6),'[::ffff:213.138.100.255]:65535')
180  ipv6 = pyNetAddr(str(ipv6))
181  self.assertEqual(str(ipv6),'[::ffff:213.138.100.255]:65535')
182  #
183  ipstr= '255.255.255.255:65535'
184  ipv4 = pyNetAddr(ipstr)
185  self.assertEqual(str(ipv4),ipstr)
186  ipv6 = ipv4.toIPv6()
187  self.assertEqual(str(ipv6),'[::ffff:255.255.255.255]:65535')
188  ipv6 = ipv6.toIPv6()
189  self.assertEqual(str(ipv6),'[::ffff:255.255.255.255]:65535')
190  ipv6 = pyNetAddr(str(ipv6))
191  self.assertEqual(str(ipv6),'[::ffff:255.255.255.255]:65535')
192 
193  def test_ipv6_strinit(self):
194  'Test constructing ipv6 addresses from strings.'
195  if DEBUG:
196  for j in range(1,5):
197  proj_class_incr_debug('NetAddr')
198  if DEBUG: print >>sys.stderr, "===============test_ipv6_strinit(pyNetAddrTest)"
199  ipv6 = pyNetAddr('::1')
200  self.assertEqual(str(ipv6),'::1')
201 
202  ipv6 = pyNetAddr('::')
203  self.assertEqual(str(ipv6),'::')
204 
205  ipv6 = pyNetAddr('0:1:2:3:4:5:6:7')
206  self.assertEqual(str(ipv6),'0:1:2:3:4:5:6:7')
207 
208  ipv6 = pyNetAddr('::2:3:4:5:6:7')
209  self.assertEqual(str(ipv6),'::2:3:4:5:6:7')
210 
211  ipv6 = pyNetAddr('[::]:1984')
212  self.assertEqual(str(ipv6),'[::]:1984')
213 
214  ipv6 = pyNetAddr('[::1]:80')
215  self.assertEqual(str(ipv6),'[::1]:80')
216 
217  ipv6 = pyNetAddr('[0:1:2:3:4:5:6:7]:8080')
218  self.assertEqual(str(ipv6),'[0:1:2:3:4:5:6:7]:8080')
219 
220  ipv6 = pyNetAddr('::2:3:4:5:6:7')
221  self.assertEqual(str(ipv6),'::2:3:4:5:6:7')
222 
223  ipv6 = pyNetAddr('::a:b:c:d:e:f')
224  self.assertEqual(str(ipv6),'::a:b:c:d:e:f')
225 
226  ipv6 = pyNetAddr('::ffff:1.2.3.4')
227  self.assertEqual(str(ipv6),'::ffff:1.2.3.4')
228 
229  ipv6 = pyNetAddr('[::ffff:1.2.3.4]:80')
230  self.assertEqual(str(ipv6),'[::ffff:1.2.3.4]:80')
231 
232  ipv6 = pyNetAddr('[::ffff:255.255.255.255]:65535')
233  self.assertEqual(str(ipv6), '[::ffff:255.255.255.255]:65535')
234 
235  self.assertRaises(ValueError, pyNetAddr, '0:1:2:3:4:5:6:7::')
236  self.assertRaises(ValueError, pyNetAddr, '::fffff')
237  self.assertRaises(ValueError, pyNetAddr, '[0:1:2:3:4:5:6:7]10')
238  self.assertRaises(ValueError, pyNetAddr, '[0:1:2:3:4:5:6:7]:ff')
239  self.assertRaises(ValueError, pyNetAddr, '0:1:2:3:4:5g:6:7')
240  self.assertRaises(ValueError, pyNetAddr, '[0:1:2:3:4:5g:6:7]:10')
241  self.assertRaises(ValueError, pyNetAddr, 'ffff:1.2.3.256')
242  self.assertRaises(ValueError, pyNetAddr, 'ffff:1.2.3:4')
243  self.assertRaises(ValueError, pyNetAddr, '[ffff:1.2.3:4]:99')
244  self.assertRaises(ValueError, pyNetAddr, '[ffff:1.2.3.4]:65536')
245  if DEBUG:
246  for j in range(1,5):
247  proj_class_decr_debug('NetAddr')
248 
249  @class_teardown
250  def tearDown(self):
251  assert_no_dangling_Cclasses()
252 
253 class pyFrameTest(TestCase):
254  '''Frames are our basic superclass for things we put on the wire.
255  This base class just has a generic binary blob with no special
256  properties. They are all valid (if they have a value)'''
257  def test_constructor(self):
258  if DEBUG: print >>sys.stderr, "===============test_constructor(pyFrameTest)"
259  pyf = pyFrame(100)
260  self.assertEqual(pyf.frametype(), 100)
261  self.assertTrue(pyf.isvalid())
262 
263  def test_setvalue(self):
264  if DEBUG: print >>sys.stderr, "===============test_setvalue(pyFrameTest)"
265  pyf = pyFrame(101)
266  pyf.setvalue('fred')
267  self.assertTrue(pyf.isvalid(), "PyFrame('fred') failed isvalid())")
268  self.assertEqual(pyf.framelen(), 5)
269  self.assertEqual(pyf.dataspace(), 9) # Total space for this Frame on the wire
270  self.assertEqual(string_at(pyf.framevalue()), 'fred') # Raw from 'C'
271 
272  @class_teardown
273  def tearDown(self):
274  assert_no_dangling_Cclasses()
275 
276 class pyAddrFrameTest(TestCase):
277  'An AddrFrame wraps a NetAddr for sending on the wire'
278  def test_constructor(self):
279  if DEBUG: print >>sys.stderr, "===============test_constructor(pyAddrFrameTest)"
280  pyf = pyAddrFrame(200, addrstring=(1,2,3,4))
281  self.assertEqual(pyf.frametype(), 200)
282  self.assertEqual(pyf.framelen(), 6)
283  self.assertEqual(str(pyf), 'pyAddrFrame(200, (1.2.3.4))')
284  self.assertEqual(pyf.addrtype(), 1)
285  self.assertTrue(pyf.isvalid(), "AddrFrame(200, (1,2,3,4)) failed isvalid()")
286  self.assertRaises(ValueError, pyAddrFrame, 201, addrstring=(1,2,3))
287 
288  @class_teardown
289  def tearDown(self):
290  assert_no_dangling_Cclasses()
291 
292 
293 class pyIpPortFrameTest1(TestCase):
294  'An IpPortFrame wraps a NetAddr (with port!) for sending on the wire'
295  def test_constructor(self):
296  if DEBUG: print >>sys.stderr, '===============test_constructor(pyIpPortFrameTest1)'
297  addrv4=pyNetAddr('1.2.3.4:1984')
298  py4 = pyIpPortFrame(201, addrv4)
299  self.assertEqual(py4.frametype(), 201)
300  self.assertEqual(py4.framelen(), 8)
301  self.assertEqual(str(py4), '201: IpPortFrame(201, 1.2.3.4:1984)')
302  self.assertEqual(py4.addrtype(), 1)
303  self.assertTrue(py4.isvalid(), 'pyIpPortFrame(201, (1.2.3.4:1984)) failed isvalid()')
304  self.assertRaises(ValueError, pyIpPortFrame, 201, pyNetAddr((1,2,3,4,5,6),)) # MAC address
305  addrv6=pyNetAddr('[::1]:79')
306  py6 = pyIpPortFrame(302, addrv6)
307  self.assertEqual(py6.frametype(), 302)
308  self.assertEqual(py6.framelen(), 20)
309  self.assertEqual(str(py6), '302: IpPortFrame(302, [::1]:79)')
310 
311  @class_teardown
312  def tearDown(self):
313  assert_no_dangling_Cclasses()
314 
315 
316 class pyIpPortFrameTest2(TestCase):
317  'An AddrFrame wraps a NetAddr *with a port* for sending on the wire'
318  def test_constructor(self):
319  if DEBUG: print >>sys.stderr, "===============test_constructor(pyIpAddrFrameTest)"
320  pyf = pyIpPortFrame(200, (1,2,3,4), 1984)
321  self.assertEqual(pyf.frametype(), 200)
322  self.assertEqual(pyf.framelen(), 8)
323  self.assertEqual(str(pyf), '200: IpPortFrame(200, 1.2.3.4:1984)')
324  self.assertEqual(pyf.getnetaddr(), pyNetAddr('1.2.3.4:1984'))
325  self.assertEqual(pyf.addrtype(), 1)
326  self.assertTrue(pyf.isvalid(), "pyIpPortFrame(200, (1,2,3,4:1984)) failed isvalid()")
327  self.assertRaises(ValueError, pyIpPortFrame, 201, (1,2,3),80)
328  pyf = pyIpPortFrame(202, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16), 1984)
329  self.assertEqual(pyf.frametype(), 202)
330  self.assertEqual(pyf.framelen(), 20)
331  self.assertEqual(pyf.addrtype(), 2)
332  self.assertTrue(pyf.isvalid(), 'pyIpPortFrame(202, ([102:304:506:708:90a:b0c:d0e:f10]:1984))')
333  self.assertEqual(str(pyf), '202: IpPortFrame(202, [102:304:506:708:90a:b0c:d0e:f10]:1984)')
334  sameaddr = pyNetAddr([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10], port=1984)
335  self.assertEqual(pyf.getnetaddr(), sameaddr)
336  pyf = pyIpPortFrame(202, sameaddr, None)
337  self.assertEqual(str(pyf), '202: IpPortFrame(202, [102:304:506:708:90a:b0c:d0e:f10]:1984)')
338  self.assertEqual(pyf.getnetaddr(), sameaddr)
339 
340 
341  @class_teardown
342  def tearDown(self):
343  assert_no_dangling_Cclasses()
344 
345 class pyIntFrameTest(TestCase):
346  'An IntFrame wraps various sizes of unsigned integers for sending on the wire'
347  def test_constructor(self):
348  if DEBUG: print >>sys.stderr, "========================test_constructor(pyIntFrameTest)"
349  # Test a variety of illegal/unsupported integer sizes
350  for size in (5, 6, 7, 9, 10):
351  self.assertRaises(ValueError, pyIntFrame, 300+size-1, intbytes=size)
352  # Some network protocols have 3-byte integers. So I implemented them.
353  for size in (1, 2, 3, 4, 8):
354  pyf = pyIntFrame(310+size, initval=42, intbytes=size)
355  self.assertTrue(pyf.isvalid())
356  self.assertEqual(pyf.intlength(), size)
357  self.assertEqual(int(pyf), 42)
358  self.assertEqual(str(pyf), 'pyIntFrame(%d, (42))' % (310+size))
359 
360  def test_set(self):
361  'Test setting integer values for all the size integers'
362  if DEBUG: print >>sys.stderr, "========================test_set(pyIntFrameTest)"
363  for size in (1, 2, 3, 4, 8):
364  pyf = pyIntFrame(320, initval=0, intbytes=size)
365  val = 42 + size
366  pyf.setint(val)
367  self.assertEqual(int(pyf), val)
368  self.assertEqual(str(pyf), ('pyIntFrame(320, (%d))' % val))
369 
370 
371  @class_teardown
372  def tearDown(self):
373  assert_no_dangling_Cclasses()
374 
375 class pyUnknownFrameTest(TestCase):
376  "An unknown frame is one we don't recognize the type of."
377  def test_constructor(self):
378  if DEBUG: print >>sys.stderr, "========================test_constructor(pyUnknownFrameTest)"
379  pyf = pyUnknownFrame(400)
380  self.assertEqual(pyf.frametype(), 400)
381  # All Unknown frames are invalid...
382  self.assertFalse(pyf.isvalid(), "pyUnkownFrame(400) should not have passed isvalid()")
383 
384  @class_teardown
385  def tearDown(self):
386  assert_no_dangling_Cclasses()
387 
388 
389 class pySeqnoFrameTest(TestCase):
390  'A SeqnoFrame is a frame wrapping an ordered pair for a sequence number'
391  def test_constructor(self):
392  if DEBUG: print >>sys.stderr, "========================test_constructor(pySeqnoFrameTest)"
393  pyf = pySeqnoFrame(500)
394  self.assertEqual(pyf.frametype(), 500)
395  self.assertTrue(pyf.isvalid(), 'pySeqnoFrame(500) did not pass isvalid()')
396  pyf = pySeqnoFrame(501,(1,2))
397  self.assertEqual(pyf.frametype(), 501)
398  self.assertTrue(pyf.isvalid(), 'pySeqnoFrame(501) did not pass isvalid()')
399 
400  def test_reqid(self):
401  'reqid is the request id of a sequence number'
402  if DEBUG: print >>sys.stderr, "========================test_reqid(pySeqnoFrameTest)"
403  pyf = pySeqnoFrame(502)
404  pyf.setreqid(42)
405  self.assertTrue(pyf.getreqid, 42)
406  pyf.setreqid(43)
407  self.assertTrue(pyf.getreqid, 43)
408 
409  def test_qid(self):
410  'qid is analogous to a port - it is the id of a queue on the other side'
411  if DEBUG: print >>sys.stderr, "========================test_qid(pySeqnoFrameTest)"
412  pyf = pySeqnoFrame(503)
413  pyf.setqid(6)
414  self.assertTrue(pyf.getqid, 6)
415  pyf.setqid(7)
416  self.assertTrue(pyf.getqid, 7)
417 
418  def test_equal(self):
419  'A bit of overkill, but nothing really wrong with it'
420  if DEBUG: print >>sys.stderr, "========================test_equal(pySeqnoFrameTest)"
421  seqFrame1 = pySeqnoFrame( 504, (1,1))
422  seqFrame1b = pySeqnoFrame(505, (1,1))
423  seqFrame2 = pySeqnoFrame( 506, (1,2))
424  seqFrame3 = pySeqnoFrame( 507, (2,1))
425  seqFrame4 = pySeqnoFrame( 508, (2,2))
426  seqFrame4b = pySeqnoFrame(509, (2,2))
427  self.assertTrue(seqFrame1 == seqFrame1)
428  self.assertTrue(seqFrame1 == seqFrame1b)
429  self.assertTrue(seqFrame1b == seqFrame1)
430  self.assertFalse(seqFrame1 == seqFrame2)
431  self.assertFalse(seqFrame1 == seqFrame3)
432  self.assertFalse(seqFrame1 == seqFrame4)
433  self.assertFalse(seqFrame1 == seqFrame4b)
434  self.assertFalse(seqFrame2 == seqFrame1)
435  self.assertFalse(seqFrame2 == seqFrame1b)
436  self.assertFalse(seqFrame2 == seqFrame1)
437  self.assertTrue (seqFrame2 == seqFrame2)
438  self.assertFalse(seqFrame2 == seqFrame3)
439  self.assertFalse(seqFrame2 == seqFrame4)
440  self.assertFalse(seqFrame1 == seqFrame4b)
441  self.assertFalse(seqFrame3 == seqFrame1)
442  self.assertFalse(seqFrame3 == seqFrame1b)
443  self.assertFalse(seqFrame3 == seqFrame1)
444  self.assertFalse(seqFrame3 == seqFrame2)
445  self.assertTrue (seqFrame3 == seqFrame3)
446  self.assertFalse(seqFrame3 == seqFrame4)
447  self.assertFalse(seqFrame3 == seqFrame4b)
448  self.assertFalse(seqFrame4 == seqFrame1)
449  self.assertFalse(seqFrame4 == seqFrame1b)
450  self.assertFalse(seqFrame4 == seqFrame1)
451  self.assertFalse(seqFrame4 == seqFrame2)
452  self.assertFalse(seqFrame4 == seqFrame3)
453  self.assertTrue (seqFrame4 == seqFrame4)
454  self.assertTrue (seqFrame4 == seqFrame4b)
455  self.assertTrue(seqFrame4b == seqFrame4)
456 
457  @class_teardown
458  def tearDown(self):
459  assert_no_dangling_Cclasses()
460 
461 
462 class pyCstringFrameTest(TestCase):
463  '''A CstringFrame is a frame which can only hold NUL-terminated C strings.
464  The last byte must be the one and only NUL character in a CstringFrame value.'''
465  def test_constructor(self):
466  if DEBUG: print >>sys.stderr, "========================test_constructor(pyCstringFrameTest)"
467  pyf = pyCstringFrame(600, "Hello, World.")
468  self.assertTrue(pyf.isvalid())
469  self.assertEqual(str(pyf), '600: CstringFrame(600, "Hello, World.")')
470  pyf2 = pyCstringFrame(601)
471  self.assertFalse(pyf2.isvalid())
472  pyf2.setvalue("42")
473  self.assertTrue(pyf2.isvalid())
474  self.assertEqual(str(pyf2), '601: CstringFrame(601, "42")')
475 
476  @class_teardown
477  def tearDown(self):
478  assert_no_dangling_Cclasses()
479 
480 class pySignFrameTest(TestCase):
481  'A SignFrame is a digital signature frame.'
482  def test_constructor(self):
483  if DEBUG: print >>sys.stderr, "========================test_constructor(pySignFrameTest)"
484  pyf = pySignFrame(1) # the 1 determines the type of digital signature
485  self.assertTrue(pyf.isvalid())
486  self.assertRaises(ValueError, pySignFrame, 935) # Just a random invalid signature type
487 
488  @class_teardown
489  def tearDown(self):
490  assert_no_dangling_Cclasses()
491 
492 
493 class pyFrameSetTest(TestCase):
494  'A FrameSet is a collection of frames - typically to be sent over the wire'
495 
496  @staticmethod
497  def cmpstring(frame):
498  s=str(frame)
499  s = re.sub(' at 0x[^{}]*', ' at 0xsomewhere', s)
500  return s
501 
502  def test_constructor(self):
503  if DEBUG: print >>sys.stderr, "========================test_constructor(pyFrameSetTest)"
504  pyf = pyFrameSet(700) # The 700 is the frameset (message) type
505  self.assertEqual(pyf.get_framesettype(), 700)
506 
507  def test_flags(self):
508  if DEBUG: print >>sys.stderr, "========================test_flags(pyFrameSetTest)"
509  'Flags are bit masks, to be turned on or off. They are 16-bits only.'
510  pyf = pyFrameSet(701)
511  self.assertEqual(pyf.get_flags(), 0x00)
512  pyf.set_flags(0x01)
513  self.assertEqual(pyf.get_flags(), 0x01)
514  pyf.set_flags(0x01)
515  self.assertEqual(pyf.get_flags(), 0x01)
516  pyf.set_flags(0x02)
517  self.assertEqual(pyf.get_flags(), 0x03)
518  pyf.clear_flags(0x01)
519  self.assertEqual(pyf.get_flags(), 0x02)
520  pyf.set_flags(0x0fffffffffffffffff)
521  self.assertEqual(pyf.get_flags(), 0x0ffff)
522  pyf.clear_flags(0x5555)
523  self.assertEqual(pyf.get_flags(), 0x0AAAA)
524 
525  def test_buildlistforward(self):
526  'Build a FrameSet using append and verify that it gets built right'
527  if DEBUG: print >>sys.stderr, "========================test_buildlistforward(pyFrameSetTest)"
528  pyfs = pyFrameSet(702)
529  sign = pySignFrame(1) # digital signature frame
530  flist = (pyFrame(703), pyAddrFrame(704, (42,42,42,42)), pyIntFrame(705,42),
531  pyCstringFrame(706, "HhGttG"),
532  pySeqnoFrame(707, (42, 424242424242)),
533  pyIpPortFrame(200, (1,2,3,4), 1984),
534  pyIpPortFrame(202, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16), 1984)
535  )
536  for frame in flist:
537  pyfs.append(frame)
538  self.assertEqual(len(pyfs), 7)
539  #pyfs.dump()
540  ylist = []
541  # The iter member function is a generator. I love it.
542  # It goes to a lot of trouble to wrap the underlying C Classes with Python classes.
543  for frame in pyfs.iter():
544  ylist.append(frame)
545  for i in range(0,len(ylist)):
546  f=flist[i]
547  y=ylist[i]
548  # This isn't exhaustive, but it isn't bad.
549  self.assertEqual(f.frametype(), y.frametype())
550  self.assertEqual(type(f), type(y))
551  self.assertEqual(str(f), str(y))
552  # Constructing the packet will add a signature frame at the beginning
553  # and an END (type 0) frame at the end
554  pyfs.construct_packet(sign)
555  # So we do it over again to make sure everything still looks OK
556  ylist = []
557  for frame in pyfs.iter():
558  ylist.append(frame)
559  self.assertEqual(len(pyfs), 9)
560  self.assertEqual(len(ylist), len(pyfs)) # len(pyfs) traverses the linked list
561  for i in range(0,len(flist)):
562  f=flist[i]
563  y=ylist[i+1]
564  # This isn't exhaustive, but it isn't bad.
565  self.assertEqual(f.frametype(), y.frametype())
566  self.assertEqual(type(f), type(y))
567  # Check on our automatically added frames.
568  self.assertEqual(ylist[0].frametype(), 1)
569  self.assertEqual(ylist[8].frametype(), 0)
570 
571  def test_buildlistbackwards(self):
572  '''Build a FrameSet using prepend and verify that it gets built right.
573  Similar to the append testing above, but backwards ;-)'''
574  if DEBUG: print >>sys.stderr, "========================test_buildlistbackwards(pyFrameSetTest)"
575  pyfs = pyFrameSet(707)
576  sign = pySignFrame(1)
577  flist = (pyFrame(708), pyAddrFrame(709, (42,42,42,42)), pyIntFrame(710,42), pyCstringFrame(711, "HhGttG"),
578  pySeqnoFrame(712, (42, 424242424242)))
579  for frame in flist:
580  pyfs.prepend(frame)
581  self.assertEqual(len(pyfs), 5)
582  #pyfs.dump()
583  ylist = []
584  for frame in pyfs.iter():
585  ylist.append(frame)
586  for i in range(0,len(flist)):
587  f=flist[i]
588  y=ylist[4-i]
589  # This isn't exhaustive, but it isn't bad.
590  self.assertEqual(f.frametype(), y.frametype())
591  self.assertEqual(type(f), type(y))
592  self.assertEqual(f.__class__, y.__class__)
593  if DEBUG: print >>sys.stderr, "Classes are", f.__class__, "lens are", f.framelen(), y.framelen()
594  self.assertEqual(f.framelen(), y.framelen())
595 
596  def test_buildpacket(self):
597  'Build a FrameSet, then make it into a packet, and make a frameset list out of the packet'
598  if DEBUG: print >>sys.stderr, "========================test_buildpacket(pyFrameSetTest)"
599  pyfs = pyFrameSet(801)
600  sign = pySignFrame(1) # digital signature frame
601  flist = (pyAddrFrame(FrameTypes.IPADDR, (42,42,42,42)), pyIntFrame(FrameTypes.WALLCLOCK,42), pyCstringFrame(FrameTypes.INTERFACE, "HhGttG"),
602  pyIntFrame(FrameTypes.CINTVAL,3000000, intbytes=4),
603  pyIntFrame(FrameTypes.CINTVAL,3000000000000, intbytes=8),
604  pySeqnoFrame(FrameTypes.REQID, (42, 424242424242)),
605  pyIpPortFrame(FrameTypes.IPPORT, (1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16), 1984),
606  pyIntFrame(FrameTypes.CINTVAL,4242, intbytes=3))
607  if DEBUG: print >>sys.stderr, "flist:", flist
608  decoder = pyPacketDecoder(0)
609  for frame in flist:
610  pyfs.append(frame)
611  pyfs.construct_packet(sign)
612  if DEBUG: print >>sys.stderr, "packet constructed"
613  xlist=[]
614  for frame in pyfs.iter():
615  xlist.append(frame)
616  if DEBUG: print >>sys.stderr, "xlist constructed"
617  pktdata = pyfs.getpacket()
618  if DEBUG: print >>sys.stderr, "getpacket done", pktdata
619  cp_pyfs = decoder.fslist_from_pktdata(pktdata)
620  if DEBUG: print >>sys.stderr, "decoder done", cp_pyfs
621  fs0 = cp_pyfs[0]
622  ylist=[]
623  for frame in fs0.iter():
624  ylist.append(frame)
625  for i in range(0,len(xlist)):
626  x=xlist[i]
627  y=ylist[i]
628  self.assertEqual(x.frametype(), y.frametype())
629  self.assertEqual(x.framelen(), y.framelen())
630  self.assertEqual(x.dataspace(), y.dataspace())
631  self.assertEqual(type(x), type(y))
632  self.assertEqual(x.__class__, y.__class__)
633  # Not all our classes have a __str__ method defined.
634  strx = re.sub(str(x), ' instance at .*>', ' instance at -somewhere- >')
635  stry = re.sub(str(y), ' instance at .*>', ' instance at -somewhere- >')
636  self.assertEqual(strx, stry)
637  self.assertEqual(pyFrameSetTest.cmpstring(x), pyFrameSetTest.cmpstring(y))
638 
639 
640  @class_teardown
641  def tearDown(self):
642  assert_no_dangling_Cclasses()
643 
644 class pyConfigContextTest(TestCase):
645 
646  def test_constructor(self):
647  if DEBUG: print >>sys.stderr, "===============test_constructor(pyConfigContextTest)"
648  pyConfigContext()
649  foo = pyConfigContext(init={'int1': 42, 'str1': 'forty-two', 'bar': pyNetAddr((1,2,3,4),) })
650  foo = pyConfigContext(init={'int1': 42, 'str1': 'forty-two', 'bar': pyNetAddr((1,2,3,4),), 'csf': pyCstringFrame(42, '41+1')})
651  self.assertEqual(foo.getint('int1'), 42)
652  self.assertEqual(foo.getstring('str1'), 'forty-two')
653  self.assertRaises(IndexError, foo.getaddr, ('int1'))
654  self.assertRaises(IndexError, foo.getstring, ('int1'))
655  self.assertRaises(IndexError, foo.getaddr, ('str1'))
656  self.assertRaises(IndexError, foo.getframe, ('str1'))
657  self.assertRaises(IndexError, foo.getframe, ('int1'))
658  self.assertEqual(foo['int1'], 42)
659  self.assertEqual(foo['str1'], 'forty-two')
660  self.assertEqual(foo.getint('fred'), -1)
661  foo['bar']
662  self.assertEqual(foo['bar'], pyNetAddr((1,2,3,4),))
663  self.assertEqual(str(foo['bar']), '1.2.3.4')
664  self.assertEqual(str(foo['csf']), '42: CstringFrame(42, "41+1")')
665  self.assertEqual(str(foo), '{"bar":"1.2.3.4","csf":"CstringFrame(42, \\"41+1\\")","int1":42,"str1":"forty-two"}')
666 
667  foo['isf'] = pyIntFrame(310, initval=42, intbytes=3)
668  if DEBUG: print >>sys.stderr, "test_constructor.18(pyConfigContextTest)"
669  self.assertEqual(str(foo),
670  '{"bar":"1.2.3.4","csf":"CstringFrame(42, \\"41+1\\")","int1":42,"isf":"IntFrame(310, 3, 42)","str1":"forty-two"}')
671  if DEBUG: print >>sys.stderr, "test_constructor.19(pyConfigContextTest)"
672 
673  def test_string(self):
674  if DEBUG: print >>sys.stderr, "test_string(pyConfigContextTest)"
675  foo = pyConfigContext()
676  foo['arthur'] = 'dent'
677  foo['seven'] = 'ofnine'
678  foo['JeanLuc'] = 'Picard'
679  foo['important'] = 'towel'
680  foo['integer'] = 42
681  self.assertEqual(foo['arthur'], 'dent')
682  self.assertEqual(foo['seven'], 'ofnine')
683  self.assertEqual(foo['JeanLuc'], 'Picard')
684  self.assertEqual(foo['important'], 'towel')
685  self.assertRaises(IndexError, foo.getstring, ('towel'))
686  self.assertEqual(str(foo), '{"JeanLuc":"Picard","arthur":"dent","important":"towel","integer":42,"seven":"ofnine"}')
687  foo['seven'] = '7'
688  self.assertEqual(foo['seven'], '7')
689  self.assertEqual(type(foo['seven']), str)
690  foo['JeanLuc'] = 'Locutus'
691  self.assertEqual(foo['JeanLuc'], 'Locutus')
692  self.assertEqual(str(foo), '{"JeanLuc":"Locutus","arthur":"dent","important":"towel","integer":42,"seven":"7"}')
693 
694  def test_int(self):
695  if DEBUG: print >>sys.stderr, "test_int(pyConfigContextTest)"
696  foo = pyConfigContext()
697  foo['arthur'] = 42
698  foo['seven'] = 9
699  self.assertEqual(foo['arthur'], 42)
700  self.assertEqual(foo['seven'], 9)
701  self.assertEqual(str(foo), '{"arthur":42,"seven":9}')
702  foo['seven'] = 7
703  self.assertEqual(type(foo['seven']), int)
704  self.assertEqual(foo["seven"], 7)
705  self.assertEqual(str(foo), '{"arthur":42,"seven":7}')
706 
707  def test_child_ConfigContext(self):
708  if DEBUG: print >>sys.stderr, "========================test_child_ConfigContext(pyConfigContextTest)"
709  foo = pyConfigContext()
710  foo['ford'] = 'prefect'
711  baz = pyConfigContext()
712  baz['Kathryn'] = 'Janeway'
713  baz['there\'s no place like'] = pyNetAddr((127,0,0,1),)
714  bar = pyConfigContext()
715  bar['hhgttg'] = foo
716  bar['voyager'] = baz
717  if DEBUG: print >>sys.stderr, "EQUAL TEST"
718  self.assertEqual(str(bar), '{"hhgttg":{"ford":"prefect"},"voyager":{"Kathryn":"Janeway","there\'s no place like":"127.0.0.1"}}')
719  # We make a new pyConfigContext object from the str() of another one. Cool!
720  if DEBUG: print >>sys.stderr, "JSON TEST"
721  bar2 = pyConfigContext(str(bar))
722  if DEBUG: print >>sys.stderr, "JSON COMPARE"
723  self.assertEqual(str(bar), str(bar2))
724  self.assertEqual(bar["voyager"]["Kathryn"], "Janeway")
725  self.assertEqual(bar["hhgttg"]["ford"], "prefect")
726  self.assertEqual(bar2["voyager"]["Kathryn"], "Janeway")
727  self.assertEqual(bar2["hhgttg"]["ford"], "prefect")
728  if DEBUG: print >>sys.stderr, "COMPARE DONE"
729  # However... The pyNetAddr() was turned into a mere string :-( - at least for the moment... Sigh...
730  if DEBUG: print >>sys.stderr, "END OF ========================test_child_ConfigContext(pyConfigContextTest)"
731 
732 
733  def test_keys(self):
734  if DEBUG: print >>sys.stderr, "===============test_keys(pyConfigContextTest)"
735  foo = pyConfigContext()
736  foo['arthur'] = 'dent'
737  foo['seven'] = 'ofnine'
738  foo['JeanLuc'] = 'Picard'
739  foo['important'] = 'towel'
740  foo['integer'] = 42
741  self.assertEqual(str(foo.keys()), "['JeanLuc', 'arthur', 'important', 'integer', 'seven']")
742 
743  def test_ConfigContext_array(self):
744  if DEBUG: print >>sys.stderr, "===============textConfigContext_array(pyConfigContextTest)"
745  array1str = '{"a":[1,2,3,4,"a",{"b":true},[5,6,7,8,3.14]]}'
746  array1config = pyConfigContext(array1str)
747  self.assertEqual(array1str, str(array1config))
748 
749  def test_ConfigContext_array_with_netaddr(self):
750  'This problem actually occurred - hence the test case...'
751  if DEBUG: print >>sys.stderr, "===============textConfigContext_array_with_netaddr(pyConfigContextTest)"
752  array1str = '{"a":["1.2.3.4",1,2,3,4,"a",{"b":true},[5,6,7,8,3.14,"10.10.10.1"],"::1"]}'
753  array1str = '{"a":["1.2.3.4"]}'
754  array1config = pyConfigContext(array1str)
755  #foo = array1config['a']
756  self.assertEqual(array1str, str(array1config))
757 
758  def test_kitchen_sink(self):
759  if DEBUG:
760  for j in range(1,5):
761  proj_class_incr_debug('NetAddr')
762  strings = [
763  # The next two originally failed intermittently
764  '{"a":["1.2.3.4"]}',
765  '{"cmdline":["192.168.122.1"]}',
766  # All the remaining ones originally failed pretty reliably...
767  '{"cmdline":["/usr/sbin/dnsmasq","-u","libvirt-dnsmasq","--strict-order","--bind-interfaces","--pid-file=/var/run/libvirt/network/default.pid","--conf-file=","--except-interface","lo","--listen-address","192.168.122.1"]}',
768  '{"cmdline":["--listen-address","192.168.122.1"]}',
769  '{"cmdline":["/usr/sbin/dnsmasq","-u","libvirt-dnsmasq","--strict-order","--bind-interfaces","--pid-file=/var/run/libvirt/network/default.pid","--conf-file=","--except-interface","lo","--listen-address","192.168.122.1","--dhcp-range","192.168.122.2,192.168.122.254","--dhcp-leasefile=/var/lib/libvirt/dnsmasq/default.leases","--dhcp-lease-max=253","--dhcp-no-override" ]}',
770  '{"cmdline":["--listen-address","192.168.122.1","--dhcp-range","192.168.122.2,192.168.122.254","--dhcp-leasefile=/var/lib/libvirt/dnsmasq/default.leases","--dhcp-lease-max=253","--dhcp-no-override" ]}',
771  '{"cmdline":["--listen-address","192.168.122.1","--dhcp-range"]}',
772  '{"cmdline":["192.168.122.1","--dhcp-range"]}',
773  '{"cmdline":["192.168.122.1","anything"]}',
774  '{"cmdline":["192.168.122.1",1]}',
775  '{"cmdline":["192.168.122.1",false]}',
776  ]
777  for s in strings:
778  if DEBUG: print >>sys.stderr, ('Creating pyConfigContext("%s")' % s)
779  sc = pyConfigContext(s)
780  if DEBUG: rint >>sys.stderr, ('sc.keys() == %s' % sc.keys())
781  for key in sc.keys():
782  elemcount=0
783  if DEBUG: print >>sys.stderr, ('Looking at key %s: sc[key] = %s' % (key, sc[key]))
784  for elem in sc[key]:
785  if DEBUG: print >>sys.stderr, ('Looking at element %s' % str(elem))
786  self.assertNotEqual(str(elem), "")
787  if isinstance(elem, pyAssimObj):
788  if DEBUG: print '++++++++++++++++++ REFCOUNT(%s): %d' % (str(elem), elem.refcount())
789  #CCref(elem._Cstruct)
790  self.assertEqual(elem.refcount(), 2)
791  if DEBUG:
792  gc.collect()
793  print >>sys.stderr, ":::::::::::::GC GARBAGE: %s" % gc.garbage
794  foo=elem._Cstruct[0]
795  while (hasattr(foo, 'baseclass')):
796  foo=foo.baseclass
797  if DEBUG:
798  print >>sys.stderr, ":::::::::::::GC refcount %d, REFERRERS: %s" % (sys.getrefcount(elem), gc.get_referrers(elem))
799  print >>sys.stderr, ":::::::::::::FOO: %s" % foo
800  del elem
801  if DEBUG: print '++++++++++++++++++ REFCOUNT SECOND VERSE: %s' % (foo._refcount)
802 
803  elemcount += 1
804  if DEBUG:
805  for j in range(1,5):
806  proj_class_decr_debug('NetAddr')
807 
808  @class_teardown
809  def tearDown(self):
810  assert_no_dangling_Cclasses()
811 
812 class pyNetIOudpTest(TestCase):
813 
814  def test_constructor(self):
815  if DEBUG: print >>sys.stderr, "========================test_constructor(pyNetIOudpTest)"
816  config = pyConfigContext(init={'outsig': pySignFrame(1)})
817  io = pyNetIOudp(config, pyPacketDecoder(0))
818  self.assertTrue(io.getfd() > 2)
819 
820  def test_members(self):
821  if DEBUG: print >>sys.stderr, "========================test_members(pyNetIOudpTest)"
822  io = pyNetIOudp(pyConfigContext(init={'outsig': pySignFrame(1)}), pyPacketDecoder(0))
823  self.assertTrue(io.getmaxpktsize() > 65000)
824  self.assertTrue(io.getmaxpktsize() < 65535)
825  io.setmaxpktsize(1500)
826  self.assertEqual(io.getmaxpktsize(), 1500)
827  # Does signframe really work? Next statement seems to crash things
828  #self.assertEqual(type(io.signframe()), type(pySignFrame(1)))
829 
830  def test_send(self):
831  if DEBUG: print >>sys.stderr, "========================test_send(pyNetIOudpTest)"
832  home = pyNetAddr((127,0,0,1),1984)
833  fs = pyFrameSet(801)
834  flist = (pyAddrFrame(FrameTypes.IPADDR, (42,42,42,42)), pyIntFrame(FrameTypes.HBWARNTIME,42), pyCstringFrame(FrameTypes.HOSTNAME, "HhGttG"),
835  pyIntFrame(FrameTypes.PORTNUM,3000000, intbytes=4),
836  pyIntFrame(FrameTypes.HBINTERVAL,3000000000000, intbytes=8),
837  pySeqnoFrame(FrameTypes.REQID, (42, 424242424242)),
838  pyIntFrame(FrameTypes.CINTVAL,4242, intbytes=3))
839  for frame in flist:
840  fs.append(frame)
841  io = pyNetIOudp(pyConfigContext(init={'outsig': pySignFrame(1)}), pyPacketDecoder(0))
842  io.sendframesets(home, fs)
843  io.sendframesets(home, (fs,fs,fs))
844 
845  def test_receiveandsend(self):
846  if DEBUG: print >>sys.stderr, "========================test_receiveandsend(pyNetIOudpTest)"
847  home = pyNetAddr("::1", 1984)
848  anyaddr = pyNetAddr("::",1984)
849  fs = pyFrameSet(801)
850  #flist = (pyIntFrame(7,42), pyCstringFrame(8, "HhGttG"),
851  flist = (pyAddrFrame(FrameTypes.IPADDR, (42,42,42,42)), pyIntFrame(7,42), pyCstringFrame(8, "HhGttG"),
852  pyAddrFrame(FrameTypes.IPADDR,(1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16)),
853  pyIntFrame(FrameTypes.HBINTERVAL,3000000, intbytes=4),
854  pyIntFrame(FrameTypes.HBDEADTIME,3000000000000, intbytes=8),
855  pySeqnoFrame(FrameTypes.REQID, (42, 424242424242)),
856  pyIntFrame(FrameTypes.PORTNUM,4242, intbytes=3))
857  for frame in flist:
858  fs.append(frame)
859  io = pyNetIOudp(pyConfigContext(init={'outsig': pySignFrame(1)}), pyPacketDecoder(0))
860  io.bindaddr(anyaddr)
861  io.sendframesets(home, fs) # Send a packet with a single frameset containing a bunch of frames
862  (addr, framesetlist) = io.recvframesets() # Receive a packet - with some framesets in it
863  #print >>sys.stderr, 'ADDR: [%s] HOME: [%s]' % (addr, home)
864  self.assertEqual(addr, home)
865  self.assertEqual(len(framesetlist), 1)
866  ylist = []
867  for frame in framesetlist[0].iter():
868  ylist.append(frame)
869  self.assertEqual(len(flist), len(ylist)-2)
870  for i in range(0,len(flist)):
871  x=flist[i]
872  y=ylist[i+1]
873  self.assertEqual(x.frametype(), y.frametype())
874  self.assertEqual(x.framelen(), y.framelen())
875  self.assertEqual(x.dataspace(), y.dataspace())
876  self.assertEqual(type(x), type(y))
877  self.assertEqual(x.__class__, y.__class__)
878  self.assertEqual(pyFrameSetTest.cmpstring(x), pyFrameSetTest.cmpstring(y))
879 
880  @class_teardown
881  def tearDown(self):
882  assert_no_dangling_Cclasses()
883 
884 if __name__ == "__main__":
885  run()