test_8600_cursor_scrollable_async.py

# -----------------------------------------------------------------------------
# Copyright (c) 2025, Oracle and/or its affiliates.
#
# This software is dual-licensed to you under the Universal Permissive License
# (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
# 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
# either license.
#
# If you elect to accept the software under the Apache License, Version 2.0,
# the following applies:
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------

"""
8600 - Module for testing scrollable cursors with asyncio
"""

importunittest

importtest_env


@unittest.skipUnless(
test_env.get_is_thin(), "asyncio not supported in thick mode"
)
classTestCase(test_env.BaseAsyncTestCase):
asyncdeftest_8600(self):
"8600 - test creating a scrollable cursor"
cursor=self.conn.cursor()
self.assertEqual(cursor.scrollable, False)
cursor=self.conn.cursor(True)
self.assertEqual(cursor.scrollable, True)
cursor=self.conn.cursor(scrollable=True)
self.assertEqual(cursor.scrollable, True)
cursor.scrollable=False
self.assertEqual(cursor.scrollable, False)

asyncdeftest_8601(self):
"8601 - test scrolling absolute yields an exception (after result set)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
withself.assertRaisesFullCode("DPY-2063"):
awaitcursor.scroll(12, "absolute")

asyncdeftest_8602(self):
"8602 - test scrolling absolute (when in buffers)"
cursor=self.conn.cursor(scrollable=True)
cursor.prefetchrows=0
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchmany()
self.assertTrue(
cursor.arraysize>1,
"array size must exceed 1 for this test to work correctly",
 )
awaitcursor.scroll(1, mode="absolute")
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 1.25)
self.assertEqual(cursor.rowcount, 1)

asyncdeftest_8603(self):
"8603 - test scrolling absolute (when not in buffers)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.scroll(6, mode="absolute")
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 7.5)
self.assertEqual(cursor.rowcount, 6)

asyncdeftest_8604(self):
"8604 - test scrolling to first row in result set (in buffers)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
cursor.prefetchrows=0
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchmany()
awaitcursor.scroll(mode="first")
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 1.25)
self.assertEqual(cursor.rowcount, 1)

asyncdeftest_8605(self):
"8605 - test scrolling to first row in result set (not in buffers)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
cursor.prefetchrows=0
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchmany()
awaitcursor.fetchmany()
awaitcursor.scroll(mode="first")
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 1.25)
self.assertEqual(cursor.rowcount, 1)

asyncdeftest_8606(self):
"8606 - test scrolling to last row in result set"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.scroll(mode="last")
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 12.5)
self.assertEqual(cursor.rowcount, 10)

asyncdeftest_8607(self):
"8607 - test scrolling relative yields an exception (after result set)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
withself.assertRaisesFullCode("DPY-2063"):
awaitcursor.scroll(15)

asyncdeftest_8608(self):
"8608 - test scrolling relative yields exception (before result set)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
withself.assertRaisesFullCode("DPY-2063"):
awaitcursor.scroll(-5)

asyncdeftest_8609(self):
"8609 - test scrolling relative (when in buffers)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
cursor.prefetchrows=0
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchmany()
message="array size must exceed 1 for this test to work correctly"
self.assertTrue(cursor.arraysize>1, message)
awaitcursor.scroll(2-cursor.rowcount)
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 2.5)
self.assertEqual(cursor.rowcount, 2)

asyncdeftest_8610(self):
"8610 - test scrolling relative (when not in buffers)"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchmany()
awaitcursor.fetchmany()
message="array size must exceed 1 for this test to work correctly"
self.assertTrue(cursor.arraysize>1, message)
awaitcursor.scroll(3-cursor.rowcount)
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 3.75)
self.assertEqual(cursor.rowcount, 3)

asyncdeftest_8611(self):
"8611 - test scrolling when there are no rows"
awaitself.cursor.execute("truncate table TestTempTable")
cursor=self.conn.cursor(scrollable=True)
awaitcursor.execute("select * from TestTempTable")
awaitcursor.scroll(mode="last")
self.assertEqual(awaitcursor.fetchall(), [])
awaitcursor.scroll(mode="first")
self.assertEqual(awaitcursor.fetchall(), [])
withself.assertRaisesFullCode("DPY-2063"):
awaitcursor.scroll(1, mode="absolute")

asyncdeftest_8612(self):
"8612 - test scrolling with differing array and fetch array sizes"
awaitself.cursor.execute("truncate table TestTempTable")
foriinrange(30):
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (:1, null)
 """,
 [i+1],
 )
forarraysizeinrange(1, 6):
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=arraysize
awaitcursor.execute(
"select IntCol from TestTempTable order by IntCol"
 )
fornum_rowsinrange(1, arraysize+1):
awaitcursor.scroll(15, "absolute")
rows=awaitcursor.fetchmany(num_rows)
self.assertEqual(rows[0][0], 15)
self.assertEqual(cursor.rowcount, 15+num_rows-1)
awaitcursor.scroll(9)
rows=awaitcursor.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
 )
awaitcursor.scroll(-12)
rows=awaitcursor.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)

asyncdeftest_8613(self):
"8613 - test calling scroll() with invalid mode"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=self.cursor.arraysize
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchmany()
withself.assertRaisesFullCode("DPY-2009"):
awaitcursor.scroll(mode="middle")

asyncdeftest_8614(self):
"8614 - test scroll after fetching all rows"
cursor=self.conn.cursor(scrollable=True)
cursor.arraysize=5
cursor.prefetchrows=0
awaitcursor.execute(
"select NumberCol from TestNumbers order by IntCol"
 )
awaitcursor.fetchall()
awaitcursor.scroll(5, mode="absolute")
 (value,) =awaitcursor.fetchone()
self.assertEqual(value, 6.25)
self.assertEqual(cursor.rowcount, 5)


if__name__=="__main__":
test_env.run_test_cases()