<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import errno <EOL> import shutil <EOL> import string <EOL> import tempfile <EOL> from pprint import pprint <EOL> import unittest <EOL> import time <EOL> try : <EOL> from bsddb3 import db <EOL> except ImportError : <EOL> from bsddb import db <EOL> from test_all import verbose <EOL> DASH = '<STR_LIT:->' <EOL> class VersionTestCase ( unittest . TestCase ) : <EOL> def test00_version ( self ) : <EOL> info = db . version ( ) <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:20> <EOL> print '<STR_LIT>' % ( info , ) <EOL> print db . DB_VERSION_STRING <EOL> print '<STR_LIT>' * <NUM_LIT:20> <EOL> assert info == ( db . DB_VERSION_MAJOR , db . DB_VERSION_MINOR , <EOL> db . DB_VERSION_PATCH ) <EOL> class BasicTestCase ( unittest . TestCase ) : <EOL> dbtype = db . DB_UNKNOWN <EOL> dbopenflags = <NUM_LIT:0> <EOL> dbsetflags = <NUM_LIT:0> <EOL> dbmode = <NUM_LIT:0> <NUM_LIT> <EOL> dbname = None <EOL> useEnv = <NUM_LIT:0> <EOL> envflags = <NUM_LIT:0> <EOL> envsetflags = <NUM_LIT:0> <EOL> _numKeys = <NUM_LIT> <EOL> def setUp ( self ) : <EOL> if self . useEnv : <EOL> homeDir = os . path . join ( os . path . dirname ( sys . argv [ <NUM_LIT:0> ] ) , '<STR_LIT>' ) <EOL> self . homeDir = homeDir <EOL> try : <EOL> shutil . rmtree ( homeDir ) <EOL> except OSError , e : <EOL> if e . errno not in ( errno . ENOENT , errno . ESRCH ) : raise <EOL> os . mkdir ( homeDir ) <EOL> try : <EOL> self . env = db . DBEnv ( ) <EOL> self . env . set_lg_max ( <NUM_LIT> * <NUM_LIT> ) <EOL> self . env . set_tx_max ( <NUM_LIT:30> ) <EOL> self . env . set_tx_timestamp ( int ( time . time ( ) ) ) <EOL> self . env . set_flags ( self . envsetflags , <NUM_LIT:1> ) <EOL> self . env . open ( homeDir , self . envflags | db . DB_CREATE ) <EOL> tempfile . tempdir = homeDir <EOL> self . filename = os . path . split ( tempfile . mktemp ( ) ) [ <NUM_LIT:1> ] <EOL> tempfile . tempdir = None <EOL> except : <EOL> shutil . rmtree ( homeDir ) <EOL> raise <EOL> else : <EOL> self . env = None <EOL> self . filename = tempfile . mktemp ( ) <EOL> self . d = db . DB ( self . env ) <EOL> self . d . set_flags ( self . dbsetflags ) <EOL> if self . dbname : <EOL> self . d . open ( self . filename , self . dbname , self . dbtype , <EOL> self . dbopenflags | db . DB_CREATE , self . dbmode ) <EOL> else : <EOL> self . d . open ( self . filename , <EOL> mode = self . dbmode , <EOL> dbtype = self . dbtype , <EOL> flags = self . dbopenflags | db . DB_CREATE ) <EOL> self . populateDB ( ) <EOL> def tearDown ( self ) : <EOL> self . d . close ( ) <EOL> if self . env is not None : <EOL> self . env . close ( ) <EOL> shutil . rmtree ( self . homeDir ) <EOL> else : <EOL> os . remove ( self . filename ) <EOL> def populateDB ( self , _txn = None ) : <EOL> d = self . d <EOL> for x in range ( self . _numKeys / <NUM_LIT:2> ) : <EOL> key = '<STR_LIT>' % ( self . _numKeys - x ) <EOL> data = self . makeData ( key ) <EOL> d . put ( key , data , _txn ) <EOL> d . put ( '<STR_LIT>' , '<STR_LIT>' , _txn ) <EOL> for x in range ( self . _numKeys / <NUM_LIT:2> - <NUM_LIT:1> ) : <EOL> key = '<STR_LIT>' % x <EOL> data = self . makeData ( key ) <EOL> d . put ( key , data , _txn ) <EOL> if _txn : <EOL> _txn . commit ( ) <EOL> num = len ( d ) <EOL> if verbose : <EOL> print "<STR_LIT>" % num <EOL> def makeData ( self , key ) : <EOL> return DASH . join ( [ key ] * <NUM_LIT:5> ) <EOL> def test01_GetsAndPuts ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> for key in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> data = d . get ( key ) <EOL> if verbose : <EOL> print data <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> assert d . get ( '<STR_LIT>' ) == None <EOL> try : <EOL> d . delete ( '<STR_LIT>' ) <EOL> except db . DBNotFoundError , val : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_NOTFOUND <EOL> if verbose : print val <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> d . put ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> d . put ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if self . dbsetflags & db . DB_DUP : <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> else : <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> try : <EOL> d . put ( '<STR_LIT>' , '<STR_LIT>' , flags = db . DB_NOOVERWRITE ) <EOL> except db . DBKeyExistError , val : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_KEYEXIST <EOL> if verbose : print val <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> if self . dbsetflags & db . DB_DUP : <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> else : <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> d . sync ( ) <EOL> d . close ( ) <EOL> del d <EOL> self . d = db . DB ( self . env ) <EOL> if self . dbname : <EOL> self . d . open ( self . filename , self . dbname ) <EOL> else : <EOL> self . d . open ( self . filename ) <EOL> d = self . d <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> if self . dbsetflags & db . DB_DUP : <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> else : <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> rec = d . get_both ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if verbose : <EOL> print rec <EOL> assert d . get_both ( '<STR_LIT>' , '<STR_LIT>' ) == None <EOL> data = d . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert data == '<STR_LIT>' <EOL> data = d . get ( '<STR_LIT>' , self ) <EOL> assert data == self <EOL> s = d . stat ( ) <EOL> assert type ( s ) == type ( { } ) <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> pprint ( s ) <EOL> def test02_DictionaryMethods ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> for key in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> data = d [ key ] <EOL> assert data == self . makeData ( key ) <EOL> if verbose : <EOL> print data <EOL> assert len ( d ) == self . _numKeys <EOL> keys = d . keys ( ) <EOL> assert len ( keys ) == self . _numKeys <EOL> assert type ( keys ) == type ( [ ] ) <EOL> d [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> assert len ( d ) == self . _numKeys + <NUM_LIT:1> <EOL> keys = d . keys ( ) <EOL> assert len ( keys ) == self . _numKeys + <NUM_LIT:1> <EOL> d [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> assert len ( d ) == self . _numKeys + <NUM_LIT:1> <EOL> keys = d . keys ( ) <EOL> assert len ( keys ) == self . _numKeys + <NUM_LIT:1> <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> pprint ( keys [ : <NUM_LIT:10> ] ) <EOL> assert d [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert d . has_key ( '<STR_LIT>' ) == <NUM_LIT:1> <EOL> assert d . has_key ( '<STR_LIT>' ) == <NUM_LIT:0> <EOL> items = d . items ( ) <EOL> assert len ( items ) == self . _numKeys + <NUM_LIT:1> <EOL> assert type ( items ) == type ( [ ] ) <EOL> assert type ( items [ <NUM_LIT:0> ] ) == type ( ( ) ) <EOL> assert len ( items [ <NUM_LIT:0> ] ) == <NUM_LIT:2> <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> pprint ( items [ : <NUM_LIT:10> ] ) <EOL> values = d . values ( ) <EOL> assert len ( values ) == self . _numKeys + <NUM_LIT:1> <EOL> assert type ( values ) == type ( [ ] ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> pprint ( values [ : <NUM_LIT:10> ] ) <EOL> def test03_SimpleCursorStuff ( self , get_raises_error = <NUM_LIT:0> , set_raises_error = <NUM_LIT:0> ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % ( self . __class__ . __name__ , get_raises_error , set_raises_error ) <EOL> if self . env and self . dbopenflags & db . DB_AUTO_COMMIT : <EOL> txn = self . env . txn_begin ( ) <EOL> else : <EOL> txn = None <EOL> c = self . d . cursor ( txn = txn ) <EOL> rec = c . first ( ) <EOL> count = <NUM_LIT:0> <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose and count % <NUM_LIT:100> == <NUM_LIT:0> : <EOL> print rec <EOL> try : <EOL> rec = c . next ( ) <EOL> except db . DBNotFoundError , val : <EOL> if get_raises_error : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_NOTFOUND <EOL> if verbose : print val <EOL> rec = None <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> assert c . get_current_size ( ) == len ( c . current ( ) [ <NUM_LIT:1> ] ) , "<STR_LIT>" % ( c . get_current_size ( ) , c . current ( ) [ <NUM_LIT:1> ] ) <EOL> assert count == self . _numKeys <EOL> rec = c . last ( ) <EOL> count = <NUM_LIT:0> <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose and count % <NUM_LIT:100> == <NUM_LIT:0> : <EOL> print rec <EOL> try : <EOL> rec = c . prev ( ) <EOL> except db . DBNotFoundError , val : <EOL> if get_raises_error : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_NOTFOUND <EOL> if verbose : print val <EOL> rec = None <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> assert count == self . _numKeys <EOL> rec = c . set ( '<STR_LIT>' ) <EOL> rec2 = c . current ( ) <EOL> assert rec == rec2 <EOL> assert rec [ <NUM_LIT:0> ] == '<STR_LIT>' <EOL> assert rec [ <NUM_LIT:1> ] == self . makeData ( '<STR_LIT>' ) <EOL> assert c . get_current_size ( ) == len ( rec [ <NUM_LIT:1> ] ) <EOL> rec = c . set ( '<STR_LIT>' ) <EOL> assert rec [ <NUM_LIT:1> ] == '<STR_LIT>' <EOL> assert c . get_current_size ( ) == <NUM_LIT:0> <EOL> try : <EOL> n = c . set ( '<STR_LIT>' ) <EOL> except db . DBNotFoundError , val : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_NOTFOUND <EOL> if verbose : print val <EOL> else : <EOL> if set_raises_error : <EOL> self . fail ( "<STR_LIT>" ) <EOL> if n != None : <EOL> self . fail ( "<STR_LIT>" % ( n , ) ) <EOL> rec = c . get_both ( '<STR_LIT>' , self . makeData ( '<STR_LIT>' ) ) <EOL> assert rec == ( '<STR_LIT>' , self . makeData ( '<STR_LIT>' ) ) <EOL> try : <EOL> n = c . get_both ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> except db . DBNotFoundError , val : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_NOTFOUND <EOL> if verbose : print val <EOL> else : <EOL> if get_raises_error : <EOL> self . fail ( "<STR_LIT>" ) <EOL> if n != None : <EOL> self . fail ( "<STR_LIT>" % ( n , ) ) <EOL> if self . d . get_type ( ) == db . DB_BTREE : <EOL> rec = c . set_range ( '<STR_LIT>' ) <EOL> if verbose : <EOL> print "<STR_LIT>" , rec <EOL> rec = c . set_range ( '<STR_LIT>' , dlen = <NUM_LIT:0> , doff = <NUM_LIT:0> ) <EOL> if verbose : <EOL> print "<STR_LIT>" , rec <EOL> if rec [ <NUM_LIT:1> ] != '<STR_LIT>' : self . fail ( '<STR_LIT>' ) <EOL> ev = c . set_range ( '<STR_LIT>' ) <EOL> if verbose : <EOL> print "<STR_LIT>" , ev <EOL> if ev [ <NUM_LIT:1> ] != '<STR_LIT>' : self . fail ( '<STR_LIT>' ) <EOL> c . set ( '<STR_LIT>' ) <EOL> c . delete ( ) <EOL> try : <EOL> rec = c . current ( ) <EOL> except db . DBKeyEmptyError , val : <EOL> if get_raises_error : <EOL> assert val [ <NUM_LIT:0> ] == db . DB_KEYEMPTY <EOL> if verbose : print val <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> else : <EOL> if get_raises_error : <EOL> self . fail ( '<STR_LIT>' ) <EOL> c . next ( ) <EOL> c2 = c . dup ( db . DB_POSITION ) <EOL> assert c . current ( ) == c2 . current ( ) <EOL> c2 . put ( '<STR_LIT>' , '<STR_LIT>' , db . DB_CURRENT ) <EOL> assert c . current ( ) == c2 . current ( ) <EOL> assert c . current ( ) [ <NUM_LIT:1> ] == '<STR_LIT>' <EOL> c2 . put ( '<STR_LIT>' , '<STR_LIT>' , db . DB_CURRENT , dlen = <NUM_LIT:0> , doff = <NUM_LIT:5> ) <EOL> assert c2 . current ( ) [ <NUM_LIT:1> ] == '<STR_LIT>' <EOL> c . close ( ) <EOL> c2 . close ( ) <EOL> if txn : <EOL> txn . commit ( ) <EOL> methods_to_test = { <EOL> '<STR_LIT>' : ( ) , <EOL> '<STR_LIT>' : ( ) , <EOL> '<STR_LIT>' : ( db . DB_POSITION , ) , <EOL> '<STR_LIT>' : ( ) , <EOL> '<STR_LIT>' : ( <NUM_LIT:0> , ) , <EOL> '<STR_LIT>' : ( ) , <EOL> '<STR_LIT>' : ( ) , <EOL> '<STR_LIT>' : ( ) , <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' , db . DB_CURRENT ) , <EOL> '<STR_LIT>' : ( "<STR_LIT>" , ) , <EOL> } <EOL> for method , args in methods_to_test . items ( ) : <EOL> try : <EOL> if verbose : <EOL> print "<STR_LIT>" % method <EOL> apply ( getattr ( c , method ) , args ) <EOL> except db . DBError , val : <EOL> assert val [ <NUM_LIT:0> ] == <NUM_LIT:0> <EOL> if verbose : print val <EOL> else : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" % method ) <EOL> oldcursor = self . d . cursor ( txn = txn ) <EOL> self . d . close ( ) <EOL> del oldcursor <EOL> def test03b_SimpleCursorWithoutGetReturnsNone0 ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> old = self . d . set_get_returns_none ( <NUM_LIT:0> ) <EOL> assert old == <NUM_LIT:2> <EOL> self . test03_SimpleCursorStuff ( get_raises_error = <NUM_LIT:1> , set_raises_error = <NUM_LIT:1> ) <EOL> def test03b_SimpleCursorWithGetReturnsNone1 ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> old = self . d . set_get_returns_none ( <NUM_LIT:1> ) <EOL> self . test03_SimpleCursorStuff ( get_raises_error = <NUM_LIT:0> , set_raises_error = <NUM_LIT:1> ) <EOL> def test03c_SimpleCursorGetReturnsNone2 ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> old = self . d . set_get_returns_none ( <NUM_LIT:1> ) <EOL> assert old == <NUM_LIT:2> <EOL> old = self . d . set_get_returns_none ( <NUM_LIT:2> ) <EOL> assert old == <NUM_LIT:1> <EOL> self . test03_SimpleCursorStuff ( get_raises_error = <NUM_LIT:0> , set_raises_error = <NUM_LIT:0> ) <EOL> def test04_PartialGetAndPut ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> key = "<STR_LIT>" <EOL> data = "<STR_LIT:1>" * <NUM_LIT:1000> + "<STR_LIT:2>" * <NUM_LIT:1000> <EOL> d . put ( key , data ) <EOL> assert d . get ( key ) == data <EOL> assert d . get ( key , dlen = <NUM_LIT:20> , doff = <NUM_LIT> ) == ( "<STR_LIT:1>" * <NUM_LIT:10> ) + ( "<STR_LIT:2>" * <NUM_LIT:10> ) <EOL> d . put ( "<STR_LIT>" , ( "<STR_LIT:1>" * <NUM_LIT> ) + "<STR_LIT>" ) <EOL> assert d . get ( "<STR_LIT>" , dlen = <NUM_LIT:5> , doff = <NUM_LIT> ) == "<STR_LIT>" <EOL> if self . dbsetflags != db . DB_DUP : <EOL> d . put ( key , "<STR_LIT>" , dlen = <NUM_LIT> , doff = <NUM_LIT:0> ) <EOL> assert d . get ( key ) == "<STR_LIT>" <EOL> d . put ( key , "<STR_LIT>" , dlen = <NUM_LIT:1> , doff = <NUM_LIT:2> ) <EOL> assert d . get ( key ) == "<STR_LIT>" <EOL> def test05_GetSize ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" % i <EOL> d . put ( key , "<STR_LIT:1>" * i ) <EOL> assert d . get_size ( key ) == i <EOL> def test06_Truncate ( self ) : <EOL> if db . version ( ) < ( <NUM_LIT:3> , <NUM_LIT:3> ) : <EOL> return <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> d . put ( "<STR_LIT>" , "<STR_LIT>" ) ; <EOL> num = d . truncate ( ) <EOL> assert num >= <NUM_LIT:1> , "<STR_LIT>" <EOL> num = d . truncate ( ) <EOL> assert num == <NUM_LIT:0> , "<STR_LIT>" % ( num , ) <EOL> class BasicBTreeTestCase ( BasicTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> class BasicHashTestCase ( BasicTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> class BasicBTreeWithThreadFlagTestCase ( BasicTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> dbopenflags = db . DB_THREAD <EOL> class BasicHashWithThreadFlagTestCase ( BasicTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> dbopenflags = db . DB_THREAD <EOL> class BasicWithEnvTestCase ( BasicTestCase ) : <EOL> dbopenflags = db . DB_THREAD <EOL> useEnv = <NUM_LIT:1> <EOL> envflags = db . DB_THREAD | db . DB_INIT_MPOOL | db . DB_INIT_LOCK <EOL> def test07_EnvRemoveAndRename ( self ) : <EOL> if not self . env : <EOL> return <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> self . d . close ( ) <EOL> newname = self . filename + '<STR_LIT>' <EOL> self . env . dbrename ( self . filename , None , newname ) <EOL> self . env . dbremove ( newname ) <EOL> if db . version ( ) < ( <NUM_LIT:4> , <NUM_LIT:1> ) : <EOL> del test07_EnvRemoveAndRename <EOL> class BasicBTreeWithEnvTestCase ( BasicWithEnvTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> class BasicHashWithEnvTestCase ( BasicWithEnvTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> class BasicTransactionTestCase ( BasicTestCase ) : <EOL> dbopenflags = db . DB_THREAD | db . DB_AUTO_COMMIT <EOL> useEnv = <NUM_LIT:1> <EOL> envflags = ( db . DB_THREAD | db . DB_INIT_MPOOL | db . DB_INIT_LOCK | <EOL> db . DB_INIT_TXN ) <EOL> envsetflags = db . DB_AUTO_COMMIT <EOL> def tearDown ( self ) : <EOL> self . txn . commit ( ) <EOL> BasicTestCase . tearDown ( self ) <EOL> def populateDB ( self ) : <EOL> txn = self . env . txn_begin ( ) <EOL> BasicTestCase . populateDB ( self , _txn = txn ) <EOL> self . txn = self . env . txn_begin ( ) <EOL> def test06_Transactions ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> assert d . get ( '<STR_LIT>' , txn = self . txn ) == None <EOL> d . put ( '<STR_LIT>' , '<STR_LIT>' , self . txn ) <EOL> assert d . get ( '<STR_LIT>' , txn = self . txn ) == '<STR_LIT>' <EOL> self . txn . abort ( ) <EOL> assert d . get ( '<STR_LIT>' ) == None <EOL> self . txn = self . env . txn_begin ( ) <EOL> assert d . get ( '<STR_LIT>' , txn = self . txn ) == None <EOL> d . put ( '<STR_LIT>' , '<STR_LIT>' , self . txn ) <EOL> assert d . get ( '<STR_LIT>' , txn = self . txn ) == '<STR_LIT>' <EOL> self . txn . commit ( ) <EOL> assert d . get ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> self . txn = self . env . txn_begin ( ) <EOL> c = d . cursor ( self . txn ) <EOL> rec = c . first ( ) <EOL> count = <NUM_LIT:0> <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose and count % <NUM_LIT:100> == <NUM_LIT:0> : <EOL> print rec <EOL> rec = c . next ( ) <EOL> assert count == self . _numKeys + <NUM_LIT:1> <EOL> c . close ( ) <EOL> self . txn . commit ( ) <EOL> try : <EOL> self . env . txn_checkpoint ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> except db . DBIncompleteError : <EOL> pass <EOL> if db . version ( ) >= ( <NUM_LIT:4> , <NUM_LIT:0> ) : <EOL> statDict = self . env . log_stat ( <NUM_LIT:0> ) ; <EOL> assert statDict . has_key ( '<STR_LIT>' ) <EOL> assert statDict . has_key ( '<STR_LIT:version>' ) <EOL> assert statDict . has_key ( '<STR_LIT>' ) <EOL> assert statDict . has_key ( '<STR_LIT>' ) <EOL> logs = self . env . log_archive ( db . DB_ARCH_ABS | db . DB_ARCH_LOG ) <EOL> assert logs != None <EOL> for log in logs : <EOL> if verbose : <EOL> print '<STR_LIT>' + log <EOL> if db . version ( ) >= ( <NUM_LIT:4> , <NUM_LIT:2> ) : <EOL> logs = self . env . log_archive ( db . DB_ARCH_REMOVE ) <EOL> assert not logs <EOL> self . txn = self . env . txn_begin ( ) <EOL> def test07_TxnTruncate ( self ) : <EOL> if db . version ( ) < ( <NUM_LIT:3> , <NUM_LIT:3> ) : <EOL> return <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> d . put ( "<STR_LIT>" , "<STR_LIT>" ) ; <EOL> txn = self . env . txn_begin ( ) <EOL> num = d . truncate ( txn ) <EOL> assert num >= <NUM_LIT:1> , "<STR_LIT>" <EOL> num = d . truncate ( txn ) <EOL> assert num == <NUM_LIT:0> , "<STR_LIT>" % ( num , ) <EOL> txn . commit ( ) <EOL> def test08_TxnLateUse ( self ) : <EOL> txn = self . env . txn_begin ( ) <EOL> txn . abort ( ) <EOL> try : <EOL> txn . abort ( ) <EOL> except db . DBError , e : <EOL> pass <EOL> else : <EOL> raise RuntimeError , "<STR_LIT>" <EOL> txn = self . env . txn_begin ( ) <EOL> txn . commit ( ) <EOL> try : <EOL> txn . commit ( ) <EOL> except db . DBError , e : <EOL> pass <EOL> else : <EOL> raise RuntimeError , "<STR_LIT>" <EOL> class BTreeTransactionTestCase ( BasicTransactionTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> class HashTransactionTestCase ( BasicTransactionTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> class BTreeRecnoTestCase ( BasicTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> dbsetflags = db . DB_RECNUM <EOL> def test07_RecnoInBTree ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> rec = d . get ( <NUM_LIT:200> ) <EOL> assert type ( rec ) == type ( ( ) ) <EOL> assert len ( rec ) == <NUM_LIT:2> <EOL> if verbose : <EOL> print "<STR_LIT>" , rec <EOL> c = d . cursor ( ) <EOL> c . set ( '<STR_LIT>' ) <EOL> num = c . get_recno ( ) <EOL> assert type ( num ) == type ( <NUM_LIT:1> ) <EOL> if verbose : <EOL> print "<STR_LIT>" , num <EOL> rec = c . current ( ) <EOL> assert c . set_recno ( num ) == rec <EOL> c . close ( ) <EOL> class BTreeRecnoWithThreadFlagTestCase ( BTreeRecnoTestCase ) : <EOL> dbopenflags = db . DB_THREAD <EOL> class BasicDUPTestCase ( BasicTestCase ) : <EOL> dbsetflags = db . DB_DUP <EOL> def test08_DuplicateKeys ( self ) : <EOL> d = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> d . put ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> for x in "<STR_LIT>" . split ( ) : <EOL> d . put ( "<STR_LIT>" , x ) <EOL> d . put ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> data = d . get ( "<STR_LIT>" ) <EOL> assert data == "<STR_LIT>" <EOL> if verbose : <EOL> print data <EOL> c = d . cursor ( ) <EOL> rec = c . set ( "<STR_LIT>" ) <EOL> assert rec == ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> next = c . next ( ) <EOL> assert next == ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> rec = c . set ( "<STR_LIT>" ) <EOL> count = c . count ( ) <EOL> assert count == <NUM_LIT:9> <EOL> next_dup = c . next_dup ( ) <EOL> assert next_dup == ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> rec = c . set ( '<STR_LIT>' ) <EOL> while rec is not None : <EOL> if verbose : <EOL> print rec <EOL> rec = c . next_dup ( ) <EOL> c . set ( '<STR_LIT>' ) <EOL> rec = c . next_nodup ( ) <EOL> assert rec [ <NUM_LIT:0> ] != '<STR_LIT>' <EOL> if verbose : <EOL> print rec <EOL> c . close ( ) <EOL> class BTreeDUPTestCase ( BasicDUPTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> class HashDUPTestCase ( BasicDUPTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> class BTreeDUPWithThreadTestCase ( BasicDUPTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> dbopenflags = db . DB_THREAD <EOL> class HashDUPWithThreadTestCase ( BasicDUPTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> dbopenflags = db . DB_THREAD <EOL> class BasicMultiDBTestCase ( BasicTestCase ) : <EOL> dbname = '<STR_LIT>' <EOL> def otherType ( self ) : <EOL> if self . dbtype == db . DB_BTREE : <EOL> return db . DB_HASH <EOL> else : <EOL> return db . DB_BTREE <EOL> def test09_MultiDB ( self ) : <EOL> d1 = self . d <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> d2 = db . DB ( self . env ) <EOL> d2 . open ( self . filename , "<STR_LIT>" , self . dbtype , <EOL> self . dbopenflags | db . DB_CREATE ) <EOL> d3 = db . DB ( self . env ) <EOL> d3 . open ( self . filename , "<STR_LIT>" , self . otherType ( ) , <EOL> self . dbopenflags | db . DB_CREATE ) <EOL> for x in "<STR_LIT>" . split ( ) : <EOL> d2 . put ( x , self . makeData ( x ) ) <EOL> for x in string . letters : <EOL> d3 . put ( x , x * <NUM_LIT> ) <EOL> d1 . sync ( ) <EOL> d2 . sync ( ) <EOL> d3 . sync ( ) <EOL> d1 . close ( ) <EOL> d2 . close ( ) <EOL> d3 . close ( ) <EOL> self . d = d1 = d2 = d3 = None <EOL> self . d = d1 = db . DB ( self . env ) <EOL> d1 . open ( self . filename , self . dbname , flags = self . dbopenflags ) <EOL> d2 = db . DB ( self . env ) <EOL> d2 . open ( self . filename , "<STR_LIT>" , flags = self . dbopenflags ) <EOL> d3 = db . DB ( self . env ) <EOL> d3 . open ( self . filename , "<STR_LIT>" , flags = self . dbopenflags ) <EOL> c1 = d1 . cursor ( ) <EOL> c2 = d2 . cursor ( ) <EOL> c3 = d3 . cursor ( ) <EOL> count = <NUM_LIT:0> <EOL> rec = c1 . first ( ) <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose and ( count % <NUM_LIT:50> ) == <NUM_LIT:0> : <EOL> print rec <EOL> rec = c1 . next ( ) <EOL> assert count == self . _numKeys <EOL> count = <NUM_LIT:0> <EOL> rec = c2 . first ( ) <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose : <EOL> print rec <EOL> rec = c2 . next ( ) <EOL> assert count == <NUM_LIT:9> <EOL> count = <NUM_LIT:0> <EOL> rec = c3 . first ( ) <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose : <EOL> print rec <EOL> rec = c3 . next ( ) <EOL> assert count == <NUM_LIT> <EOL> c1 . close ( ) <EOL> c2 . close ( ) <EOL> c3 . close ( ) <EOL> d2 . close ( ) <EOL> d3 . close ( ) <EOL> class BTreeMultiDBTestCase ( BasicMultiDBTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> dbopenflags = db . DB_THREAD <EOL> useEnv = <NUM_LIT:1> <EOL> envflags = db . DB_THREAD | db . DB_INIT_MPOOL | db . DB_INIT_LOCK <EOL> class HashMultiDBTestCase ( BasicMultiDBTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> dbopenflags = db . DB_THREAD <EOL> useEnv = <NUM_LIT:1> <EOL> envflags = db . DB_THREAD | db . DB_INIT_MPOOL | db . DB_INIT_LOCK <EOL> def test_suite ( ) : <EOL> suite = unittest . TestSuite ( ) <EOL> suite . addTest ( unittest . makeSuite ( VersionTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BasicBTreeTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BasicHashTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BasicBTreeWithThreadFlagTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BasicHashWithThreadFlagTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BasicBTreeWithEnvTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BasicHashWithEnvTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeTransactionTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( HashTransactionTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeRecnoTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeRecnoWithThreadFlagTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeDUPTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( HashDUPTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeDUPWithThreadTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( HashDUPWithThreadTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeMultiDBTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( HashMultiDBTestCase ) ) <EOL> return suite <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
