HackBase/python-cx_Oracle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added test cases for AQ visibility and delivery modes.

Browse Source
This commit is contained in:
Anthony Tuininga 2018-01-30 15:25:56 -07:00
parent 00330794be
commit 65580226cd
2 changed files with 151 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

144
test/AQ.py
View File

@ -19,7 +19,7 @@ class TestAQ(BaseTestCase):
]
def __deqInThread(self, results):
connection = cx_Oracle.connect(USERNAME, PASSWORD, TNSENTRY)
connection = self.getConnection()
booksType = connection.gettype("UDT_BOOK")
book = booksType.newobject()
options = connection.deqoptions()
@ -98,6 +98,7 @@ class TestAQ(BaseTestCase):
"TEST_TRANSFORMATION")
self.__verifyAttribute(options, "visibility", cx_Oracle.ENQ_IMMEDIATE)
self.__verifyAttribute(options, "wait", 1287)
self.__verifyAttribute(options, "msgid", b'mID')
def testDeqWithWait(self):
"test waiting for dequeue"
@ -146,3 +147,144 @@ class TestAQ(BaseTestCase):
self.__verifyAttribute(props, "priority", 1)
self.assertEqual(props.state, cx_Oracle.MSG_READY)
def testVisibilityModeCommit(self):
"test enqueue visibility option - ENQ_ON_COMMIT"
booksType = self.connection.gettype("UDT_BOOK")
book = booksType.newobject()
book.TITLE, book.AUTHORS, book.PRICE = self.bookData[0]
enqOptions = self.connection.enqoptions()
enqOptions.visibility = cx_Oracle.ENQ_ON_COMMIT
props = self.connection.msgproperties()
self.connection.enq("BOOKS", enqOptions, props, book)
otherConnection = self.getConnection()
deqOptions = otherConnection.deqoptions()
deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
booksType = otherConnection.gettype("UDT_BOOK")
book = booksType.newobject()
props = otherConnection.msgproperties()
messageId = otherConnection.deq("BOOKS", deqOptions, props, book)
self.assertTrue(messageId is None)
def testVisibilityModeImmediate(self):
"test enqueue visibility option - ENQ_IMMEDIATE"
booksType = self.connection.gettype("UDT_BOOK")
book = booksType.newobject()
book.TITLE, book.AUTHORS, book.PRICE = self.bookData[0]
enqOptions = self.connection.enqoptions()
enqOptions.visibility = cx_Oracle.ENQ_IMMEDIATE
props = self.connection.msgproperties()
self.connection.enq("BOOKS", enqOptions, props, book)
otherConnection = self.getConnection()
deqOptions = otherConnection.deqoptions()
deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
deqOptions.visibility = cx_Oracle.DEQ_ON_COMMIT
deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
booksType = otherConnection.gettype("UDT_BOOK")
book = booksType.newobject()
props = otherConnection.msgproperties()
otherConnection.deq("BOOKS", deqOptions, props, book)
results = (book.TITLE, book.AUTHORS, book.PRICE)
otherConnection.commit()
self.assertEqual(results, self.bookData[0])
def testDeliveryModeSameBuffered(self):
"test enqueue/dequeue delivery modes identical - buffered"
booksType = self.connection.gettype("UDT_BOOK")
book = booksType.newobject()
book.TITLE, book.AUTHORS, book.PRICE = self.bookData[0]
enqOptions = self.connection.enqoptions()
enqOptions.deliverymode = cx_Oracle.MSG_BUFFERED
enqOptions.visibility = cx_Oracle.ENQ_IMMEDIATE
props = self.connection.msgproperties()
self.connection.enq("BOOKS", enqOptions, props, book)
otherConnection = self.getConnection()
deqOptions = otherConnection.deqoptions()
deqOptions.deliverymode = cx_Oracle.MSG_BUFFERED
deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
deqOptions.visibility = cx_Oracle.DEQ_IMMEDIATE
deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
booksType = otherConnection.gettype("UDT_BOOK")
book = booksType.newobject()
props = otherConnection.msgproperties()
otherConnection.deq("BOOKS", deqOptions, props, book)
results = (book.TITLE, book.AUTHORS, book.PRICE)
otherConnection.commit()
self.assertEqual(results, self.bookData[0])
def testDeliveryModeSamePersistent(self):
"test enqueue/dequeue delivery modes identical - persistent"
booksType = self.connection.gettype("UDT_BOOK")
book = booksType.newobject()
book.TITLE, book.AUTHORS, book.PRICE = self.bookData[0]
enqOptions = self.connection.enqoptions()
enqOptions.deliverymode = cx_Oracle.MSG_PERSISTENT
enqOptions.visibility = cx_Oracle.ENQ_IMMEDIATE
props = self.connection.msgproperties()
self.connection.enq("BOOKS", enqOptions, props, book)
otherConnection = self.getConnection()
deqOptions = otherConnection.deqoptions()
deqOptions.deliverymode = cx_Oracle.MSG_PERSISTENT
deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
deqOptions.visibility = cx_Oracle.DEQ_IMMEDIATE
deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
booksType = otherConnection.gettype("UDT_BOOK")
book = booksType.newobject()
props = otherConnection.msgproperties()
otherConnection.deq("BOOKS", deqOptions, props, book)
results = (book.TITLE, book.AUTHORS, book.PRICE)
otherConnection.commit()
self.assertEqual(results, self.bookData[0])
def testDeliveryModeSamePersistentBuffered(self):
"test enqueue/dequeue delivery modes identical - persistent/buffered"
booksType = self.connection.gettype("UDT_BOOK")
book = booksType.newobject()
book.TITLE, book.AUTHORS, book.PRICE = self.bookData[0]
enqOptions = self.connection.enqoptions()
enqOptions.deliverymode = cx_Oracle.MSG_PERSISTENT_OR_BUFFERED
enqOptions.visibility = cx_Oracle.ENQ_IMMEDIATE
props = self.connection.msgproperties()
self.connection.enq("BOOKS", enqOptions, props, book)
otherConnection = self.getConnection()
deqOptions = otherConnection.deqoptions()
deqOptions.deliverymode = cx_Oracle.MSG_PERSISTENT_OR_BUFFERED
deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
deqOptions.visibility = cx_Oracle.DEQ_IMMEDIATE
deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
booksType = otherConnection.gettype("UDT_BOOK")
book = booksType.newobject()
props = otherConnection.msgproperties()
otherConnection.deq("BOOKS", deqOptions, props, book)
results = (book.TITLE, book.AUTHORS, book.PRICE)
otherConnection.commit()
self.assertEqual(results, self.bookData[0])
def testDeliveryModeDifferent(self):
"test enqueue/dequeue delivery modes different"
booksType = self.connection.gettype("UDT_BOOK")
book = booksType.newobject()
book.TITLE, book.AUTHORS, book.PRICE = self.bookData[0]
enqOptions = self.connection.enqoptions()
enqOptions.deliverymode = cx_Oracle.MSG_BUFFERED
enqOptions.visibility = cx_Oracle.ENQ_IMMEDIATE
props = self.connection.msgproperties()
self.connection.enq("BOOKS", enqOptions, props, book)
otherConnection = self.getConnection()
deqOptions = otherConnection.deqoptions()
deqOptions.deliverymode = cx_Oracle.MSG_PERSISTENT
deqOptions.navigation = cx_Oracle.DEQ_FIRST_MSG
deqOptions.visibility = cx_Oracle.DEQ_IMMEDIATE
deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
booksType = otherConnection.gettype("UDT_BOOK")
book = booksType.newobject()
props = otherConnection.msgproperties()
messageId = otherConnection.deq("BOOKS", deqOptions, props, book)
self.assertTrue(messageId is None)

12
test/test.py
View File

@ -62,12 +62,16 @@ else:
class BaseTestCase(unittest.TestCase):
def setUp(self):
def getConnection(self, **kwargs):
import cx_Oracle
import TestEnv
self.connection = cx_Oracle.Connection(TestEnv.MAIN_USER,
TestEnv.MAIN_PASSWORD, TestEnv.CONNECT_STRING,
encoding = TestEnv.ENCODING, nencoding = TestEnv.NENCODING)
return cx_Oracle.Connection(TestEnv.MAIN_USER, TestEnv.MAIN_PASSWORD,
TestEnv.CONNECT_STRING, encoding = TestEnv.ENCODING,
nencoding = TestEnv.NENCODING, **kwargs)
def setUp(self):
import TestEnv
self.connection = self.getConnection()
self.cursor = self.connection.cursor()
self.cursor.arraysize = TestEnv.ARRAY_SIZE