test_1200_cursor.py

#------------------------------------------------------------------------------
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
#
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
#
# Portions Copyright 2001-2007, Computronix (Canada) Ltd., Edmonton, Alberta,
# Canada. All rights reserved.
#------------------------------------------------------------------------------

"""
1200 - Module for testing cursors
"""

import test_env
import cx_Oracle as oracledb
import decimal

class TestCase(test_env.BaseTestCase):

    def test_1200_create_scrollable_cursor(self):
        "1200 - test creating a scrollable cursor"
        cursor = self.connection.cursor()
        self.assertEqual(cursor.scrollable, False)
        cursor = self.connection.cursor(True)
        self.assertEqual(cursor.scrollable, True)
        cursor = self.connection.cursor(scrollable=True)
        self.assertEqual(cursor.scrollable, True)
        cursor.scrollable = False
        self.assertEqual(cursor.scrollable, False)

    def test_1201_execute_no_args(self):
        "1201 - test executing a statement without any arguments"
        result = self.cursor.execute("begin null; end;")
        self.assertEqual(result, None)

    def test_1202_execute_no_statement_with_args(self):
        "1202 - test executing a None statement with bind variables"
        self.assertRaises(oracledb.ProgrammingError, self.cursor.execute, None,
                          x=5)

    def test_1203_execute_empty_keyword_args(self):
        "1203 - test executing a statement with args and empty keyword args"
        simple_var = self.cursor.var(oracledb.NUMBER)
        args = [simple_var]
        kwargs = {}
        result = self.cursor.execute("begin :1 := 25; end;", args, **kwargs)
        self.assertEqual(result, None)
        self.assertEqual(simple_var.getvalue(), 25)

    def test_1204_execute_keyword_args(self):
        "1204 - test executing a statement with keyword arguments"
        simple_var = self.cursor.var(oracledb.NUMBER)
        result = self.cursor.execute("begin :value := 5; end;",
                                     value=simple_var)
        self.assertEqual(result, None)
        self.assertEqual(simple_var.getvalue(), 5)

    def test_1205_execute_dictionary_arg(self):
        "1205 - test executing a statement with a dictionary argument"
        simple_var = self.cursor.var(oracledb.NUMBER)
        dict_arg = dict(value=simple_var)
        result = self.cursor.execute("begin :value := 10; end;", dict_arg)
        self.assertEqual(result, None)
        self.assertEqual(simple_var.getvalue(), 10)

    def test_1206_execute_multiple_arg_types(self):
        "1206 - test executing a statement with both a dict and keyword args"
        simple_var = self.cursor.var(oracledb.NUMBER)
        dict_arg = dict(value=simple_var)
        self.assertRaises(oracledb.InterfaceError, self.cursor.execute,
                          "begin :value := 15; end;", dict_arg,
                          value=simple_var)

    def test_1207_execute_and_modify_array_size(self):
        "1207 - test executing a statement and then changing the array size"
        self.cursor.execute("select IntCol from TestNumbers")
        self.cursor.arraysize = 20
        self.assertEqual(len(self.cursor.fetchall()), 10)

    def test_1208_callproc(self):
        "1208 - test executing a stored procedure"
        var = self.cursor.var(oracledb.NUMBER)
        results = self.cursor.callproc("proc_Test", ("hi", 5, var))
        self.assertEqual(results, ["hi", 10, 2.0])

    def test_1209_callproc_all_keywords(self):
        "1209 - test executing a stored procedure with all args keyword args"
        inout_value = self.cursor.var(oracledb.NUMBER)
        inout_value.setvalue(0, 5)
        out_value = self.cursor.var(oracledb.NUMBER)
        kwargs = dict(a_InOutValue=inout_value, a_InValue="hi",
                      a_OutValue=out_value)
        results = self.cursor.callproc("proc_Test", [], kwargs)
        self.assertEqual(results, [])
        self.assertEqual(inout_value.getvalue(), 10)
        self.assertEqual(out_value.getvalue(), 2.0)

    def test_1210_callproc_only_last_keyword(self):
        "1210 - test executing a stored procedure with last arg as keyword arg"
        out_value = self.cursor.var(oracledb.NUMBER)
        kwargs = dict(a_OutValue=out_value)
        results = self.cursor.callproc("proc_Test", ("hi", 5), kwargs)
        self.assertEqual(results, ["hi", 10])
        self.assertEqual(out_value.getvalue(), 2.0)

    def test_1211_callproc_repeated_keyword_parameters(self):
        "1211 - test executing a stored procedure, repeated keyword arg"
        kwargs = dict(a_InValue="hi",
                      a_OutValue=self.cursor.var(oracledb.NUMBER))
        self.assertRaises(oracledb.DatabaseError, self.cursor.callproc,
                          "proc_Test", ("hi", 5), kwargs)

    def test_1212_callproc_no_args(self):
        "1212 - test executing a stored procedure without any arguments"
        results = self.cursor.callproc("proc_TestNoArgs")
        self.assertEqual(results, [])

    def test_1213_callfunc(self):
        "1213 - test executing a stored function"
        results = self.cursor.callfunc("func_Test", oracledb.NUMBER, ("hi", 5))
        self.assertEqual(results, 7)

    def test_1214_callfunc_no_args(self):
        "1214 - test executing a stored function without any arguments"
        results = self.cursor.callfunc("func_TestNoArgs", oracledb.NUMBER)
        self.assertEqual(results, 712)

    def test_1215_callfunc_negative(self):
        "1215 - test executing a stored function with wrong parameters"
        func_name = "func_Test"
        self.assertRaises(TypeError, self.cursor.callfunc, oracledb.NUMBER,
                          func_name, ("hi", 5))
        self.assertRaises(oracledb.DatabaseError, self.cursor.callfunc,
                          func_name, oracledb.NUMBER, ("hi", 5, 7))
        self.assertRaises(TypeError, self.cursor.callfunc, func_name,
                          oracledb.NUMBER, "hi", 7)
        self.assertRaises(oracledb.DatabaseError, self.cursor.callfunc,
                          func_name, oracledb.NUMBER, [5, "hi"])
        self.assertRaises(oracledb.DatabaseError, self.cursor.callfunc,
                          func_name, oracledb.NUMBER)
        self.assertRaises(TypeError, self.cursor.callfunc, func_name,
                          oracledb.NUMBER, 5)

    def test_1216_executemany_by_name(self):
        "1216 - test executing a statement multiple times (named args)"
        self.cursor.execute("truncate table TestTempTable")
        rows = [{"value": n} for n in range(250)]
        self.cursor.arraysize = 100
        statement = "insert into TestTempTable (IntCol) values (:value)"
        self.cursor.executemany(statement, rows)
        self.connection.commit()
        self.cursor.execute("select count(*) from TestTempTable")
        count, = self.cursor.fetchone()
        self.assertEqual(count, len(rows))

    def test_1217_executemany_by_position(self):
        "1217 - test executing a statement multiple times (positional args)"
        self.cursor.execute("truncate table TestTempTable")
        rows = [[n] for n in range(230)]
        self.cursor.arraysize = 100
        statement = "insert into TestTempTable (IntCol) values (:1)"
        self.cursor.executemany(statement, rows)
        self.connection.commit()
        self.cursor.execute("select count(*) from TestTempTable")
        count, = self.cursor.fetchone()
        self.assertEqual(count, len(rows))

    def test_1218_executemany_with_prepare(self):
        "1218 - test executing a statement multiple times (with prepare)"
        self.cursor.execute("truncate table TestTempTable")
        rows = [[n] for n in range(225)]
        self.cursor.arraysize = 100
        statement = "insert into TestTempTable (IntCol) values (:1)"
        self.cursor.prepare(statement)
        self.cursor.executemany(None, rows)
        self.connection.commit()
        self.cursor.execute("select count(*) from TestTempTable")
        count, = self.cursor.fetchone()
        self.assertEqual(count, len(rows))

    def test_1219_executemany_with_rebind(self):
        "1219 - test executing a statement multiple times (with rebind)"
        self.cursor.execute("truncate table TestTempTable")
        rows = [[n] for n in range(235)]
        self.cursor.arraysize = 100
        statement = "insert into TestTempTable (IntCol) values (:1)"
        self.cursor.executemany(statement, rows[:50])
        self.cursor.executemany(statement, rows[50:])
        self.connection.commit()
        self.cursor.execute("select count(*) from TestTempTable")
        count, = self.cursor.fetchone()
        self.assertEqual(count, len(rows))

    def test_1220_executemany_with_input_sizes_wrong(self):
        "1220 - test executing multiple times (with input sizes wrong)"
        cursor = self.connection.cursor()
        cursor.setinputsizes(oracledb.NUMBER)
        data = [[decimal.Decimal("25.8")], [decimal.Decimal("30.0")]]
        cursor.executemany("declare t number; begin t := :1; end;", data)

    def test_1221_executemany_with_multiple_batches(self):
        "1221 - test executing multiple times (with multiple batches)"
        self.cursor.execute("truncate table TestTempTable")
        sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
        self.cursor.executemany(sql, [(1, None), (2, None)])
        self.cursor.executemany(sql, [(3, None), (4, "Testing")])

    def test_1222_executemany_numeric(self):
        "1222 - test executemany() with various numeric types"
        self.cursor.execute("truncate table TestTempTable")
        data = [
            (1, 5),
            (2, 7.0),
            (3, 6.5),
            (4, 2 ** 65),
            (5, decimal.Decimal("24.5"))
        ]
        sql = "insert into TestTempTable (IntCol, NumberCol) values (:1, :2)"
        self.cursor.executemany(sql, data)
        self.cursor.execute("""
                select IntCol, NumberCol
                from TestTempTable
                order by IntCol""")
        self.assertEqual(self.cursor.fetchall(), data)

    def test_1223_executemany_with_resize(self):
        "1223 - test executing a statement multiple times (with resize)"
        self.cursor.execute("truncate table TestTempTable")
        rows = [
             (1, "First"),
             (2, "Second"),
             (3, "Third"),
             (4, "Fourth"),
             (5, "Fifth"),
             (6, "Sixth"),
             (7, "Seventh and the longest one")
        ]
        sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
        self.cursor.executemany(sql, rows)
        self.cursor.execute("""
                select IntCol, StringCol
                from TestTempTable
                order by IntCol""")
        fetched_rows = self.cursor.fetchall()
        self.assertEqual(fetched_rows, rows)

    def test_1224_executemany_with_exception(self):
        "1224 - test executing a statement multiple times (with exception)"
        self.cursor.execute("truncate table TestTempTable")
        rows = [{"value": n} for n in (1, 2, 3, 2, 5)]
        statement = "insert into TestTempTable (IntCol) values (:value)"
        self.assertRaises(oracledb.DatabaseError, self.cursor.executemany,
                          statement, rows)
        self.assertEqual(self.cursor.rowcount, 3)

    def test_1225_executemany_with_invalid_parameters(self):
        "1225 - test calling executemany() with invalid parameters"
        sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
        self.assertRaises(TypeError, self.cursor.executemany, sql,
                          "These are not valid parameters")

    def test_1226_executemany_no_parameters(self):
        "1226 - test calling executemany() without any bind parameters"
        num_rows = 5
        self.cursor.execute("truncate table TestTempTable")
        self.cursor.executemany("""
                declare
                    t_Id          number;
                begin
                    select nvl(count(*), 0) + 1 into t_Id
                    from TestTempTable;

                    insert into TestTempTable (IntCol, StringCol)
                    values (t_Id, 'Test String ' || t_Id);
                end;""", num_rows)
        self.assertEqual(self.cursor.rowcount, num_rows)
        self.cursor.execute("select count(*) from TestTempTable")
        count, = self.cursor.fetchone()
        self.assertEqual(count, num_rows)

    def test_1227_executemany_bound_earlier(self):
        "1227 - test calling executemany() with binds performed earlier"
        num_rows = 9
        self.cursor.execute("truncate table TestTempTable")
        var = self.cursor.var(int, arraysize=num_rows)
        self.cursor.setinputsizes(var)
        self.cursor.executemany("""
                declare
                    t_Id          number;
                begin
                    select nvl(count(*), 0) + 1 into t_Id
                    from TestTempTable;

                    insert into TestTempTable (IntCol, StringCol)
                    values (t_Id, 'Test String ' || t_Id);

                    select sum(IntCol) into :1
                    from TestTempTable;
                end;""", num_rows)
        self.assertEqual(self.cursor.rowcount, num_rows)
        expected_data = [1, 3, 6, 10, 15, 21, 28, 36, 45]
        self.assertEqual(var.values, expected_data)

    def test_1228_prepare(self):
        "1228 - test preparing a statement and executing it multiple times"
        self.assertEqual(self.cursor.statement, None)
        statement = "begin :value := :value + 5; end;"
        self.cursor.prepare(statement)
        var = self.cursor.var(oracledb.NUMBER)
        self.assertEqual(self.cursor.statement, statement)
        var.setvalue(0, 2)
        self.cursor.execute(None, value = var)
        self.assertEqual(var.getvalue(), 7)
        self.cursor.execute(None, value = var)
        self.assertEqual(var.getvalue(), 12)
        self.cursor.execute("begin :value2 := 3; end;", value2 = var)
        self.assertEqual(var.getvalue(), 3)

    def test_1229_exception_on_close(self):
        "1229 - confirm an exception is raised after closing a cursor"
        self.cursor.close()
        self.assertRaises(oracledb.InterfaceError, self.cursor.execute,
                          "select 1 from dual")

    def test_1230_iterators(self):
        "1230 - test iterators"
        self.cursor.execute("""
                select IntCol
                from TestNumbers
                where IntCol between 1 and 3
                order by IntCol""")
        rows = [v for v, in self.cursor]
        self.assertEqual(rows, [1, 2, 3])

    def test_1231_iterators_interrupted(self):
        "1231 - test iterators (with intermediate execute)"
        self.cursor.execute("truncate table TestTempTable")
        self.cursor.execute("""
                select IntCol
                from TestNumbers
                where IntCol between 1 and 3
                order by IntCol""")
        test_iter = iter(self.cursor)
        value, = next(test_iter)
        self.cursor.execute("insert into TestTempTable (IntCol) values (1)")
        self.assertRaises(oracledb.InterfaceError, next, test_iter)

    def test_1232_bind_names(self):
        "1232 - test that bindnames() works correctly."
        self.assertRaises(oracledb.ProgrammingError, self.cursor.bindnames)
        self.cursor.prepare("begin null; end;")
        self.assertEqual(self.cursor.bindnames(), [])
        self.cursor.prepare("begin :retval := :inval + 5; end;")
        self.assertEqual(self.cursor.bindnames(), ["RETVAL", "INVAL"])
        self.cursor.prepare("begin :retval := :a * :a + :b * :b; end;")
        self.assertEqual(self.cursor.bindnames(), ["RETVAL", "A", "B"])
        self.cursor.prepare("begin :a := :b + :c + :d + :e + :f + :g + " + \
                            ":h + :i + :j + :k + :l; end;")
        names = ["A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L"]
        self.assertEqual(self.cursor.bindnames(), names)
        self.cursor.prepare("select :a * :a + :b * :b from dual")
        self.assertEqual(self.cursor.bindnames(), ["A", "B"])

    def test_1233_bad_execute(self):
        "1233 - test that subsequent executes succeed after bad execute"
        self.assertRaises(oracledb.DatabaseError,
                          self.cursor.execute,
                          "begin raise_application_error(-20000, 'this); end;")
        self.cursor.execute("begin null; end;")

    def test_1234_fetch_after_bad_execute(self):
        "1234 - test that subsequent fetches fail after bad execute"
        self.assertRaises(oracledb.DatabaseError,
                          self.cursor.execute, "select y from dual")
        self.assertRaises(oracledb.InterfaceError, self.cursor.fetchall)

    def test_1235_scroll_absolute_exception_after(self):
        "1235 - test scrolling absolute yields an exception (after result set)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        self.assertRaises(oracledb.DatabaseError, cursor.scroll, 12,
                          "absolute")

    def test_1236_scroll_absolute_in_buffer(self):
        "1236 - test scrolling absolute (when in buffers)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.fetchmany()
        self.assertTrue(cursor.arraysize > 1,
                "array size must exceed 1 for this test to work correctly")
        cursor.scroll(1, mode = "absolute")
        row = cursor.fetchone()
        self.assertEqual(row[0], 1.25)
        self.assertEqual(cursor.rowcount, 1)

    def test_1237_scroll_absolute_not_in_buffer(self):
        "1237 - test scrolling absolute (when not in buffers)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.scroll(6, mode = "absolute")
        row = cursor.fetchone()
        self.assertEqual(row[0], 7.5)
        self.assertEqual(cursor.rowcount, 6)

    def test_1238_scroll_first_in_buffer(self):
        "1238 - test scrolling to first row in result set (in buffers)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.fetchmany()
        cursor.scroll(mode="first")
        row = cursor.fetchone()
        self.assertEqual(row[0], 1.25)
        self.assertEqual(cursor.rowcount, 1)

    def test_1239_scroll_first_not_in_buffer(self):
        "1239 - test scrolling to first row in result set (not in buffers)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.fetchmany()
        cursor.fetchmany()
        cursor.scroll(mode="first")
        row = cursor.fetchone()
        self.assertEqual(row[0], 1.25)
        self.assertEqual(cursor.rowcount, 1)

    def test_1240_scroll_last(self):
        "1240 - test scrolling to last row in result set"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.scroll(mode="last")
        row = cursor.fetchone()
        self.assertEqual(row[0], 12.5)
        self.assertEqual(cursor.rowcount, 10)

    def test_1241_scroll_relative_exception_after(self):
        "1241 - test scrolling relative yields an exception (after result set)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        self.assertRaises(oracledb.DatabaseError, cursor.scroll, 15)

    def test_1242_scroll_relative_exception_before(self):
        "1242 - test scrolling relative yields exception (before result set)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        self.assertRaises(oracledb.DatabaseError, cursor.scroll, -5)

    def test_1243_scroll_relative_in_buffer(self):
        "1243 - test scrolling relative (when in buffers)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.fetchmany()
        message = "array size must exceed 1 for this test to work correctly"
        self.assertTrue(cursor.arraysize > 1, message)
        cursor.scroll(2 - cursor.rowcount)
        row = cursor.fetchone()
        self.assertEqual(row[0], 2.5)
        self.assertEqual(cursor.rowcount, 2)

    def test_1244_scroll_relative_not_in_buffer(self):
        "1244 - test scrolling relative (when not in buffers)"
        cursor = self.connection.cursor(scrollable=True)
        cursor.arraysize = self.cursor.arraysize
        cursor.execute("""
                select NumberCol
                from TestNumbers
                order by IntCol""")
        cursor.fetchmany()
        cursor.fetchmany()
        message = "array size must exceed 1 for this test to work correctly"
        self.assertTrue(cursor.arraysize > 1, message)
        cursor.scroll(3 - cursor.rowcount)
        row = cursor.fetchone()
        self.assertEqual(row[0], 3.75)
        self.assertEqual(cursor.rowcount, 3)

    def test_1245_scroll_no_rows(self):
        "1245 - test scrolling when there are no rows"
        self.cursor.execute("truncate table TestTempTable")
        cursor = self.connection.cursor(scrollable=True)
        cursor.execute("select * from TestTempTable")
        cursor.scroll(mode = "last")
        self.assertEqual(cursor.fetchall(), [])
        cursor.scroll(mode = "first")
        self.assertEqual(cursor.fetchall(), [])
        self.assertRaises(oracledb.DatabaseError, cursor.scroll, 1,
                          mode="absolute")

    def test_1246_scroll_differing_array_and_fetch_sizes(self):
        "1246 - test scrolling with differing array and fetch array sizes"
        self.cursor.execute("truncate table TestTempTable")
        for i in range(30):
            self.cursor.execute("""
                    insert into TestTempTable (IntCol, StringCol)
                    values (:1, null)""",
                    (i + 1,))
        for arraysize in range(1, 6):
            cursor = self.connection.cursor(scrollable = True)
            cursor.arraysize = arraysize
            cursor.execute("select IntCol from TestTempTable order by IntCol")
            for num_rows in range(1, arraysize + 1):
                cursor.scroll(15, "absolute")
                rows = cursor.fetchmany(num_rows)
                self.assertEqual(rows[0][0], 15)
                self.assertEqual(cursor.rowcount, 15 + num_rows - 1)
                cursor.scroll(9)
                rows = cursor.fetchmany(num_rows)
                num_rows_fetched = len(rows)
                self.assertEqual(rows[0][0], 15 + num_rows + 8)
                self.assertEqual(cursor.rowcount,
                        15 + num_rows + num_rows_fetched + 7)
                cursor.scroll(-12)
                rows = cursor.fetchmany(num_rows)
                count = 15 + num_rows + num_rows_fetched - 5
                self.assertEqual(rows[0][0], count)
                count = 15 + num_rows + num_rows_fetched + num_rows - 6
                self.assertEqual(cursor.rowcount, count)

    def test_1247_set_input_sizes_negative(self):
        "1247 - test cursor.setinputsizes() with invalid parameters"
        val = decimal.Decimal(5)
        self.assertRaises(oracledb.InterfaceError,
                          self.cursor.setinputsizes, val, x=val)
        self.assertRaises(TypeError, self.cursor.setinputsizes, val)

    def test_1248_set_input_sizes_no_parameters(self):
        "1248 - test setting input sizes without any parameters"
        self.cursor.setinputsizes()
        self.cursor.execute("select :val from dual", val="Test Value")
        self.assertEqual(self.cursor.fetchall(), [("Test Value",)])

    def test_1249_set_input_sizes_empty_dict(self):
        "1249 - test setting input sizes with an empty dictionary"
        empty_dict = {}
        self.cursor.prepare("select 236 from dual")
        self.cursor.setinputsizes(**empty_dict)
        self.cursor.execute(None, empty_dict)
        self.assertEqual(self.cursor.fetchall(), [(236,)])

    def test_1250_set_input_sizes_empty_list(self):
        "1250 - test setting input sizes with an empty list"
        empty_list = {}
        self.cursor.prepare("select 239 from dual")
        self.cursor.setinputsizes(*empty_list)
        self.cursor.execute(None, empty_list)
        self.assertEqual(self.cursor.fetchall(), [(239,)])

    def test_1251_set_input_sizes_by_position(self):
        "1251 - test setting input sizes with positional args"
        var = self.cursor.var(oracledb.STRING, 100)
        self.cursor.setinputsizes(None, 5, None, 10, None, oracledb.NUMBER)
        self.cursor.execute("""
                begin
                  :1 := :2 || to_char(:3) || :4 || to_char(:5) || to_char(:6);
                end;""", [var, 'test_', 5, '_second_', 3, 7])
        self.assertEqual(var.getvalue(), "test_5_second_37")

    def test_1252_string_format(self):
        "1252 - test string format of cursor"
        format_string = "<cx_Oracle.Cursor on <cx_Oracle.Connection to %s@%s>>"
        expected_value = format_string % \
                (test_env.get_main_user(), test_env.get_connect_string())
        self.assertEqual(str(self.cursor), expected_value)

    def test_1253_cursor_fetch_raw(self):
        "1253 - test cursor.fetchraw()"
        cursor = self.connection.cursor()
        cursor.arraysize = 25
        cursor.execute("select LongIntCol from TestNumbers order by IntCol")
        self.assertEqual(cursor.fetchraw(), 10)
        self.assertEqual(cursor.fetchvars[0].getvalue(), 38)

    def test_1254_parse(self):
        "1254 - test parsing statements"
        sql = "select LongIntCol from TestNumbers where IntCol = :val"
        self.cursor.parse(sql)
        self.assertEqual(self.cursor.statement, sql)
        self.assertEqual(self.cursor.description,
                         [('LONGINTCOL', oracledb.DB_TYPE_NUMBER, 17, None,
                           16, 0, 0)])

    def test_1255_set_output_size(self):
        "1255 - test cursor.setoutputsize() does not fail (but does nothing)"
        self.cursor.setoutputsize(100, 2)

    def test_1256_var_negative(self):
        "1256 - test cursor.var() with invalid parameters"
        self.assertRaises(TypeError, self.cursor.var, 5)

    def test_1257_arrayvar_negative(self):
        "1257 - test cursor.arrayvar() with invalid parameters"
        self.assertRaises(TypeError, self.cursor.arrayvar, 5, 1)

    def test_1258_boolean_without_plsql(self):
        "1258 - test binding boolean data without the use of PL/SQL"
        self.cursor.execute("truncate table TestTempTable")
        sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
        self.cursor.execute(sql, (False, "Value should be 0"))
        self.cursor.execute(sql, (True, "Value should be 1"))
        self.cursor.execute("""
                select IntCol, StringCol
                from TestTempTable
                order by IntCol""")
        self.assertEqual(self.cursor.fetchall(),
                         [(0, "Value should be 0"), (1, "Value should be 1")])

    def test_1259_as_context_manager(self):
        "1259 - test using a cursor as a context manager"
        with self.cursor as cursor:
            cursor.execute("truncate table TestTempTable")
            cursor.execute("select count(*) from TestTempTable")
            count, = cursor.fetchone()
            self.assertEqual(count, 0)
        self.assertRaises(oracledb.InterfaceError, self.cursor.close)

    def test_1260_query_row_count(self):
        "1260 - test that rowcount attribute is reset to zero on query execute"
        sql = "select * from dual where 1 = :s"
        self.cursor.execute(sql, [0])
        self.cursor.fetchone()
        self.assertEqual(self.cursor.rowcount, 0)
        self.cursor.execute(sql, [1])
        self.cursor.fetchone()
        self.assertEqual(self.cursor.rowcount, 1)
        self.cursor.execute(sql, [1])
        self.cursor.fetchone()
        self.assertEqual(self.cursor.rowcount, 1)
        self.cursor.execute(sql, [0])
        self.cursor.fetchone()
        self.assertEqual(self.cursor.rowcount, 0)

    def test_1261_var_type_name_none(self):
        "1261 - test that the typename attribute can be passed a value of None"
        value_to_set = 5
        var = self.cursor.var(int, typename=None)
        var.setvalue(0, value_to_set)
        self.assertEqual(var.getvalue(), value_to_set)

    def test_1262_var_type_with_object_type(self):
        "1262 - test that an object type can be used as type in cursor.var()"
        obj_type = self.connection.gettype("UDT_OBJECT")
        var = self.cursor.var(obj_type)
        self.cursor.callproc("pkg_TestBindObject.BindObjectOut",
                             (28, "Bind obj out", var))
        obj = var.getvalue()
        result = self.cursor.callfunc("pkg_TestBindObject.GetStringRep", str,
                                      (obj,))
        exp = "udt_Object(28, 'Bind obj out', null, null, null, null, null)"
        self.assertEqual(result, exp)

    def test_1263_fetch_xmltype(self):
        "1263 - test that fetching an XMLType returns a string"
        int_val = 5
        label = "IntCol"
        expected_result = "<%s>%s</%s>" % (label, int_val, label)
        self.cursor.execute("""
                select XMLElement("%s", IntCol)
                from TestStrings
                where IntCol = :int_val""" % label,
                int_val=int_val)
        result, = self.cursor.fetchone()
        self.assertEqual(result, expected_result)

    def test_1264_lastrowid(self):
        "1264 - test last rowid"

        # no statement executed: no rowid
        self.assertEqual(None, self.cursor.lastrowid)

        # DDL statement executed: no rowid
        self.cursor.execute("truncate table TestTempTable")
        self.assertEqual(None, self.cursor.lastrowid)

        # statement prepared: no rowid
        self.cursor.prepare("insert into TestTempTable (IntCol) values (:1)")
        self.assertEqual(None, self.cursor.lastrowid)

        # multiple rows inserted: rowid of last row inserted
        rows = [(n,) for n in range(225)]
        self.cursor.executemany(None, rows)
        rowid = self.cursor.lastrowid
        self.cursor.execute("""
                select rowid
                from TestTempTable
                where IntCol = :1""", rows[-1])
        self.assertEqual(rowid, self.cursor.fetchone()[0])

        # statement executed but no rows updated: no rowid
        self.cursor.execute("delete from TestTempTable where 1 = 0")
        self.assertEqual(None, self.cursor.lastrowid)

        # stetement executed with one row updated: rowid of updated row
        self.cursor.execute("""
                update TestTempTable set
                    StringCol = 'Modified'
                where IntCol = :1""", rows[-2])
        rowid = self.cursor.lastrowid
        self.cursor.execute("""
                select rowid
                from TestTempTable
                where IntCol = :1""", rows[-2])
        self.assertEqual(rowid, self.cursor.fetchone()[0])

        # statement executed with many rows updated: rowid of last updated row
        self.cursor.execute("""
                update TestTempTable set
                    StringCol = 'Row ' || to_char(IntCol)
                where IntCol = :1""", rows[-3])
        rowid = self.cursor.lastrowid
        self.cursor.execute("""
                select StringCol
                from TestTempTable
                where rowid = :1""", [rowid])
        self.assertEqual("Row %s" % rows[-3], self.cursor.fetchone()[0])

    def test_1265_prefetchrows(self):
        "1265 - test prefetch rows"
        self.setup_round_trip_checker()

        # perform simple query and verify only one round trip is needed
        with self.connection.cursor() as cursor:
            cursor.execute("select sysdate from dual").fetchall()
            self.assertRoundTrips(1)

        # set prefetchrows to 1 and verify that two round trips are now needed
        with self.connection.cursor() as cursor:
            cursor.prefetchrows = 1
            cursor.execute("select sysdate from dual").fetchall()
            self.assertRoundTrips(2)

        # simple DDL only requires a single round trip
        with self.connection.cursor() as cursor:
            cursor.execute("truncate table TestTempTable")
            self.assertRoundTrips(1)

        # array execution only requires a single round trip
        num_rows = 590
        with self.connection.cursor() as cursor:
            sql = "insert into TestTempTable (IntCol) values (:1)"
            data = [(n + 1,) for n in range(num_rows)]
            cursor.executemany(sql, data)
            self.assertRoundTrips(1)

        # setting prefetch and array size to 1 requires a round-trip for each
        # row
        with self.connection.cursor() as cursor:
            cursor.prefetchrows = 1
            cursor.arraysize = 1
            cursor.execute("select IntCol from TestTempTable").fetchall()
            self.assertRoundTrips(num_rows + 1)

        # setting prefetch and array size to 300 requires 2 round-trips
        with self.connection.cursor() as cursor:
            cursor.prefetchrows = 300
            cursor.arraysize = 300
            cursor.execute("select IntCol from TestTempTable").fetchall()
            self.assertRoundTrips(2)

    def test_1266_refcursor_prefetchrows(self):
        "1266 - test prefetch rows and arraysize using a refcursor"
        self.setup_round_trip_checker()

        # simple DDL only requires a single round trip
        with self.connection.cursor() as cursor:
            cursor.execute("truncate table TestTempTable")
            self.assertRoundTrips(1)

        # array execution only requires a single round trip
        num_rows = 590
        with self.connection.cursor() as cursor:
            sql = "insert into TestTempTable (IntCol) values (:1)"
            data = [(n + 1,) for n in range(num_rows)]
            cursor.executemany(sql, data)
            self.assertRoundTrips(1)

        # create refcursor and execute stored procedure
        with self.connection.cursor() as cursor:
            refcursor = self.connection.cursor()
            refcursor.prefetchrows = 300
            refcursor.arraysize = 300
            cursor.callproc("myrefcursorproc", [refcursor])
            refcursor.fetchall()
            self.assertRoundTrips(2)

    def test_1267_existing_cursor_prefetchrows(self):
        "1267 - test prefetch rows using existing cursor"
        self.setup_round_trip_checker()

        # Set prefetch rows on an existing cursor
        num_rows = 590
        with self.connection.cursor() as cursor:
            cursor.execute("truncate table TestTempTable")
            sql = "insert into TestTempTable (IntCol) values (:1)"
            data = [(n + 1,) for n in range(num_rows)]
            cursor.executemany(sql, data)
            cursor.prefetchrows = 300
            cursor.arraysize = 300
            cursor.execute("select IntCol from TestTempTable").fetchall()
            self.assertRoundTrips(4)

    def test_1268_bind_names_with_single_line_comments(self):
        "1268 - test bindnames() with single line comments"
        self.cursor.prepare("""--begin :value2 := :a + :b + :c +:a +3; end;
                            begin :value2 := :a + :c +3; end;
                            """)
        self.assertEqual(self.cursor.bindnames(), ["VALUE2", "A", "C"])

    def test_1269_bind_names_with_multi_line_comments(self):
        "1269 - test bindnames() with multi line comments"
        self.cursor.prepare("""/*--select * from :a where :a = 1
                            select * from table_names where :a = 1*/
                            select * from :table_name where :value = 1
                            """)
        self.assertEqual(self.cursor.bindnames(), ["TABLE_NAME", "VALUE"])

    def test_1270_execute_bind_names_with_incorrect_bind(self):
        "1270 - test executing a statement with an incorrect named bind"
        statement = "select * from TestStrings where IntCol = :value"
        self.assertRaises(oracledb.DatabaseError, self.cursor.execute,
                          statement, value2=3)

    def test_1271_execute_with_named_binds(self):
        "1271 - test executing a statement with named binds"
        statement = "select * from TestNumbers where IntCol = :value1 " + \
                    "and LongIntCol = :value2"
        result = self.cursor.execute(statement, value1=1, value2=38)
        self.assertEqual(len(result.fetchall()), 1)

    def test_1272_execute_bind_position_with_incorrect_bind(self):
        "1272 - test executing a statement with an incorrect positional bind"
        statement = "select * from TestNumbers where IntCol = :value " + \
                    "and LongIntCol = :value2"
        self.assertRaises(oracledb.DatabaseError, self.cursor.execute,
                          statement, [3])

    def test_1273_execute_with_positional_binds(self):
        "1273 - test executing a statement with positional binds"
        statement = "select * from TestNumbers where IntCol = :value " + \
                    "and LongIntCol = :value2"
        result = self.cursor.execute(statement, [1,38])
        self.assertEqual(len(result.fetchall()), 1)

    def test_1274_execute_with_rebinding_bind_name(self):
        "1274 - test executing a statement after rebinding a named bind"
        statement = "begin :value := :value2 + 5; end;"
        simple_var = self.cursor.var(oracledb.NUMBER)
        simple_var2 = self.cursor.var(oracledb.NUMBER)
        simple_var2.setvalue(0, 5)
        result = self.cursor.execute(statement, value=simple_var,
                                     value2=simple_var2)
        self.assertEqual(result, None)
        self.assertEqual(simple_var.getvalue(), 10)

        simple_var = self.cursor.var(oracledb.NATIVE_FLOAT)
        simple_var2 = self.cursor.var(oracledb.NATIVE_FLOAT)
        simple_var2.setvalue(0, 10)
        result = self.cursor.execute(statement, value=simple_var,
                                     value2=simple_var2)
        self.assertEqual(result, None)
        self.assertEqual(simple_var.getvalue(), 15)

    def test_1275_bind_names_with_strings(self):
        "1275 - test bindnames() with strings in the statement"
        statement = """
                    begin
                    :value := to_date('20021231 12:31:00',
                    'YYYYMMDD HH24:MI:SS');
                    end;"""
        self.cursor.prepare(statement)
        self.assertEqual(self.cursor.bindnames(), ["VALUE"])

    def test_1276_bind_by_name_with_duplicates(self):
        "1276 - test executing a PL/SQL statement with duplicate binds"
        statement = "begin :value := :value + 5; end;"
        simple_var = self.cursor.var(oracledb.NUMBER)
        simple_var.setvalue(0, 5)
        result = self.cursor.execute(statement, value=simple_var)
        self.assertEqual(result, None)
        self.assertEqual(simple_var.getvalue(), 10)

    def test_1277_positional_bind_with_duplicates(self):
        "1277 - test executing a PL/SQL statement with duplicate binds"
        statement = "begin :value := :value + 5; end;"
        simple_var = self.cursor.var(oracledb.NUMBER)
        simple_var.setvalue(0, 5)
        self.cursor.execute(statement, [simple_var])
        self.assertEqual(simple_var.getvalue(), 10)

    def test_1278_execute_with_incorrect_bind_values(self):
        "1278 - test executing a statement with an incorrect number of binds"
        statement = "begin :value := :value2 + 5; end;"
        var = self.cursor.var(oracledb.NUMBER)
        var.setvalue(0, 5)
        self.assertRaises(oracledb.DatabaseError, self.cursor.execute,
                          statement)
        self.assertRaises(oracledb.DatabaseError, self.cursor.execute,
                          statement, value=var)
        self.assertRaises(oracledb.DatabaseError, self.cursor.execute,
                          statement, value=var, value2=var, value3=var)

if __name__ == "__main__":
    test_env.run_test_cases()