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

Renamed table TestExecuteMany to TestTempTable to be more in keeping with its

Browse Source 
actual uses (not just testing executemany()).
This commit is contained in:
Anthony Tuininga 2017-01-12 15:20:44 -07:00
parent 59053718be
commit 0e50be2a08
7 changed files with 61 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
test/Connection.py
View File

@ -76,13 +76,13 @@ class TestConnection(TestCase):
connection = cx_Oracle.connect(self.username, self.password, connection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute("truncate table TestExecuteMany") cursor.execute("truncate table TestTempTable")
otherConnection = cx_Oracle.connect(self.username, self.password, otherConnection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
otherCursor = otherConnection.cursor() otherCursor = otherConnection.cursor()
otherCursor.execute("insert into TestExecuteMany (IntCol) values (1)") otherCursor.execute("insert into TestTempTable (IntCol) values (1)")
otherConnection.close() otherConnection.close()
cursor.execute("select count(*) from TestExecuteMany") cursor.execute("select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)
@ -91,14 +91,14 @@ class TestConnection(TestCase):
connection = cx_Oracle.connect(self.username, self.password, connection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute("truncate table TestExecuteMany") cursor.execute("truncate table TestTempTable")
otherConnection = cx_Oracle.connect(self.username, self.password, otherConnection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
otherCursor = otherConnection.cursor() otherCursor = otherConnection.cursor()
otherCursor.execute("insert into TestExecuteMany (IntCol) values (1)") otherCursor.execute("insert into TestTempTable (IntCol) values (1)")
del otherCursor del otherCursor
del otherConnection del otherConnection
cursor.execute("select count(*) from TestExecuteMany") cursor.execute("select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)

38
test/Cursor.py
View File

@ -88,57 +88,57 @@ class TestCursor(BaseTestCase):
def testExecuteManyByName(self): def testExecuteManyByName(self):
"""test executing a statement multiple times (named args)""" """test executing a statement multiple times (named args)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
rows = [ { "value" : n } for n in range(250) ] rows = [ { "value" : n } for n in range(250) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = "insert into TestExecuteMany (IntCol) values (:value)" statement = "insert into TestTempTable (IntCol) values (:value)"
self.cursor.executemany(statement, rows) self.cursor.executemany(statement, rows)
self.connection.commit() self.connection.commit()
self.cursor.execute("select count(*) from TestExecuteMany") self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyByPosition(self): def testExecuteManyByPosition(self):
"""test executing a statement multiple times (positional args)""" """test executing a statement multiple times (positional args)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
rows = [ [n] for n in range(230) ] rows = [ [n] for n in range(230) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = "insert into TestExecuteMany (IntCol) values (:1)" statement = "insert into TestTempTable (IntCol) values (:1)"
self.cursor.executemany(statement, rows) self.cursor.executemany(statement, rows)
self.connection.commit() self.connection.commit()
self.cursor.execute("select count(*) from TestExecuteMany") self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyWithPrepare(self): def testExecuteManyWithPrepare(self):
"""test executing a statement multiple times (with prepare)""" """test executing a statement multiple times (with prepare)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
rows = [ [n] for n in range(225) ] rows = [ [n] for n in range(225) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = "insert into TestExecuteMany (IntCol) values (:1)" statement = "insert into TestTempTable (IntCol) values (:1)"
self.cursor.prepare(statement) self.cursor.prepare(statement)
self.cursor.executemany(None, rows) self.cursor.executemany(None, rows)
self.connection.commit() self.connection.commit()
self.cursor.execute("select count(*) from TestExecuteMany") self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyWithRebind(self): def testExecuteManyWithRebind(self):
"""test executing a statement multiple times (with rebind)""" """test executing a statement multiple times (with rebind)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
rows = [ [n] for n in range(235) ] rows = [ [n] for n in range(235) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = "insert into TestExecuteMany (IntCol) values (:1)" statement = "insert into TestTempTable (IntCol) values (:1)"
self.cursor.executemany(statement, rows[:50]) self.cursor.executemany(statement, rows[:50])
self.cursor.executemany(statement, rows[50:]) self.cursor.executemany(statement, rows[50:])
self.connection.commit() self.connection.commit()
self.cursor.execute("select count(*) from TestExecuteMany") self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyWithResize(self): def testExecuteManyWithResize(self):
"""test executing a statement multiple times (with resize)""" """test executing a statement multiple times (with resize)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
rows = [ ( 1, "First" ), rows = [ ( 1, "First" ),
( 2, "Second" ), ( 2, "Second" ),
( 3, "Third" ), ( 3, "Third" ),
@ -148,10 +148,10 @@ class TestCursor(BaseTestCase):
( 7, "Seventh" ) ] ( 7, "Seventh" ) ]
self.cursor.bindarraysize = 5 self.cursor.bindarraysize = 5
self.cursor.setinputsizes(int, 100) self.cursor.setinputsizes(int, 100)
sql = "insert into TestExecuteMany (IntCol, StringCol) values (:1, :2)" sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
self.cursor.executemany(sql, rows) self.cursor.executemany(sql, rows)
var = self.cursor.bindvars[1] var = self.cursor.bindvars[1]
self.cursor.execute("select count(*) from TestExecuteMany") self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
self.assertEqual(var.bufferSize, self.assertEqual(var.bufferSize,
@ -159,9 +159,9 @@ class TestCursor(BaseTestCase):
def testExecuteManyWithExecption(self): def testExecuteManyWithExecption(self):
"""test executing a statement multiple times (with exception)""" """test executing a statement multiple times (with exception)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
rows = [ { "value" : n } for n in (1, 2, 3, 2, 5) ] rows = [ { "value" : n } for n in (1, 2, 3, 2, 5) ]
statement = "insert into TestExecuteMany (IntCol) values (:value)" statement = "insert into TestTempTable (IntCol) values (:value)"
self.assertRaises(cx_Oracle.DatabaseError, self.cursor.executemany, self.assertRaises(cx_Oracle.DatabaseError, self.cursor.executemany,
statement, rows) statement, rows)
self.assertEqual(self.cursor.rowcount, 3) self.assertEqual(self.cursor.rowcount, 3)
@ -201,7 +201,7 @@ class TestCursor(BaseTestCase):
def testIteratorsInterrupted(self): def testIteratorsInterrupted(self):
"""test iterators (with intermediate execute)""" """test iterators (with intermediate execute)"""
self.cursor.execute("truncate table TestExecuteMany") self.cursor.execute("truncate table TestTempTable")
self.cursor.execute(""" self.cursor.execute("""
select IntCol select IntCol
from TestNumbers from TestNumbers
@ -212,7 +212,7 @@ class TestCursor(BaseTestCase):
value, = next(testIter) value, = next(testIter)
else: else:
value, = testIter.next() value, = testIter.next()
self.cursor.execute("insert into TestExecuteMany (IntCol) values (1)") self.cursor.execute("insert into TestTempTable (IntCol) values (1)")
if sys.version_info[0] >= 3: if sys.version_info[0] >= 3:
self.assertRaises(cx_Oracle.InterfaceError, next, testIter) self.assertRaises(cx_Oracle.InterfaceError, next, testIter)
else: else:

12
test/SessionPool.py
View File

@ -66,12 +66,12 @@ class TestConnection(TestCase):
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute("truncate table TestExecuteMany") cursor.execute("truncate table TestTempTable")
cursor.execute("insert into TestExecuteMany (IntCol) values (1)") cursor.execute("insert into TestTempTable (IntCol) values (1)")
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute("select count(*) from TestExecuteMany") cursor.execute("select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)
@ -80,13 +80,13 @@ class TestConnection(TestCase):
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute("truncate table TestExecuteMany") cursor.execute("truncate table TestTempTable")
cursor.execute("insert into TestExecuteMany (IntCol) values (1)") cursor.execute("insert into TestTempTable (IntCol) values (1)")
pool.release(connection) pool.release(connection)
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute("select count(*) from TestExecuteMany") cursor.execute("select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)

4
test/SetupTest.sql
View File

@ -116,10 +116,10 @@ create table cx_Oracle.TestLongRaws (
LongRawCol long raw not null LongRawCol long raw not null
) tablespace users; ) tablespace users;
create table cx_Oracle.TestExecuteMany ( create table cx_Oracle.TestTempTable (
IntCol number(9) not null, IntCol number(9) not null,
StringCol varchar2(100), StringCol varchar2(100),
constraint TestExecuteMany_pk primary key (IntCol) constraint TestTempTable_pk primary key (IntCol)
using index tablespace users using index tablespace users
) tablespace users; ) tablespace users;

12
test/uConnection.py
View File

@ -76,13 +76,13 @@ class TestConnection(TestCase):
connection = cx_Oracle.connect(self.username, self.password, connection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(u"truncate table TestExecuteMany") cursor.execute(u"truncate table TestTempTable")
otherConnection = cx_Oracle.connect(self.username, self.password, otherConnection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
otherCursor = otherConnection.cursor() otherCursor = otherConnection.cursor()
otherCursor.execute(u"insert into TestExecuteMany (IntCol) values (1)") otherCursor.execute(u"insert into TestTempTable (IntCol) values (1)")
otherConnection.close() otherConnection.close()
cursor.execute(u"select count(*) from TestExecuteMany") cursor.execute(u"select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)
@ -91,14 +91,14 @@ class TestConnection(TestCase):
connection = cx_Oracle.connect(self.username, self.password, connection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(u"truncate table TestExecuteMany") cursor.execute(u"truncate table TestTempTable")
otherConnection = cx_Oracle.connect(self.username, self.password, otherConnection = cx_Oracle.connect(self.username, self.password,
self.tnsentry) self.tnsentry)
otherCursor = otherConnection.cursor() otherCursor = otherConnection.cursor()
otherCursor.execute(u"insert into TestExecuteMany (IntCol) values (1)") otherCursor.execute(u"insert into TestTempTable (IntCol) values (1)")
del otherCursor del otherCursor
del otherConnection del otherConnection
cursor.execute(u"select count(*) from TestExecuteMany") cursor.execute(u"select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)

32
test/uCursor.py
View File

@ -76,59 +76,59 @@ class TestCursor(BaseTestCase):
def testExecuteManyByName(self): def testExecuteManyByName(self):
"""test executing a statement multiple times (named args)""" """test executing a statement multiple times (named args)"""
self.cursor.execute(u"truncate table TestExecuteMany") self.cursor.execute(u"truncate table TestTempTable")
rows = [ { u"value" : n } for n in range(250) ] rows = [ { u"value" : n } for n in range(250) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = u"insert into TestExecuteMany (IntCol) values (:value)" statement = u"insert into TestTempTable (IntCol) values (:value)"
self.cursor.executemany(statement, rows) self.cursor.executemany(statement, rows)
self.connection.commit() self.connection.commit()
self.cursor.execute(u"select count(*) from TestExecuteMany") self.cursor.execute(u"select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyByPosition(self): def testExecuteManyByPosition(self):
"""test executing a statement multiple times (positional args)""" """test executing a statement multiple times (positional args)"""
self.cursor.execute(u"truncate table TestExecuteMany") self.cursor.execute(u"truncate table TestTempTable")
rows = [ [n] for n in range(230) ] rows = [ [n] for n in range(230) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = u"insert into TestExecuteMany (IntCol) values (:1)" statement = u"insert into TestTempTable (IntCol) values (:1)"
self.cursor.executemany(statement, rows) self.cursor.executemany(statement, rows)
self.connection.commit() self.connection.commit()
self.cursor.execute(u"select count(*) from TestExecuteMany") self.cursor.execute(u"select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyWithPrepare(self): def testExecuteManyWithPrepare(self):
"""test executing a statement multiple times (with prepare)""" """test executing a statement multiple times (with prepare)"""
self.cursor.execute(u"truncate table TestExecuteMany") self.cursor.execute(u"truncate table TestTempTable")
rows = [ [n] for n in range(225) ] rows = [ [n] for n in range(225) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = u"insert into TestExecuteMany (IntCol) values (:1)" statement = u"insert into TestTempTable (IntCol) values (:1)"
self.cursor.prepare(statement) self.cursor.prepare(statement)
self.cursor.executemany(None, rows) self.cursor.executemany(None, rows)
self.connection.commit() self.connection.commit()
self.cursor.execute(u"select count(*) from TestExecuteMany") self.cursor.execute(u"select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyWithRebind(self): def testExecuteManyWithRebind(self):
"""test executing a statement multiple times (with rebind)""" """test executing a statement multiple times (with rebind)"""
self.cursor.execute(u"truncate table TestExecuteMany") self.cursor.execute(u"truncate table TestTempTable")
rows = [ [n] for n in range(235) ] rows = [ [n] for n in range(235) ]
self.cursor.arraysize = 100 self.cursor.arraysize = 100
statement = u"insert into TestExecuteMany (IntCol) values (:1)" statement = u"insert into TestTempTable (IntCol) values (:1)"
self.cursor.executemany(statement, rows[:50]) self.cursor.executemany(statement, rows[:50])
self.cursor.executemany(statement, rows[50:]) self.cursor.executemany(statement, rows[50:])
self.connection.commit() self.connection.commit()
self.cursor.execute(u"select count(*) from TestExecuteMany") self.cursor.execute(u"select count(*) from TestTempTable")
count, = self.cursor.fetchone() count, = self.cursor.fetchone()
self.assertEqual(count, len(rows)) self.assertEqual(count, len(rows))
def testExecuteManyWithExecption(self): def testExecuteManyWithExecption(self):
"""test executing a statement multiple times (with exception)""" """test executing a statement multiple times (with exception)"""
self.cursor.execute(u"truncate table TestExecuteMany") self.cursor.execute(u"truncate table TestTempTable")
rows = [ { u"value" : n } for n in (1, 2, 3, 2, 5) ] rows = [ { u"value" : n } for n in (1, 2, 3, 2, 5) ]
statement = u"insert into TestExecuteMany (IntCol) values (:value)" statement = u"insert into TestTempTable (IntCol) values (:value)"
self.assertRaises(cx_Oracle.DatabaseError, self.cursor.executemany, self.assertRaises(cx_Oracle.DatabaseError, self.cursor.executemany,
statement, rows) statement, rows)
self.assertEqual(self.cursor.rowcount, 3) self.assertEqual(self.cursor.rowcount, 3)
@ -168,7 +168,7 @@ class TestCursor(BaseTestCase):
def testIteratorsInterrupted(self): def testIteratorsInterrupted(self):
"""test iterators (with intermediate execute)""" """test iterators (with intermediate execute)"""
self.cursor.execute(u"truncate table TestExecuteMany") self.cursor.execute(u"truncate table TestTempTable")
self.cursor.execute(u""" self.cursor.execute(u"""
select IntCol select IntCol
from TestNumbers from TestNumbers
@ -176,7 +176,7 @@ class TestCursor(BaseTestCase):
order by IntCol""") order by IntCol""")
testIter = iter(self.cursor) testIter = iter(self.cursor)
value, = testIter.next() value, = testIter.next()
self.cursor.execute(u"insert into TestExecuteMany (IntCol) values (1)") self.cursor.execute(u"insert into TestTempTable (IntCol) values (1)")
self.assertRaises(cx_Oracle.InterfaceError, testIter.next) self.assertRaises(cx_Oracle.InterfaceError, testIter.next)
def testBindNames(self): def testBindNames(self):

12
test/uSessionPool.py
View File

@ -59,12 +59,12 @@ class TestConnection(TestCase):
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(u"truncate table TestExecuteMany") cursor.execute(u"truncate table TestTempTable")
cursor.execute(u"insert into TestExecuteMany (IntCol) values (1)") cursor.execute(u"insert into TestTempTable (IntCol) values (1)")
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(u"select count(*) from TestExecuteMany") cursor.execute(u"select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)
@ -73,13 +73,13 @@ class TestConnection(TestCase):
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(u"truncate table TestExecuteMany") cursor.execute(u"truncate table TestTempTable")
cursor.execute(u"insert into TestExecuteMany (IntCol) values (1)") cursor.execute(u"insert into TestTempTable (IntCol) values (1)")
pool.release(connection) pool.release(connection)
pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3) pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, 1, 8, 3)
connection = pool.acquire() connection = pool.acquire()
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(u"select count(*) from TestExecuteMany") cursor.execute(u"select count(*) from TestTempTable")
count, = cursor.fetchone() count, = cursor.fetchone()
self.assertEqual(count, 0) self.assertEqual(count, 0)