#------------------------------------------------------------------------------ # Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved. #------------------------------------------------------------------------------ """Module for testing DML returning clauses.""" import TestEnv import cx_Oracle class TestCase(TestEnv.BaseTestCase): def testInsert(self): "test insert statement (single row) with DML returning" self.cursor.execute("truncate table TestTempTable") intVal = 5 strVal = "A test string" intVar = self.cursor.var(cx_Oracle.NUMBER) strVar = self.cursor.var(str) self.cursor.execute(""" insert into TestTempTable (IntCol, StringCol) values (:intVal, :strVal) returning IntCol, StringCol into :intVar, :strVar""", intVal = intVal, strVal = strVal, intVar = intVar, strVar = strVar) self.assertEqual(intVar.values, [[intVal]]) self.assertEqual(strVar.values, [[strVal]]) def testInsertMany(self): "test insert statement (multiple rows) with DML returning" self.cursor.execute("truncate table TestTempTable") intValues = [5, 8, 17, 24, 6] strValues = ["Test 5", "Test 8", "Test 17", "Test 24", "Test 6"] intVar = self.cursor.var(cx_Oracle.NUMBER, arraysize = len(intValues)) strVar = self.cursor.var(str, arraysize = len(intValues)) self.cursor.setinputsizes(None, None, intVar, strVar) data = list(zip(intValues, strValues)) self.cursor.executemany(""" insert into TestTempTable (IntCol, StringCol) values (:intVal, :strVal) returning IntCol, StringCol into :intVar, :strVar""", data) self.assertEqual(intVar.values, [[v] for v in intValues]) self.assertEqual(strVar.values, [[v] for v in strValues]) def testInsertWithSmallSize(self): "test insert statement with DML returning into too small a variable" self.cursor.execute("truncate table TestTempTable") intVal = 6 strVal = "A different test string" intVar = self.cursor.var(cx_Oracle.NUMBER) strVar = self.cursor.var(str, 2) parameters = dict(intVal = intVal, strVal = strVal, intVar = intVar, strVar = strVar) self.assertRaises(cx_Oracle.DatabaseError, self.cursor.execute, """ insert into TestTempTable (IntCol, StringCol) values (:intVal, :strVal) returning IntCol, StringCol into :intVar, :strVar""", parameters) def testUpdateSingleRow(self): "test update single row statement with DML returning" intVal = 7 strVal = "The updated value of the string" self.cursor.execute("truncate table TestTempTable") self.cursor.execute(""" insert into TestTempTable (IntCol, StringCol) values (:1, :2)""", (intVal, "The initial value of the string")) intVar = self.cursor.var(cx_Oracle.NUMBER) strVar = self.cursor.var(str) self.cursor.execute(""" update TestTempTable set StringCol = :strVal where IntCol = :intVal returning IntCol, StringCol into :intVar, :strVar""", intVal = intVal, strVal = strVal, intVar = intVar, strVar = strVar) self.assertEqual(intVar.values, [[intVal]]) self.assertEqual(strVar.values, [[strVal]]) def testUpdateNoRows(self): "test update no rows statement with DML returning" intVal = 8 strVal = "The updated value of the string" self.cursor.execute("truncate table TestTempTable") self.cursor.execute(""" insert into TestTempTable (IntCol, StringCol) values (:1, :2)""", (intVal, "The initial value of the string")) intVar = self.cursor.var(cx_Oracle.NUMBER) strVar = self.cursor.var(str) self.cursor.execute(""" update TestTempTable set StringCol = :strVal where IntCol = :intVal returning IntCol, StringCol into :intVar, :strVar""", intVal = intVal + 1, strVal = strVal, intVar = intVar, strVar = strVar) self.assertEqual(intVar.values, [[]]) self.assertEqual(strVar.values, [[]]) self.assertEqual(intVar.getvalue(), []) self.assertEqual(strVar.getvalue(), []) def testUpdateMultipleRows(self): "test update multiple rows statement with DML returning" self.cursor.execute("truncate table TestTempTable") for i in (8, 9, 10): self.cursor.execute(""" insert into TestTempTable (IntCol, StringCol) values (:1, :2)""", (i, "The initial value of string %d" % i)) intVar = self.cursor.var(cx_Oracle.NUMBER) strVar = self.cursor.var(str) self.cursor.execute(""" update TestTempTable set IntCol = IntCol + 15, StringCol = 'The final value of string ' || to_char(IntCol) returning IntCol, StringCol into :intVar, :strVar""", intVar = intVar, strVar = strVar) self.assertEqual(self.cursor.rowcount, 3) self.assertEqual(intVar.values, [[23, 24, 25]]) self.assertEqual(strVar.values, [[ "The final value of string 8", "The final value of string 9", "The final value of string 10" ]]) def testUpdateMultipleRowsExecuteMany(self): "test update multiple rows with DML returning (executeMany)" data = [(i, "The initial value of string %d" % i) \ for i in range(1, 11)] self.cursor.execute("truncate table TestTempTable") self.cursor.executemany(""" insert into TestTempTable (IntCol, StringCol) values (:1, :2)""", data) intVar = self.cursor.var(cx_Oracle.NUMBER, arraysize = 3) strVar = self.cursor.var(str, arraysize = 3) self.cursor.setinputsizes(None, intVar, strVar) self.cursor.executemany(""" update TestTempTable set IntCol = IntCol + 25, StringCol = 'Updated value of string ' || to_char(IntCol) where IntCol < :inVal returning IntCol, StringCol into :intVar, :strVar""", [[3], [8], [11]]) self.assertEqual(intVar.values, [ [26, 27], [28, 29, 30, 31, 32], [33, 34, 35] ]) self.assertEqual(strVar.values, [ [ "Updated value of string 1", "Updated value of string 2" ], [ "Updated value of string 3", "Updated value of string 4", "Updated value of string 5", "Updated value of string 6", "Updated value of string 7" ], [ "Updated value of string 8", "Updated value of string 9", "Updated value of string 10" ] ]) def testInsertAndReturnObject(self): "test inserting an object with DML returning" typeObj = self.connection.gettype("UDT_OBJECT") stringValue = "The string that will be verified" obj = typeObj.newobject() obj.STRINGVALUE = stringValue outVar = self.cursor.var(cx_Oracle.DB_TYPE_OBJECT, typename = "UDT_OBJECT") self.cursor.execute(""" insert into TestObjects (IntCol, ObjectCol) values (4, :obj) returning ObjectCol into :outObj""", obj = obj, outObj = outVar) result, = outVar.getvalue() self.assertEqual(result.STRINGVALUE, stringValue) self.connection.rollback() def testInsertAndReturnRowid(self): "test inserting a row and returning a rowid" self.cursor.execute("truncate table TestTempTable") var = self.cursor.var(cx_Oracle.ROWID) self.cursor.execute(""" insert into TestTempTable (IntCol, StringCol) values (278, 'String 278') returning rowid into :1""", (var,)) rowid, = var.getvalue() self.cursor.execute(""" select IntCol, StringCol from TestTempTable where rowid = :1""", (rowid,)) self.assertEqual(self.cursor.fetchall(), [(278, 'String 278')]) def testInsertWithRefCursor(self): "test inserting with a REF cursor and returning a rowid" self.cursor.execute("truncate table TestTempTable") var = self.cursor.var(cx_Oracle.ROWID) inCursor = self.connection.cursor() inCursor.execute(""" select StringCol from TestStrings where IntCol >= 5 order by IntCol""") self.cursor.execute(""" insert into TestTempTable (IntCol, StringCol) values (187, pkg_TestRefCursors.TestInCursor(:1)) returning rowid into :2""", (inCursor, var)) rowid, = var.getvalue() self.cursor.execute(""" select IntCol, StringCol from TestTempTable where rowid = :1""", (rowid,)) self.assertEqual(self.cursor.fetchall(), [(187, 'String 7 (Modified)')]) def testDeleteReturningDecreasingRowsReturned(self): "test delete returning multiple times with decreasing number of rows" data = [(i, "Test String %d" % i) for i in range(1, 11)] self.cursor.execute("truncate table TestTempTable") self.cursor.executemany(""" insert into TestTempTable (IntCol, StringCol) values (:1, :2)""", data) results = [] intVar = self.cursor.var(int) self.cursor.setinputsizes(None, intVar) for intVal in (5, 8, 10): self.cursor.execute(""" delete from TestTempTable where IntCol < :1 returning IntCol into :2""", [intVal]) results.append(intVar.getvalue()) self.assertEqual(results, [ [1, 2, 3, 4], [5, 6, 7], [8, 9] ]) def testDeleteReturningNoRowsAfterManyRows(self): "test delete returning no rows after initially returning many rows" data = [(i, "Test String %d" % i) for i in range(1, 11)] self.cursor.execute("truncate table TestTempTable") self.cursor.executemany(""" insert into TestTempTable (IntCol, StringCol) values (:1, :2)""", data) intVar = self.cursor.var(int) self.cursor.execute(""" delete from TestTempTable where IntCol < :1 returning IntCol into :2""", [5, intVar]) self.assertEqual(intVar.getvalue(), [1, 2, 3, 4]) self.cursor.execute(None, [4, intVar]) self.assertEqual(intVar.getvalue(), []) if __name__ == "__main__": TestEnv.RunTestCases()