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

Improve test suite.

Browse Source
This commit is contained in:
Anthony Tuininga 2021-05-18 16:58:03 -06:00
parent 0633e1017c
commit cd21e92edd
2 changed files with 77 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
test/test_2400_session_pool.py
View File

@ -32,6 +32,36 @@ class TestCase(test_env.BaseTestCase):
self.assertRaises(oracledb.DatabaseError, cursor.execute,
"select 1 / 0 from dual")
def __perform_reconfigure_test(self, parameter_name, parameter_value,
min=3, max=30, increment=4, timeout=5,
wait_timeout=5000, stmtcachesize=25,
max_lifetime_session=1000,
max_sessions_per_shard=3, ping_interval=30,
getmode=oracledb.SPOOL_ATTRVAL_WAIT):
creation_args = dict(min=min, max=max, increment=increment,
timeout=timeout, wait_timeout=wait_timeout,
stmtcachesize=stmtcachesize,
max_lifetime_session=max_lifetime_session,
max_sessions_per_shard=max_sessions_per_shard,
ping_interval=ping_interval, getmode=getmode)
reconfigure_args = {}
reconfigure_args[parameter_name] = parameter_value
pool = test_env.get_pool(**creation_args)
connection = pool.acquire()
pool.reconfigure(**reconfigure_args)
actual_args = dict(min=pool.min, max=pool.max,
increment=pool.increment, timeout=pool.timeout,
wait_timeout=pool.wait_timeout,
stmtcachesize=pool.stmtcachesize,
max_lifetime_session=pool.max_lifetime_session,
max_sessions_per_shard=pool.max_sessions_per_shard,
ping_interval=pool.ping_interval,
getmode=pool.getmode)
expected_args = creation_args.copy()
expected_args.update(reconfigure_args)
self.assertEqual(actual_args, expected_args)
def __verify_connection(self, connection, expected_user,
expected_proxy_user=None):
cursor = connection.cursor()
@ -376,37 +406,40 @@ class TestCase(test_env.BaseTestCase):
self.assertEqual(pool.stmtcachesize, 30, "stmtcachesize (30)")
self.assertEqual(pool.ping_interval, 30, "ping_interval (30)")
def test_2416_reconfigure_pool_with_missing_params(self):
"2416 - test to ensure reconfigure uses initial values if unspecified"
pool = test_env.get_pool(min=1, max=2, increment=1)
def test_2416_test_reconfigure_pool_with_missing_values(self):
"2416 - test the reconfigure values are changed and rest unchanged"
self.__perform_reconfigure_test("min", 5)
self.__perform_reconfigure_test("max", 20)
self.__perform_reconfigure_test("increment", 5)
self.__perform_reconfigure_test("timeout", 10)
self.__perform_reconfigure_test("wait_timeout", 8000)
self.__perform_reconfigure_test("stmtcachesize", 40)
self.__perform_reconfigure_test("max_lifetime_session", 2000)
self.__perform_reconfigure_test("max_sessions_per_shard", 5)
self.__perform_reconfigure_test("ping_interval", 50)
self.__perform_reconfigure_test("getmode",
oracledb.SPOOL_ATTRVAL_NOWAIT)
def test_2417_setting_each_pool_param(self):
"2417 - test to see if specified parameters are set during creation"
pool = test_env.get_pool(min=1, max=2, increment=1, timeout=10,
wait_timeout=10, max_lifetime_session=20,
max_sessions_per_shard=1, stmtcachesize=25,
ping_interval=25,
getmode=oracledb.SPOOL_ATTRVAL_WAIT)
self.assertEqual(pool.min, 1, "min (1)")
self.assertEqual(pool.max, 2, "max (2)")
self.assertEqual(pool.increment, 1, "increment (1)")
self.assertEqual(pool.getmode, oracledb.SPOOL_ATTRVAL_NOWAIT,
self.assertEqual(pool.getmode, oracledb.SPOOL_ATTRVAL_WAIT,
"getmode differs")
self.assertEqual(pool.timeout, 0, "timeout (0)")
self.assertEqual(pool.wait_timeout, 5000, "wait_timeout (5000)")
self.assertEqual(pool.max_lifetime_session, 0,
"max_lifetime_sessionmeout (0)")
self.assertEqual(pool.max_sessions_per_shard, 0,
"max_sessions_per_shard (0)")
self.assertEqual(pool.stmtcachesize, 20, "stmtcachesize (20)")
self.assertEqual(pool.ping_interval, 60, "ping_interval (60)")
pool.reconfigure(min=2, max=5, increment=2)
self.assertEqual(pool.min, 2, "min (2)")
self.assertEqual(pool.max, 5, "max (5)")
self.assertEqual(pool.increment, 2, "increment (2)")
self.assertEqual(pool.getmode, oracledb.SPOOL_ATTRVAL_NOWAIT,
"getmode differs")
self.assertEqual(pool.timeout, 0, "timeout (0)")
self.assertEqual(pool.wait_timeout, 5000, "wait_timeout (5000)")
self.assertEqual(pool.max_lifetime_session, 0,
"max_lifetime_sessionmeout (0)")
self.assertEqual(pool.max_sessions_per_shard, 0,
"max_sessions_per_shard (0)")
self.assertEqual(pool.stmtcachesize, 20, "stmtcachesize (20)")
self.assertEqual(pool.ping_interval, 60, "ping_interval (60)")
self.assertEqual(pool.timeout, 10, "timeout (10)")
self.assertEqual(pool.wait_timeout, 10, "wait_timeout (10)")
self.assertEqual(pool.max_lifetime_session, 20,
"max_lifetime_sessionmeout (20)")
self.assertEqual(pool.max_sessions_per_shard, 1,
"max_sessions_per_shard (1)")
self.assertEqual(pool.stmtcachesize, 25, "stmtcachesize (25)")
self.assertEqual(pool.ping_interval, 25, "ping_interval (25)")
if __name__ == "__main__":
test_env.run_test_cases()

17
test/test_2500_string_var.py
View File

@ -36,6 +36,12 @@ class TestCase(test_env.BaseTestCase):
self.raw_data.append(data_tuple)
self.data_by_key[i] = data_tuple
def __return_strings_as_bytes(self, cursor, name, default_type, size,
precision, scale):
if default_type == oracledb.DB_TYPE_VARCHAR:
return cursor.var(str, arraysize=cursor.arraysize,
bypass_decode=True)
def test_2500_array_with_increased_size(self):
"2500 - test creating array var and then increasing the internal size"
val = ["12345678901234567890"] * 3
@ -440,5 +446,16 @@ class TestCase(test_env.BaseTestCase):
self.cursor.execute(sql, start_value=8, end_value=10)
self.assertEqual(self.cursor.fetchall(), self.raw_data[7:10])
def test_2533_bypass_decode(self):
"2533 - test bypass string decode"
self.cursor.execute("truncate table TestTempTable")
string_val = "I bought a cafetière on the Champs-Élysées"
sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
self.cursor.execute(sql, (1, string_val))
self.cursor.outputtypehandler = self.__return_strings_as_bytes
self.cursor.execute("select IntCol, StringCol from TestTempTable")
expected_value = (1, string_val.encode())
self.assertEqual(self.cursor.fetchone(), expected_value)
if __name__ == "__main__":
test_env.run_test_cases()