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