home *** CD-ROM | disk | FTP | other *** search
/ PC Professionell 2004 December / PCpro_2004_12.ISO / files / webserver / xampp / xampp-python-addon-1.4.9-installer.exe / test_queue.py < prev    next >
Encoding:
Python Source  |  2003-01-28  |  4.1 KB  |  174 lines
  1. """
  2. TestCases for exercising a Queue DB.
  3. """
  4.  
  5. import sys, os, string
  6. import tempfile
  7. from pprint import pprint
  8. import unittest
  9.  
  10. try:
  11.     # For Python 2.3
  12.     from bsddb import db
  13. except ImportError:
  14.     # For earlier Pythons w/distutils pybsddb
  15.     from bsddb3 import db
  16.  
  17. from test_all import verbose
  18.  
  19.  
  20. #----------------------------------------------------------------------
  21.  
  22. class SimpleQueueTestCase(unittest.TestCase):
  23.     def setUp(self):
  24.         self.filename = tempfile.mktemp()
  25.  
  26.     def tearDown(self):
  27.         try:
  28.             os.remove(self.filename)
  29.         except os.error:
  30.             pass
  31.  
  32.  
  33.     def test01_basic(self):
  34.         # Basic Queue tests using the deprecated DBCursor.consume method.
  35.  
  36.         if verbose:
  37.             print '\n', '-=' * 30
  38.             print "Running %s.test01_basic..." % self.__class__.__name__
  39.  
  40.         d = db.DB()
  41.         d.set_re_len(40)  # Queues must be fixed length
  42.         d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)
  43.  
  44.         if verbose:
  45.             print "before appends" + '-' * 30
  46.             pprint(d.stat())
  47.  
  48.         for x in string.letters:
  49.             d.append(x * 40)
  50.  
  51.         assert len(d) == 52
  52.  
  53.         d.put(100, "some more data")
  54.         d.put(101, "and some more ")
  55.         d.put(75,  "out of order")
  56.         d.put(1,   "replacement data")
  57.  
  58.         assert len(d) == 55
  59.  
  60.         if verbose:
  61.             print "before close" + '-' * 30
  62.             pprint(d.stat())
  63.  
  64.         d.close()
  65.         del d
  66.         d = db.DB()
  67.         d.open(self.filename)
  68.  
  69.         if verbose:
  70.             print "after open" + '-' * 30
  71.             pprint(d.stat())
  72.  
  73.         d.append("one more")
  74.         c = d.cursor()
  75.  
  76.         if verbose:
  77.             print "after append" + '-' * 30
  78.             pprint(d.stat())
  79.  
  80.         rec = c.consume()
  81.         while rec:
  82.             if verbose:
  83.                 print rec
  84.             rec = c.consume()
  85.         c.close()
  86.  
  87.         if verbose:
  88.             print "after consume loop" + '-' * 30
  89.             pprint(d.stat())
  90.  
  91.         assert len(d) == 0, \
  92.                "if you see this message then you need to rebuild " \
  93.                "BerkeleyDB 3.1.17 with the patch in patches/qam_stat.diff"
  94.  
  95.         d.close()
  96.  
  97.  
  98.  
  99.     def test02_basicPost32(self):
  100.         # Basic Queue tests using the new DB.consume method in DB 3.2+
  101.         # (No cursor needed)
  102.  
  103.         if verbose:
  104.             print '\n', '-=' * 30
  105.             print "Running %s.test02_basicPost32..." % self.__class__.__name__
  106.  
  107.         if db.version() < (3, 2, 0):
  108.             if verbose:
  109.                 print "Test not run, DB not new enough..."
  110.             return
  111.  
  112.         d = db.DB()
  113.         d.set_re_len(40)  # Queues must be fixed length
  114.         d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)
  115.  
  116.         if verbose:
  117.             print "before appends" + '-' * 30
  118.             pprint(d.stat())
  119.  
  120.         for x in string.letters:
  121.             d.append(x * 40)
  122.  
  123.         assert len(d) == 52
  124.  
  125.         d.put(100, "some more data")
  126.         d.put(101, "and some more ")
  127.         d.put(75,  "out of order")
  128.         d.put(1,   "replacement data")
  129.  
  130.         assert len(d) == 55
  131.  
  132.         if verbose:
  133.             print "before close" + '-' * 30
  134.             pprint(d.stat())
  135.  
  136.         d.close()
  137.         del d
  138.         d = db.DB()
  139.         d.open(self.filename)
  140.         #d.set_get_returns_none(true)
  141.  
  142.         if verbose:
  143.             print "after open" + '-' * 30
  144.             pprint(d.stat())
  145.  
  146.         d.append("one more")
  147.  
  148.         if verbose:
  149.             print "after append" + '-' * 30
  150.             pprint(d.stat())
  151.  
  152.         rec = d.consume()
  153.         while rec:
  154.             if verbose:
  155.                 print rec
  156.             rec = d.consume()
  157.  
  158.         if verbose:
  159.             print "after consume loop" + '-' * 30
  160.             pprint(d.stat())
  161.  
  162.         d.close()
  163.  
  164.  
  165.  
  166. #----------------------------------------------------------------------
  167.  
  168. def test_suite():
  169.     return unittest.makeSuite(SimpleQueueTestCase)
  170.  
  171.  
  172. if __name__ == '__main__':
  173.     unittest.main(defaultTest='test_suite')
  174.