TestCases for exercising a Queue DB.
from pprint
import pprint
# For Pythons w/distutils pybsddb
from test_all
import verbose
#----------------------------------------------------------------------
class SimpleQueueTestCase(unittest
.TestCase
):
self
.filename
= tempfile
.mktemp()
# Basic Queue tests using the deprecated DBCursor.consume method.
print "Running %s.test01_basic..." % self
.__class
__.__name
__
d
.set_re_len(40) # Queues must be fixed length
d
.open(self
.filename
, db
.DB_QUEUE
, db
.DB_CREATE
)
print "before appends" + '-' * 30
d
.put(100, "some more data")
d
.put(101, "and some more ")
d
.put(75, "out of order")
d
.put(1, "replacement data")
print "before close" + '-' * 30
print "after open" + '-' * 30
print "after append" + '-' * 30
print "after consume loop" + '-' * 30
"if you see this message then you need to rebuild " \
"BerkeleyDB 3.1.17 with the patch in patches/qam_stat.diff"
def test02_basicPost32(self
):
# Basic Queue tests using the new DB.consume method in DB 3.2+
print "Running %s.test02_basicPost32..." % self
.__class
__.__name
__
if db
.version() < (3, 2, 0):
print "Test not run, DB not new enough..."
d
.set_re_len(40) # Queues must be fixed length
d
.open(self
.filename
, db
.DB_QUEUE
, db
.DB_CREATE
)
print "before appends" + '-' * 30
d
.put(100, "some more data")
d
.put(101, "and some more ")
d
.put(75, "out of order")
d
.put(1, "replacement data")
print "before close" + '-' * 30
#d.set_get_returns_none(true)
print "after open" + '-' * 30
print "after append" + '-' * 30
print "after consume loop" + '-' * 30
#----------------------------------------------------------------------
return unittest
.makeSuite(SimpleQueueTestCase
)
if __name__
== '__main__':
unittest
.main(defaultTest
='test_suite')