<s> """<STR_LIT>""" <EOL> import os , string , sys <EOL> import random <EOL> import unittest <EOL> from test_all import db , dbshelve , test_support , verbose , get_new_environment_path , get_new_database_path <EOL> class DataClass : <EOL> def __init__ ( self ) : <EOL> self . value = random . random ( ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % self . value <EOL> def __cmp__ ( self , other ) : <EOL> return cmp ( self . value , other ) <EOL> class DBShelveTestCase ( unittest . TestCase ) : <EOL> if ( sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) ) or ( ( sys . version_info >= ( <NUM_LIT:3> , <NUM_LIT:0> ) ) and <EOL> ( sys . version_info < ( <NUM_LIT:3> , <NUM_LIT:2> ) ) ) : <EOL> def assertIn ( self , a , b , msg = None ) : <EOL> return self . assertTrue ( a in b , msg = msg ) <EOL> def setUp ( self ) : <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> from test_all import do_proxy_db_py3k <EOL> self . _flag_proxy_db_py3k = do_proxy_db_py3k ( False ) <EOL> self . filename = get_new_database_path ( ) <EOL> self . do_open ( ) <EOL> def tearDown ( self ) : <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> from test_all import do_proxy_db_py3k <EOL> do_proxy_db_py3k ( self . _flag_proxy_db_py3k ) <EOL> self . do_close ( ) <EOL> test_support . unlink ( self . filename ) <EOL> def mk ( self , key ) : <EOL> """<STR_LIT>""" <EOL> if sys . version_info [ <NUM_LIT:0> ] < <NUM_LIT:3> : <EOL> return key <EOL> else : <EOL> return bytes ( key , "<STR_LIT>" ) <EOL> def populateDB ( self , d ) : <EOL> for x in string . letters : <EOL> d [ self . mk ( '<STR_LIT:S>' + x ) ] = <NUM_LIT:10> * x <EOL> d [ self . mk ( '<STR_LIT:I>' + x ) ] = ord ( x ) <EOL> d [ self . mk ( '<STR_LIT:L>' + x ) ] = [ x ] * <NUM_LIT:10> <EOL> inst = DataClass ( ) <EOL> inst . S = <NUM_LIT:10> * x <EOL> inst . I = ord ( x ) <EOL> inst . L = [ x ] * <NUM_LIT:10> <EOL> d [ self . mk ( '<STR_LIT:O>' + x ) ] = inst <EOL> def do_open ( self ) : <EOL> self . d = dbshelve . open ( self . filename ) <EOL> def do_close ( self ) : <EOL> self . d . close ( ) <EOL> def test01_basics ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> self . populateDB ( self . d ) <EOL> self . d . sync ( ) <EOL> self . do_close ( ) <EOL> self . do_open ( ) <EOL> d = self . d <EOL> l = len ( d ) <EOL> k = d . keys ( ) <EOL> s = d . stat ( ) <EOL> f = d . fd ( ) <EOL> if verbose : <EOL> print "<STR_LIT>" , l <EOL> print "<STR_LIT>" , k <EOL> print "<STR_LIT>" , s <EOL> self . assertEqual ( <NUM_LIT:0> , d . has_key ( self . mk ( '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , d . has_key ( self . mk ( '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , d . has_key ( self . mk ( '<STR_LIT>' ) ) ) <EOL> d . delete ( self . mk ( '<STR_LIT>' ) ) <EOL> del d [ self . mk ( '<STR_LIT>' ) ] <EOL> self . assertEqual ( <NUM_LIT:0> , d . has_key ( self . mk ( '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , d . has_key ( self . mk ( '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( len ( d ) , l - <NUM_LIT:2> ) <EOL> values = [ ] <EOL> for key in d . keys ( ) : <EOL> value = d [ key ] <EOL> values . append ( value ) <EOL> if verbose : <EOL> print "<STR_LIT>" % ( key , value ) <EOL> self . checkrec ( key , value ) <EOL> dbvalues = d . values ( ) <EOL> self . assertEqual ( len ( dbvalues ) , len ( d . keys ( ) ) ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:6> ) : <EOL> values . sort ( ) <EOL> dbvalues . sort ( ) <EOL> self . assertEqual ( values , dbvalues ) <EOL> else : <EOL> values . sort ( key = lambda x : str ( x ) ) <EOL> dbvalues . sort ( key = lambda x : str ( x ) ) <EOL> self . assertEqual ( repr ( values ) , repr ( dbvalues ) ) <EOL> items = d . items ( ) <EOL> self . assertEqual ( len ( items ) , len ( values ) ) <EOL> for key , value in items : <EOL> self . checkrec ( key , value ) <EOL> self . assertEqual ( d . get ( self . mk ( '<STR_LIT>' ) ) , None ) <EOL> self . assertEqual ( d . get ( self . mk ( '<STR_LIT>' ) , None ) , None ) <EOL> self . assertEqual ( d . get ( self . mk ( '<STR_LIT>' ) , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( d . get ( self . mk ( '<STR_LIT>' ) , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> d . set_get_returns_none ( <NUM_LIT:0> ) <EOL> self . assertRaises ( db . DBNotFoundError , d . get , self . mk ( '<STR_LIT>' ) ) <EOL> d . set_get_returns_none ( <NUM_LIT:1> ) <EOL> d . put ( self . mk ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( d . get ( self . mk ( '<STR_LIT>' ) ) , '<STR_LIT>' ) <EOL> self . assertEqual ( d [ self . mk ( '<STR_LIT>' ) ] , '<STR_LIT>' ) <EOL> def test02_cursors ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> self . populateDB ( self . d ) <EOL> d = self . d <EOL> count = <NUM_LIT:0> <EOL> c = d . cursor ( ) <EOL> rec = c . first ( ) <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose : <EOL> print rec <EOL> key , value = rec <EOL> self . checkrec ( key , value ) <EOL> rec = getattr ( c , "<STR_LIT>" ) ( ) <EOL> del c <EOL> self . assertEqual ( count , len ( d ) ) <EOL> count = <NUM_LIT:0> <EOL> c = d . cursor ( ) <EOL> rec = c . last ( ) <EOL> while rec is not None : <EOL> count = count + <NUM_LIT:1> <EOL> if verbose : <EOL> print rec <EOL> key , value = rec <EOL> self . checkrec ( key , value ) <EOL> rec = c . prev ( ) <EOL> self . assertEqual ( count , len ( d ) ) <EOL> c . set ( self . mk ( '<STR_LIT>' ) ) <EOL> key , value = c . current ( ) <EOL> self . checkrec ( key , value ) <EOL> del c <EOL> def test03_append ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> self . assertRaises ( dbshelve . DBShelveError , <EOL> self . d . append , '<STR_LIT>' ) <EOL> def test04_iterable ( self ) : <EOL> self . populateDB ( self . d ) <EOL> d = self . d <EOL> keys = d . keys ( ) <EOL> keyset = set ( keys ) <EOL> self . assertEqual ( len ( keyset ) , len ( keys ) ) <EOL> for key in d : <EOL> self . assertIn ( key , keyset ) <EOL> keyset . remove ( key ) <EOL> self . assertEqual ( len ( keyset ) , <NUM_LIT:0> ) <EOL> def checkrec ( self , key , value ) : <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> if isinstance ( key , bytes ) : <EOL> key = key . decode ( "<STR_LIT>" ) <EOL> x = key [ <NUM_LIT:1> ] <EOL> if key [ <NUM_LIT:0> ] == '<STR_LIT:S>' : <EOL> self . assertEqual ( type ( value ) , str ) <EOL> self . assertEqual ( value , <NUM_LIT:10> * x ) <EOL> elif key [ <NUM_LIT:0> ] == '<STR_LIT:I>' : <EOL> self . assertEqual ( type ( value ) , int ) <EOL> self . assertEqual ( value , ord ( x ) ) <EOL> elif key [ <NUM_LIT:0> ] == '<STR_LIT:L>' : <EOL> self . assertEqual ( type ( value ) , list ) <EOL> self . assertEqual ( value , [ x ] * <NUM_LIT:10> ) <EOL> elif key [ <NUM_LIT:0> ] == '<STR_LIT:O>' : <EOL> if sys . version_info [ <NUM_LIT:0> ] < <NUM_LIT:3> : <EOL> from types import InstanceType <EOL> self . assertEqual ( type ( value ) , InstanceType ) <EOL> else : <EOL> self . assertEqual ( type ( value ) , DataClass ) <EOL> self . assertEqual ( value . S , <NUM_LIT:10> * x ) <EOL> self . assertEqual ( value . I , ord ( x ) ) <EOL> self . assertEqual ( value . L , [ x ] * <NUM_LIT:10> ) <EOL> else : <EOL> self . assertTrue ( <NUM_LIT:0> , '<STR_LIT>' ) <EOL> class BasicShelveTestCase ( DBShelveTestCase ) : <EOL> def do_open ( self ) : <EOL> self . d = dbshelve . DBShelf ( ) <EOL> self . d . open ( self . filename , self . dbtype , self . dbflags ) <EOL> def do_close ( self ) : <EOL> self . d . close ( ) <EOL> class BTreeShelveTestCase ( BasicShelveTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> dbflags = db . DB_CREATE <EOL> class HashShelveTestCase ( BasicShelveTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> dbflags = db . DB_CREATE <EOL> class ThreadBTreeShelveTestCase ( BasicShelveTestCase ) : <EOL> dbtype = db . DB_BTREE <EOL> dbflags = db . DB_CREATE | db . DB_THREAD <EOL> class ThreadHashShelveTestCase ( BasicShelveTestCase ) : <EOL> dbtype = db . DB_HASH <EOL> dbflags = db . DB_CREATE | db . DB_THREAD <EOL> class BasicEnvShelveTestCase ( DBShelveTestCase ) : <EOL> def do_open ( self ) : <EOL> self . env = db . DBEnv ( ) <EOL> self . env . open ( self . homeDir , <EOL> self . envflags | db . DB_INIT_MPOOL | db . DB_CREATE ) <EOL> self . filename = os . path . split ( self . filename ) [ <NUM_LIT:1> ] <EOL> self . d = dbshelve . DBShelf ( self . env ) <EOL> self . d . open ( self . filename , self . dbtype , self . dbflags ) <EOL> def do_close ( self ) : <EOL> self . d . close ( ) <EOL> self . env . close ( ) <EOL> def setUp ( self ) : <EOL> self . homeDir = get_new_environment_path ( ) <EOL> DBShelveTestCase . setUp ( self ) <EOL> def tearDown ( self ) : <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> from test_all import do_proxy_db_py3k <EOL> do_proxy_db_py3k ( self . _flag_proxy_db_py3k ) <EOL> self . do_close ( ) <EOL> test_support . rmtree ( self . homeDir ) <EOL> class EnvBTreeShelveTestCase ( BasicEnvShelveTestCase ) : <EOL> envflags = <NUM_LIT:0> <EOL> dbtype = db . DB_BTREE <EOL> dbflags = db . DB_CREATE <EOL> class EnvHashShelveTestCase ( BasicEnvShelveTestCase ) : <EOL> envflags = <NUM_LIT:0> <EOL> dbtype = db . DB_HASH <EOL> dbflags = db . DB_CREATE <EOL> class EnvThreadBTreeShelveTestCase ( BasicEnvShelveTestCase ) : <EOL> envflags = db . DB_THREAD <EOL> dbtype = db . DB_BTREE <EOL> dbflags = db . DB_CREATE | db . DB_THREAD <EOL> class EnvThreadHashShelveTestCase ( BasicEnvShelveTestCase ) : <EOL> envflags = db . DB_THREAD <EOL> dbtype = db . DB_HASH <EOL> dbflags = db . DB_CREATE | db . DB_THREAD <EOL> class RecNoShelveTestCase ( BasicShelveTestCase ) : <EOL> dbtype = db . DB_RECNO <EOL> dbflags = db . DB_CREATE <EOL> def setUp ( self ) : <EOL> BasicShelveTestCase . setUp ( self ) <EOL> self . key_pool = list ( range ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> self . key_map = { } <EOL> self . intkey_map = { } <EOL> def mk ( self , key ) : <EOL> if key not in self . key_map : <EOL> self . key_map [ key ] = self . key_pool . pop ( <NUM_LIT:0> ) <EOL> self . intkey_map [ self . key_map [ key ] ] = key <EOL> return self . key_map [ key ] <EOL> def checkrec ( self , intkey , value ) : <EOL> key = self . intkey_map [ intkey ] <EOL> BasicShelveTestCase . checkrec ( self , key , value ) <EOL> def test03_append ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT:\n>' , '<STR_LIT>' * <NUM_LIT:30> <EOL> print "<STR_LIT>" % self . __class__ . __name__ <EOL> self . d [ <NUM_LIT:1> ] = '<STR_LIT>' <EOL> self . d [ <NUM_LIT:5> ] = '<STR_LIT>' <EOL> self . assertEqual ( <NUM_LIT:6> , self . d . append ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( <NUM_LIT:7> , self . d . append ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , self . d . get ( <NUM_LIT:6> ) ) <EOL> self . assertEqual ( '<STR_LIT>' , self . d . get ( <NUM_LIT:1> ) ) <EOL> self . assertEqual ( '<STR_LIT>' , self . d . get ( <NUM_LIT:7> ) ) <EOL> self . assertEqual ( '<STR_LIT>' , self . d . get ( <NUM_LIT:5> ) ) <EOL> def test_suite ( ) : <EOL> suite = unittest . TestSuite ( ) <EOL> suite . addTest ( unittest . makeSuite ( DBShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( BTreeShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( HashShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( ThreadBTreeShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( ThreadHashShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( EnvBTreeShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( EnvHashShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( EnvThreadBTreeShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( EnvThreadHashShelveTestCase ) ) <EOL> suite . addTest ( unittest . makeSuite ( RecNoShelveTestCase ) ) <EOL> return suite <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
