Add additional test cases.
This commit is contained in:
parent
46e3aff3fa
commit
07b3b93da1
@ -145,7 +145,9 @@ class TestAQ(BaseTestCase):
|
||||
self.__verifyAttribute(props, "expiration", 30)
|
||||
self.assertEqual(props.attempts, 0)
|
||||
self.__verifyAttribute(props, "priority", 1)
|
||||
self.__verifyAttribute(props, "msgid", b'mID')
|
||||
self.assertEqual(props.state, cx_Oracle.MSG_READY)
|
||||
self.assertEqual(props.deliverymode, 0)
|
||||
|
||||
def testVisibilityModeCommit(self):
|
||||
"test enqueue visibility option - ENQ_ON_COMMIT"
|
||||
|
||||
@ -238,3 +238,83 @@ class TestConnection(TestCase):
|
||||
(self.username, self.tnsentry)
|
||||
self.assertEqual(str(connection), expectedValue)
|
||||
|
||||
def testCtxMgrCommitOnSuccess(self):
|
||||
"test context manager - commit on success"
|
||||
connection = cx_Oracle.connect(self.username, self.password,
|
||||
self.tnsentry)
|
||||
cursor = connection.cursor()
|
||||
cursor.execute("truncate table TestTempTable")
|
||||
with connection:
|
||||
cursor.execute("""
|
||||
insert into TestTempTable (IntCol, StringCol)
|
||||
values (1, null)""")
|
||||
connection.rollback()
|
||||
cursor.execute("select count(*) from TestTempTable")
|
||||
count, = cursor.fetchone()
|
||||
self.assertEqual(count, 1)
|
||||
|
||||
def testCtxMgrRollbackOnFailure(self):
|
||||
"test context manager - rollback on failure"
|
||||
connection = cx_Oracle.connect(self.username, self.password,
|
||||
self.tnsentry)
|
||||
cursor = connection.cursor()
|
||||
cursor.execute("truncate table TestTempTable")
|
||||
cursor.execute("""
|
||||
insert into TestTempTable (IntCol, StringCol)
|
||||
values (1, null)""")
|
||||
try:
|
||||
with connection:
|
||||
1 / 0
|
||||
except:
|
||||
pass
|
||||
cursor.execute("select count(*) from TestTempTable")
|
||||
count, = cursor.fetchone()
|
||||
self.assertEqual(count, 0)
|
||||
|
||||
def testConnectionAttributes(self):
|
||||
"test connection attribute values"
|
||||
connection = cx_Oracle.connect(self.username, self.password,
|
||||
self.tnsentry, encoding = "ASCII")
|
||||
self.assertEqual(connection.maxBytesPerCharacter, 1)
|
||||
connection = cx_Oracle.connect(self.username, self.password,
|
||||
self.tnsentry, encoding = "UTF-8")
|
||||
self.assertEqual(connection.maxBytesPerCharacter, 4)
|
||||
self.assertEqual(connection.ltxid, b'')
|
||||
self.assertEqual(connection.current_schema, None)
|
||||
connection.current_schema = "test_schema"
|
||||
self.assertEqual(connection.current_schema, "test_schema")
|
||||
self.assertEqual(connection.edition, None)
|
||||
connection.external_name = "test_external"
|
||||
self.assertEqual(connection.external_name, "test_external")
|
||||
connection.internal_name = "test_internal"
|
||||
self.assertEqual(connection.internal_name, "test_internal")
|
||||
connection.stmtcachesize = 30
|
||||
self.assertEqual(connection.stmtcachesize, 30)
|
||||
self.assertRaises(TypeError, connection.stmtcachesize, 20.5)
|
||||
self.assertRaises(TypeError, connection.stmtcachesize, "value")
|
||||
|
||||
def testPing(self):
|
||||
"test connection ping"
|
||||
connection = cx_Oracle.connect(self.username, self.password,
|
||||
self.tnsentry)
|
||||
connection.ping()
|
||||
|
||||
def testTransactionBegin(self):
|
||||
"test begin, prepare, cancel transaction"
|
||||
connection = cx_Oracle.connect(self.username, self.password,
|
||||
self.tnsentry)
|
||||
cursor = connection.cursor()
|
||||
cursor.execute("truncate table TestTempTable")
|
||||
connection.begin(10, 'trxnId', 'branchId')
|
||||
self.assertEqual(connection.prepare(), False)
|
||||
connection.begin(10, 'trxnId', 'branchId')
|
||||
cursor.execute("""
|
||||
insert into TestTempTable (IntCol, StringCol)
|
||||
values (1, 'tesName')""")
|
||||
self.assertEqual(connection.prepare(), True)
|
||||
connection.cancel()
|
||||
connection.rollback()
|
||||
cursor.execute("select count(*) from TestTempTable")
|
||||
count, = cursor.fetchone()
|
||||
self.assertEqual(count, 0)
|
||||
|
||||
|
||||
32
test/Module.py
Normal file
32
test/Module.py
Normal file
@ -0,0 +1,32 @@
|
||||
#------------------------------------------------------------------------------
|
||||
# Copyright 2018, Oracle and/or its affiliates. All rights reserved.
|
||||
#------------------------------------------------------------------------------
|
||||
|
||||
"""Module for testing module methods."""
|
||||
|
||||
import datetime
|
||||
import time
|
||||
|
||||
class TestModule(BaseTestCase):
|
||||
|
||||
def testDateFromTicks(self):
|
||||
"test DateFromTicks()"
|
||||
today = datetime.datetime.today()
|
||||
timestamp = time.mktime(today.timetuple())
|
||||
date = cx_Oracle.DateFromTicks(timestamp)
|
||||
self.assertEqual(date, today.date())
|
||||
|
||||
def testTimestampFromTicks(self):
|
||||
"test TimestampFromTicks()"
|
||||
timestamp = time.mktime(datetime.datetime.today().timetuple())
|
||||
today = datetime.datetime.fromtimestamp(timestamp)
|
||||
date = cx_Oracle.TimestampFromTicks(timestamp)
|
||||
self.assertEqual(date, today)
|
||||
|
||||
def testUnsupportedFunctions(self):
|
||||
"test unsupported time functions"
|
||||
self.assertRaises(cx_Oracle.NotSupportedError, cx_Oracle.Time,
|
||||
12, 0, 0)
|
||||
self.assertRaises(cx_Oracle.NotSupportedError, cx_Oracle.TimeFromTicks,
|
||||
100)
|
||||
|
||||
@ -298,4 +298,6 @@ class TestObjectVar(BaseTestCase):
|
||||
objType = self.connection.gettype("UDT_OBJECT")
|
||||
self.assertEqual(str(objType),
|
||||
"<cx_Oracle.ObjectType CX_ORACLE.UDT_OBJECT>")
|
||||
self.assertEqual(str(objType.attributes[0]),
|
||||
"<cx_Oracle.ObjectAttribute NUMBERVALUE>")
|
||||
|
||||
|
||||
@ -36,6 +36,7 @@ if len(sys.argv) > 1 and not inSetup:
|
||||
moduleNames = [os.path.splitext(v)[0] for v in sys.argv[1:]]
|
||||
else:
|
||||
moduleNames = [
|
||||
"Module",
|
||||
"Connection",
|
||||
"Cursor",
|
||||
"CursorVar",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user