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