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

Added additional test for errors taking place in connections acquired from a

Browse Source 
session pool.
This commit is contained in:
Anthony Tuininga 2008-11-11 21:21:31 +00:00
parent 5b6a9b017a
commit 39333aa7d4
1 changed files with 19 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
test/SessionPool.py
View File

@ -12,6 +12,13 @@ class TestConnection(TestCase):
count, = cursor.fetchone() count, = cursor.fetchone()
self.failUnlessEqual(count, 10) self.failUnlessEqual(count, 10)
def __ConnectAndGenerateError(self):
"""Connect to the database, perform a query which raises an error"""
connection = self.pool.acquire()
cursor = connection.cursor()
self.failUnlessRaises(cx_Oracle.DatabaseError, cursor.execute,
"select 1 / 0 from dual")
def testPool(self): def testPool(self):
"""test that the pool is created and has the right attributes""" """test that the pool is created and has the right attributes"""
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 2, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 2, 8, 3)
@ -96,3 +103,15 @@ class TestConnection(TestCase):
for thread in threads: for thread in threads:
thread.join() thread.join()
def testThreadingWithErrors(self):
"""test session pool to database with multiple threads (with errors)"""
self.pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 5, 20,
2, threaded = True)
threads = []
for i in range(20):
thread = threading.Thread(None, self.__ConnectAndGenerateError)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()