@@ -238,3 +238,83 @@ def testStringFormat(self):
238238 (self .username , self .tnsentry )
239239 self .assertEqual (str (connection ), expectedValue )
240240
241+ def testCtxMgrCommitOnSuccess (self ):
242+ "test context manager - commit on success"
243+ connection = cx_Oracle .connect (self .username , self .password ,
244+ self .tnsentry )
245+ cursor = connection .cursor ()
246+ cursor .execute ("truncate table TestTempTable" )
247+ with connection :
248+ cursor .execute ("""
249+ insert into TestTempTable (IntCol, StringCol)
250+ values (1, null)""" )
251+ connection .rollback ()
252+ cursor .execute ("select count(*) from TestTempTable" )
253+ count , = cursor .fetchone ()
254+ self .assertEqual (count , 1 )
255+
256+ def testCtxMgrRollbackOnFailure (self ):
257+ "test context manager - rollback on failure"
258+ connection = cx_Oracle .connect (self .username , self .password ,
259+ self .tnsentry )
260+ cursor = connection .cursor ()
261+ cursor .execute ("truncate table TestTempTable" )
262+ cursor .execute ("""
263+ insert into TestTempTable (IntCol, StringCol)
264+ values (1, null)""" )
265+ try :
266+ with connection :
267+ 1 / 0
268+ except :
269+ pass
270+ cursor .execute ("select count(*) from TestTempTable" )
271+ count , = cursor .fetchone ()
272+ self .assertEqual (count , 0 )
273+
274+ def testConnectionAttributes (self ):
275+ "test connection attribute values"
276+ connection = cx_Oracle .connect (self .username , self .password ,
277+ self .tnsentry , encoding = "ASCII" )
278+ self .assertEqual (connection .maxBytesPerCharacter , 1 )
279+ connection = cx_Oracle .connect (self .username , self .password ,
280+ self .tnsentry , encoding = "UTF-8" )
281+ self .assertEqual (connection .maxBytesPerCharacter , 4 )
282+ self .assertEqual (connection .ltxid , b'' )
283+ self .assertEqual (connection .current_schema , None )
284+ connection .current_schema = "test_schema"
285+ self .assertEqual (connection .current_schema , "test_schema" )
286+ self .assertEqual (connection .edition , None )
287+ connection .external_name = "test_external"
288+ self .assertEqual (connection .external_name , "test_external" )
289+ connection .internal_name = "test_internal"
290+ self .assertEqual (connection .internal_name , "test_internal" )
291+ connection .stmtcachesize = 30
292+ self .assertEqual (connection .stmtcachesize , 30 )
293+ self .assertRaises (TypeError , connection .stmtcachesize , 20.5 )
294+ self .assertRaises (TypeError , connection .stmtcachesize , "value" )
295+
296+ def testPing (self ):
297+ "test connection ping"
298+ connection = cx_Oracle .connect (self .username , self .password ,
299+ self .tnsentry )
300+ connection .ping ()
301+
302+ def testTransactionBegin (self ):
303+ "test begin, prepare, cancel transaction"
304+ connection = cx_Oracle .connect (self .username , self .password ,
305+ self .tnsentry )
306+ cursor = connection .cursor ()
307+ cursor .execute ("truncate table TestTempTable" )
308+ connection .begin (10 , 'trxnId' , 'branchId' )
309+ self .assertEqual (connection .prepare (), False )
310+ connection .begin (10 , 'trxnId' , 'branchId' )
311+ cursor .execute ("""
312+ insert into TestTempTable (IntCol, StringCol)
313+ values (1, 'tesName')""" )
314+ self .assertEqual (connection .prepare (), True )
315+ connection .cancel ()
316+ connection .rollback ()
317+ cursor .execute ("select count(*) from TestTempTable" )
318+ count , = cursor .fetchone ()
319+ self .assertEqual (count , 0 )
320+
0 commit comments