<s> """<STR_LIT>""" <EOL> import sys , os , re <EOL> import test_all <EOL> from cStringIO import StringIO <EOL> import unittest <EOL> from test_all import db , dbshelve , test_support , get_new_environment_path , get_new_database_path <EOL> def cmp ( a , b ) : <EOL> if a == b : return <NUM_LIT:0> <EOL> if a < b : return - <NUM_LIT:1> <EOL> return <NUM_LIT:1> <EOL> lexical_cmp = cmp <EOL> def lowercase_cmp ( left , right ) : <EOL> return cmp ( left . lower ( ) , right . lower ( ) ) <EOL> def make_reverse_comparator ( cmp ) : <EOL> def reverse ( left , right , delegate = cmp ) : <EOL> return - delegate ( left , right ) <EOL> return reverse <EOL> _expected_lexical_test_data = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT:b>' , '<STR_LIT:c>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> _expected_lowercase_test_data = [ '<STR_LIT>' , '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT:b>' , '<STR_LIT:c>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class ComparatorTests ( unittest . TestCase ) : <EOL> def comparator_test_helper ( self , comparator , expected_data ) : <EOL> data = expected_data [ : ] <EOL> import sys <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:6> ) : <EOL> data . sort ( cmp = comparator ) <EOL> else : <EOL> data2 = [ ] <EOL> for i in data : <EOL> for j , k in enumerate ( data2 ) : <EOL> r = comparator ( k , i ) <EOL> if r == <NUM_LIT:1> : <EOL> data2 . insert ( j , i ) <EOL> break <EOL> else : <EOL> data2 . append ( i ) <EOL> data = data2 <EOL> self . assertEqual ( data , expected_data , <EOL> "<STR_LIT>" <EOL> % ( comparator , expected_data , data ) ) <EOL> def test_lexical_comparator ( self ) : <EOL> self . comparator_test_helper ( lexical_cmp , _expected_lexical_test_data ) <EOL> def test_reverse_lexical_comparator ( self ) : <EOL> rev = _expected_lexical_test_data [ : ] <EOL> rev . reverse ( ) <EOL> self . comparator_test_helper ( make_reverse_comparator ( lexical_cmp ) , <EOL> rev ) <EOL> def test_lowercase_comparator ( self ) : <EOL> self . comparator_test_helper ( lowercase_cmp , <EOL> _expected_lowercase_test_data ) <EOL> class AbstractBtreeKeyCompareTestCase ( unittest . TestCase ) : <EOL> env = None <EOL> db = None <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 assertLess ( self , a , b , msg = None ) : <EOL> return self . assertTrue ( a < b , msg = msg ) <EOL> def setUp ( self ) : <EOL> self . filename = self . __class__ . __name__ + '<STR_LIT>' <EOL> self . homeDir = get_new_environment_path ( ) <EOL> env = db . DBEnv ( ) <EOL> env . open ( self . homeDir , <EOL> db . DB_CREATE | db . DB_INIT_MPOOL <EOL> | db . DB_INIT_LOCK | db . DB_THREAD ) <EOL> self . env = env <EOL> def tearDown ( self ) : <EOL> self . closeDB ( ) <EOL> if self . env is not None : <EOL> self . env . close ( ) <EOL> self . env = None <EOL> test_support . rmtree ( self . homeDir ) <EOL> def addDataToDB ( self , data ) : <EOL> i = <NUM_LIT:0> <EOL> for item in data : <EOL> self . db . put ( item , str ( i ) ) <EOL> i = i + <NUM_LIT:1> <EOL> def createDB ( self , key_comparator ) : <EOL> self . db = db . DB ( self . env ) <EOL> self . setupDB ( key_comparator ) <EOL> self . db . open ( self . filename , "<STR_LIT:test>" , db . DB_BTREE , db . DB_CREATE ) <EOL> def setupDB ( self , key_comparator ) : <EOL> self . db . set_bt_compare ( key_comparator ) <EOL> def closeDB ( self ) : <EOL> if self . db is not None : <EOL> self . db . close ( ) <EOL> self . db = None <EOL> def startTest ( self ) : <EOL> pass <EOL> def finishTest ( self , expected = None ) : <EOL> if expected is not None : <EOL> self . check_results ( expected ) <EOL> self . closeDB ( ) <EOL> def check_results ( self , expected ) : <EOL> curs = self . db . cursor ( ) <EOL> try : <EOL> index = <NUM_LIT:0> <EOL> rec = curs . first ( ) <EOL> while rec : <EOL> key , ignore = rec <EOL> self . assertLess ( index , len ( expected ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( expected [ index ] , key , <EOL> "<STR_LIT>" <EOL> % ( expected [ index ] , index , key ) ) <EOL> index = index + <NUM_LIT:1> <EOL> rec = curs . next ( ) <EOL> self . assertEqual ( index , len ( expected ) , <EOL> "<STR_LIT>" ) <EOL> finally : <EOL> curs . close ( ) <EOL> class BtreeKeyCompareTestCase ( AbstractBtreeKeyCompareTestCase ) : <EOL> def runCompareTest ( self , comparator , data ) : <EOL> self . startTest ( ) <EOL> self . createDB ( comparator ) <EOL> self . addDataToDB ( data ) <EOL> self . finishTest ( data ) <EOL> def test_lexical_ordering ( self ) : <EOL> self . runCompareTest ( lexical_cmp , _expected_lexical_test_data ) <EOL> def test_reverse_lexical_ordering ( self ) : <EOL> expected_rev_data = _expected_lexical_test_data [ : ] <EOL> expected_rev_data . reverse ( ) <EOL> self . runCompareTest ( make_reverse_comparator ( lexical_cmp ) , <EOL> expected_rev_data ) <EOL> def test_compare_function_useless ( self ) : <EOL> self . startTest ( ) <EOL> def socialist_comparator ( l , r ) : <EOL> return <NUM_LIT:0> <EOL> self . createDB ( socialist_comparator ) <EOL> self . addDataToDB ( [ '<STR_LIT:b>' , '<STR_LIT:a>' , '<STR_LIT:d>' ] ) <EOL> self . finishTest ( [ '<STR_LIT:b>' ] ) <EOL> class BtreeExceptionsTestCase ( AbstractBtreeKeyCompareTestCase ) : <EOL> def test_raises_non_callable ( self ) : <EOL> self . startTest ( ) <EOL> self . assertRaises ( TypeError , self . createDB , '<STR_LIT:abc>' ) <EOL> self . assertRaises ( TypeError , self . createDB , None ) <EOL> self . finishTest ( ) <EOL> def test_set_bt_compare_with_function ( self ) : <EOL> self . startTest ( ) <EOL> self . createDB ( lexical_cmp ) <EOL> self . finishTest ( ) <EOL> def check_results ( self , results ) : <EOL> pass <EOL> def test_compare_function_incorrect ( self ) : <EOL> self . startTest ( ) <EOL> def bad_comparator ( l , r ) : <EOL> return <NUM_LIT:1> <EOL> self . assertRaises ( TypeError , self . createDB , bad_comparator ) <EOL> self . finishTest ( ) <EOL> def verifyStderr ( self , method , successRe ) : <EOL> """<STR_LIT>""" <EOL> stdErr = sys . stderr <EOL> sys . stderr = StringIO ( ) <EOL> try : <EOL> method ( ) <EOL> finally : <EOL> temp = sys . stderr <EOL> sys . stderr = stdErr <EOL> errorOut = temp . getvalue ( ) <EOL> if not successRe . search ( errorOut ) : <EOL> self . fail ( "<STR_LIT>" + errorOut ) <EOL> if sys . version_info < ( <NUM_LIT:3> , <NUM_LIT:0> ) : <EOL> sys . exc_traceback = sys . last_traceback = None <EOL> def _test_compare_function_exception ( self ) : <EOL> self . startTest ( ) <EOL> def bad_comparator ( l , r ) : <EOL> if l == r : <EOL> return <NUM_LIT:0> <EOL> raise RuntimeError , "<STR_LIT>" <EOL> self . createDB ( bad_comparator ) <EOL> self . addDataToDB ( [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> self . finishTest ( ) <EOL> def test_compare_function_exception ( self ) : <EOL> self . verifyStderr ( <EOL> self . _test_compare_function_exception , <EOL> re . compile ( '<STR_LIT>' , re . M | re . S ) <EOL> ) <EOL> def _test_compare_function_bad_return ( self ) : <EOL> self . startTest ( ) <EOL> def bad_comparator ( l , r ) : <EOL> if l == r : <EOL> return <NUM_LIT:0> <EOL> return l <EOL> self . createDB ( bad_comparator ) <EOL> self . addDataToDB ( [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> self . finishTest ( ) <EOL> def test_compare_function_bad_return ( self ) : <EOL> self . verifyStderr ( <EOL> self . _test_compare_function_bad_return , <EOL> re . compile ( '<STR_LIT>' , re . M | re . S ) <EOL> ) <EOL> def test_cannot_assign_twice ( self ) : <EOL> def my_compare ( a , b ) : <EOL> return <NUM_LIT:0> <EOL> self . startTest ( ) <EOL> self . createDB ( my_compare ) <EOL> self . assertRaises ( RuntimeError , self . db . set_bt_compare , my_compare ) <EOL> class AbstractDuplicateCompareTestCase ( unittest . TestCase ) : <EOL> env = None <EOL> db = None <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 assertLess ( self , a , b , msg = None ) : <EOL> return self . assertTrue ( a < b , msg = msg ) <EOL> def setUp ( self ) : <EOL> self . filename = self . __class__ . __name__ + '<STR_LIT>' <EOL> self . homeDir = get_new_environment_path ( ) <EOL> env = db . DBEnv ( ) <EOL> env . open ( self . homeDir , <EOL> db . DB_CREATE | db . DB_INIT_MPOOL <EOL> | db . DB_INIT_LOCK | db . DB_THREAD ) <EOL> self . env = env <EOL> def tearDown ( self ) : <EOL> self . closeDB ( ) <EOL> if self . env is not None : <EOL> self . env . close ( ) <EOL> self . env = None <EOL> test_support . rmtree ( self . homeDir ) <EOL> def addDataToDB ( self , data ) : <EOL> for item in data : <EOL> self . db . put ( "<STR_LIT:key>" , item ) <EOL> def createDB ( self , dup_comparator ) : <EOL> self . db = db . DB ( self . env ) <EOL> self . setupDB ( dup_comparator ) <EOL> self . db . open ( self . filename , "<STR_LIT:test>" , db . DB_BTREE , db . DB_CREATE ) <EOL> def setupDB ( self , dup_comparator ) : <EOL> self . db . set_flags ( db . DB_DUPSORT ) <EOL> self . db . set_dup_compare ( dup_comparator ) <EOL> def closeDB ( self ) : <EOL> if self . db is not None : <EOL> self . db . close ( ) <EOL> self . db = None <EOL> def startTest ( self ) : <EOL> pass <EOL> def finishTest ( self , expected = None ) : <EOL> if expected is not None : <EOL> self . check_results ( expected ) <EOL> self . closeDB ( ) <EOL> def check_results ( self , expected ) : <EOL> curs = self . db . cursor ( ) <EOL> try : <EOL> index = <NUM_LIT:0> <EOL> rec = curs . first ( ) <EOL> while rec : <EOL> ignore , data = rec <EOL> self . assertLess ( index , len ( expected ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( expected [ index ] , data , <EOL> "<STR_LIT>" <EOL> % ( expected [ index ] , index , data ) ) <EOL> index = index + <NUM_LIT:1> <EOL> rec = curs . next ( ) <EOL> self . assertEqual ( index , len ( expected ) , <EOL> "<STR_LIT>" ) <EOL> finally : <EOL> curs . close ( ) <EOL> class DuplicateCompareTestCase ( AbstractDuplicateCompareTestCase ) : <EOL> def runCompareTest ( self , comparator , data ) : <EOL> self . startTest ( ) <EOL> self . createDB ( comparator ) <EOL> self . addDataToDB ( data ) <EOL> self . finishTest ( data ) <EOL> def test_lexical_ordering ( self ) : <EOL> self . runCompareTest ( lexical_cmp , _expected_lexical_test_data ) <EOL> def test_reverse_lexical_ordering ( self ) : <EOL> expected_rev_data = _expected_lexical_test_data [ : ] <EOL> expected_rev_data . reverse ( ) <EOL> self . runCompareTest ( make_reverse_comparator ( lexical_cmp ) , <EOL> expected_rev_data ) <EOL> class DuplicateExceptionsTestCase ( AbstractDuplicateCompareTestCase ) : <EOL> def test_raises_non_callable ( self ) : <EOL> self . startTest ( ) <EOL> self . assertRaises ( TypeError , self . createDB , '<STR_LIT:abc>' ) <EOL> self . assertRaises ( TypeError , self . createDB , None ) <EOL> self . finishTest ( ) <EOL> def test_set_dup_compare_with_function ( self ) : <EOL> self . startTest ( ) <EOL> self . createDB ( lexical_cmp ) <EOL> self . finishTest ( ) <EOL> def check_results ( self , results ) : <EOL> pass <EOL> def test_compare_function_incorrect ( self ) : <EOL> self . startTest ( ) <EOL> def bad_comparator ( l , r ) : <EOL> return <NUM_LIT:1> <EOL> self . assertRaises ( TypeError , self . createDB , bad_comparator ) <EOL> self . finishTest ( ) <EOL> def test_compare_function_useless ( self ) : <EOL> self . startTest ( ) <EOL> def socialist_comparator ( l , r ) : <EOL> return <NUM_LIT:0> <EOL> self . createDB ( socialist_comparator ) <EOL> self . assertRaises ( db . DBKeyExistError , self . addDataToDB , [ '<STR_LIT:b>' , '<STR_LIT:a>' , '<STR_LIT:d>' ] ) <EOL> self . finishTest ( ) <EOL> def verifyStderr ( self , method , successRe ) : <EOL> """<STR_LIT>""" <EOL> stdErr = sys . stderr <EOL> sys . stderr = StringIO ( ) <EOL> try : <EOL> method ( ) <EOL> finally : <EOL> temp = sys . stderr <EOL> sys . stderr = stdErr <EOL> errorOut = temp . getvalue ( ) <EOL> if not successRe . search ( errorOut ) : <EOL> self . fail ( "<STR_LIT>" + errorOut ) <EOL> if sys . version_info < ( <NUM_LIT:3> , <NUM_LIT:0> ) : <EOL> sys . exc_traceback = sys . last_traceback = None <EOL> def _test_compare_function_exception ( self ) : <EOL> self . startTest ( ) <EOL> def bad_comparator ( l , r ) : <EOL> if l == r : <EOL> return <NUM_LIT:0> <EOL> raise RuntimeError , "<STR_LIT>" <EOL> self . createDB ( bad_comparator ) <EOL> self . addDataToDB ( [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> self . finishTest ( ) <EOL> def test_compare_function_exception ( self ) : <EOL> self . verifyStderr ( <EOL> self . _test_compare_function_exception , <EOL> re . compile ( '<STR_LIT>' , re . M | re . S ) <EOL> ) <EOL> def _test_compare_function_bad_return ( self ) : <EOL> self . startTest ( ) <EOL> def bad_comparator ( l , r ) : <EOL> if l == r : <EOL> return <NUM_LIT:0> <EOL> return l <EOL> self . createDB ( bad_comparator ) <EOL> self . addDataToDB ( [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> self . finishTest ( ) <EOL> def test_compare_function_bad_return ( self ) : <EOL> self . verifyStderr ( <EOL> self . _test_compare_function_bad_return , <EOL> re . compile ( '<STR_LIT>' , re . M | re . S ) <EOL> ) <EOL> def test_cannot_assign_twice ( self ) : <EOL> def my_compare ( a , b ) : <EOL> return <NUM_LIT:0> <EOL> self . startTest ( ) <EOL> self . createDB ( my_compare ) <EOL> self . assertRaises ( RuntimeError , self . db . set_dup_compare , my_compare ) <EOL> def test_suite ( ) : <EOL> res = unittest . TestSuite ( ) <EOL> res . addTest ( unittest . makeSuite ( ComparatorTests ) ) <EOL> res . addTest ( unittest . makeSuite ( BtreeExceptionsTestCase ) ) <EOL> res . addTest ( unittest . makeSuite ( BtreeKeyCompareTestCase ) ) <EOL> res . addTest ( unittest . makeSuite ( DuplicateExceptionsTestCase ) ) <EOL> res . addTest ( unittest . makeSuite ( DuplicateCompareTestCase ) ) <EOL> return res <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
