#NEWLINE# def _run_test( self ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# database_resource = provision . DatabaseResource ( self . DRIVER ) #NEWLINE# #UNINDENT# except exception . BackendNotAvailable : #NEWLINE# #INDENT# self . skip ( 'database not available' ) #NEWLINE# #UNINDENT# schema_resource = provision . SchemaResource ( transaction_resource , self . _gen_schema ) #NEWLINE# transaction_resource = provision . TransactionResource ( database_resource , schema_resource ) #NEWLINE# engine = transaction_resource . getResource ( ) #NEWLINE# with engine . connect ( ) as conn : #NEWLINE# #INDENT# rows = conn . execute ( self . test_table . select ( ) ) #NEWLINE# self . assertEqual ( rows . fetchall ( ) , [ ] ) #NEWLINE# trans = conn . begin ( ) #NEWLINE# conn . execute ( self . test_table . insert ( ) , { #NEWLINE# #INDENT# 'x' : 1 , #NEWLINE# 'y' : 2 , #NEWLINE# #UNINDENT# } ) #NEWLINE# trans . rollback ( ) #NEWLINE# rows = conn . execute ( self . test_table . select ( ) ) #NEWLINE# self . assertEqual ( rows . fetchall ( ) , [ ] ) #NEWLINE# trans = conn . begin ( ) #NEWLINE# conn . execute ( self . test_table . insert ( ) , { #NEWLINE# #INDENT# 'x' : 2 , #NEWLINE# 'y' : 3 , #NEWLINE# #UNINDENT# } ) #NEWLINE# trans . commit ( ) #NEWLINE# rows = conn . execute ( self . test_table . select ( ) ) #NEWLINE# self . assertEqual ( rows . fetchall ( ) , [ ( 2 , 3 ) ] ) #NEWLINE# #UNINDENT# transaction_resource . finishedWith ( engine )