Initial commit of OpenSPARC T2 architecture model.

[OpenSPARC-T2-SAM] / sam-t2 / devtools / v8plus / lib / python2.4 / bsddb / test / test_queue.py

"""
TestCases for exercising a Queue DB.
"""

import sys, os, string
import tempfile
from pprint import pprint
import unittest

try:
   # For Pythons w/distutils pybsddb
   from bsddb3 import db
except ImportError:
   # For Python 2.3
   from bsddb import db

from test_all import verbose


#----------------------------------------------------------------------

class SimpleQueueTestCase(unittest.TestCase):
   def setUp(self):
       self.filename = tempfile.mktemp()

   def tearDown(self):
       try:
           os.remove(self.filename)
       except os.error:
           pass


   def test01_basic(self):
       # Basic Queue tests using the deprecated DBCursor.consume method.

       if verbose:
           print '\n', '-=' * 30
           print "Running %s.test01_basic..." % self.__class__.__name__

       d = db.DB()
       d.set_re_len(40)  # Queues must be fixed length
       d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)

       if verbose:
           print "before appends" + '-' * 30
           pprint(d.stat())

       for x in string.letters:
           d.append(x * 40)

       assert len(d) == 52

       d.put(100, "some more data")
       d.put(101, "and some more ")
       d.put(75,  "out of order")
       d.put(1,   "replacement data")

       assert len(d) == 55

       if verbose:
           print "before close" + '-' * 30
           pprint(d.stat())

       d.close()
       del d
       d = db.DB()
       d.open(self.filename)

       if verbose:
           print "after open" + '-' * 30
           pprint(d.stat())

       d.append("one more")
       c = d.cursor()

       if verbose:
           print "after append" + '-' * 30
           pprint(d.stat())

       rec = c.consume()
       while rec:
           if verbose:
               print rec
           rec = c.consume()
       c.close()

       if verbose:
           print "after consume loop" + '-' * 30
           pprint(d.stat())

       assert len(d) == 0, \
              "if you see this message then you need to rebuild " \
              "BerkeleyDB 3.1.17 with the patch in patches/qam_stat.diff"

       d.close()



   def test02_basicPost32(self):
       # Basic Queue tests using the new DB.consume method in DB 3.2+
       # (No cursor needed)

       if verbose:
           print '\n', '-=' * 30
           print "Running %s.test02_basicPost32..." % self.__class__.__name__

       if db.version() < (3, 2, 0):
           if verbose:
               print "Test not run, DB not new enough..."
           return

       d = db.DB()
       d.set_re_len(40)  # Queues must be fixed length
       d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)

       if verbose:
           print "before appends" + '-' * 30
           pprint(d.stat())

       for x in string.letters:
           d.append(x * 40)

       assert len(d) == 52

       d.put(100, "some more data")
       d.put(101, "and some more ")
       d.put(75,  "out of order")
       d.put(1,   "replacement data")

       assert len(d) == 55

       if verbose:
           print "before close" + '-' * 30
           pprint(d.stat())

       d.close()
       del d
       d = db.DB()
       d.open(self.filename)
       #d.set_get_returns_none(true)

       if verbose:
           print "after open" + '-' * 30
           pprint(d.stat())

       d.append("one more")

       if verbose:
           print "after append" + '-' * 30
           pprint(d.stat())

       rec = d.consume()
       while rec:
           if verbose:
               print rec
           rec = d.consume()

       if verbose:
           print "after consume loop" + '-' * 30
           pprint(d.stat())

       d.close()



#----------------------------------------------------------------------

def test_suite():
   return unittest.makeSuite(SimpleQueueTestCase)


if __name__ == '__main__':
   unittest.main(defaultTest='test_suite')