test_7400_tpc_async.py

# -----------------------------------------------------------------------------
# Copyright (c) 2024, 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.
# -----------------------------------------------------------------------------

"""
7400 - Module for testing TPC (two-phase commit) transactions with asyncio.
"""

importunittest

importoracledb
importtest_env


@unittest.skipUnless(
test_env.get_is_thin(), "asyncio not supported in thick mode"
)
classTestCase(test_env.BaseAsyncTestCase):
asyncdeftest_7400(self):
"7400 - test begin, prepare, roll back global transaction"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(3900, b"txn3900", b"branchId")
awaitself.conn.tpc_begin(xid)
self.assertEqual(awaitself.conn.tpc_prepare(), False)
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (1, 'tesName')
 """
 )
self.assertEqual(awaitself.conn.tpc_prepare(), True)
awaitself.conn.tpc_rollback()
awaitself.cursor.execute("select count(*) from TestTempTable")
 (count,) =awaitself.cursor.fetchone()
self.assertEqual(count, 0)

asyncdeftest_7401(self):
"7401 - test begin, prepare, commit global transaction"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(3901, "txn3901", "branchId")
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (1, 'tesName')
 """
 )
self.assertEqual(awaitself.conn.tpc_prepare(), True)
awaitself.conn.tpc_commit()
awaitself.cursor.execute(
"select IntCol, StringCol1 from TestTempTable"
 )
self.assertEqual(awaitself.cursor.fetchall(), [(1, "tesName")])

asyncdeftest_7402(self):
"7402 - test multiple global transactions on the same connection"
awaitself.cursor.execute("truncate table TestTempTable")
xid1=self.conn.xid(3902, "txn3902", "branch1")
xid2=self.conn.xid(3902, b"txn3902", b"branch2")
awaitself.conn.tpc_begin(xid1)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (1, 'tesName')
 """
 )
awaitself.conn.tpc_end()
awaitself.conn.tpc_begin(xid2)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (2, 'tesName')
 """
 )
awaitself.conn.tpc_end()
needs_commit1=awaitself.conn.tpc_prepare(xid1)
needs_commit2=awaitself.conn.tpc_prepare(xid2)
ifneeds_commit1:
awaitself.conn.tpc_commit(xid1)
ifneeds_commit2:
awaitself.conn.tpc_commit(xid2)
awaitself.cursor.execute(
"select IntCol, StringCol1 from TestTempTable order by IntCol"
 )
expected_rows= [(1, "tesName"), (2, "tesName")]
self.assertEqual(awaitself.cursor.fetchall(), expected_rows)

asyncdeftest_7403(self):
"7403 - test rollback with parameter xid"
awaitself.cursor.execute("truncate table TestTempTable")
xid1=self.conn.xid(3901, b"txn3901", b"branch1")
xid2=self.conn.xid(3902, "txn3902", "branch2")
forcount, xidinenumerate([xid1, xid2]):
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (:id, 'tesName')
 """,
id=count,
 )
awaitself.conn.tpc_end()
awaitself.conn.tpc_rollback(xid1)

withself.assertRaisesFullCode("ORA-24756"):
awaitself.conn.tpc_prepare(xid1)
needs_commit=awaitself.conn.tpc_prepare(xid2)
ifneeds_commit:
awaitself.conn.tpc_commit(xid2)
awaitself.cursor.execute(
"select IntCol, StringCol1 from TestTempTable order by IntCol"
 )
self.assertEqual(awaitself.cursor.fetchall(), [(1, "tesName")])

asyncdeftest_7404(self):
"7404 - test resuming a transaction"
awaitself.cursor.execute("truncate table TestTempTable")
xid1=self.conn.xid(3939, "txn3939", "branch39")
xid2=self.conn.xid(3940, "txn3940", "branch40")
values= [[xid1, (1, "User Info")], [xid2, (2, "Other User Info")]]
forxid, datainvalues:
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (:1, :2)
 """,
data,
 )
awaitself.conn.tpc_end()
forxid, datainvalues:
awaitself.conn.tpc_begin(xid, oracledb.TPC_BEGIN_RESUME)
awaitself.cursor.execute(
"select IntCol, StringCol1 from TestTempTable"
 )
 (res,) =awaitself.cursor.fetchall()
self.assertEqual(res, data)
awaitself.conn.tpc_rollback(xid)

asyncdeftest_7405(self):
"7405 - test promoting a local transaction to a tpc transaction"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(3941, "txn3941", "branch41")
values= (1, "String 1")
awaitself.cursor.execute(
"insert into TestTempTable (IntCol, StringCol1) values (:1, :2)",
values,
 )
withself.assertRaisesFullCode("ORA-24776"):
awaitself.conn.tpc_begin(xid)
awaitself.conn.tpc_begin(xid, oracledb.TPC_BEGIN_PROMOTE)
awaitself.cursor.execute(
"select IntCol, StringCol1 from TestTempTable"
 )
 (res,) =awaitself.cursor.fetchall()
self.assertEqual(res, values)
awaitself.conn.tpc_rollback(xid)

asyncdeftest_7406(self):
"7406 - test ending a transaction with parameter xid"
awaitself.cursor.execute("truncate table TestTempTable")
xid1=self.conn.xid(7406, "txn7406a", "branch3")
xid2=self.conn.xid(7406, b"txn7406b", b"branch4")
awaitself.conn.tpc_begin(xid1)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (1, 'test7406a')
 """
 )
awaitself.conn.tpc_begin(xid2)
withself.assertRaisesFullCode("ORA-24758"):
awaitself.conn.tpc_end(xid1)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (2, 'test7406b')
 """
 )
awaitself.conn.tpc_end(xid2)
withself.assertRaisesFullCode("ORA-25352"):
awaitself.conn.tpc_end(xid1)
awaitself.conn.tpc_rollback(xid1)
awaitself.conn.tpc_rollback(xid2)

asyncdeftest_7407(self):
"7407 - test tpc_recover()"
awaitself.cursor.execute("truncate table TestTempTable")
n_xids=10
foriinrange(n_xids):
xid=self.conn.xid(7407+i, f"txn7407{i}", f"branch{i}")
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (:1, 'test7407')
 """,
 [i+1],
 )
awaitself.conn.tpc_prepare(xid)
recovers=awaitself.conn.tpc_recover()
self.assertEqual(len(recovers), n_xids)

awaitself.cursor.execute("select * from DBA_PENDING_TRANSACTIONS")
self.assertEqual(awaitself.cursor.fetchall(), recovers)

forxidinrecovers:
ifxid.format_id%2==0:
awaitself.conn.tpc_commit(xid)
recovers=awaitself.conn.tpc_recover()
self.assertEqual(len(recovers), n_xids//2)

forxidinrecovers:
awaitself.conn.tpc_rollback(xid)
recovers=awaitself.conn.tpc_recover()
self.assertEqual(len(recovers), 0)

asyncdeftest_7408(self):
"7408 - test tpc_recover() with read-only transaction"
awaitself.cursor.execute("truncate table TestTempTable")
foriinrange(4):
xid=self.conn.xid(7408+i, f"txn7408{i}", f"branch{i}")
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute("select * from TestTempTable")
awaitself.conn.tpc_prepare(xid)
recovers=awaitself.conn.tpc_recover()
self.assertEqual(len(recovers), 0)

asyncdeftest_7409(self):
"7409 - test tpc_commit() with one_phase parameter"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(7409, "txn7409", "branch1")
awaitself.conn.tpc_begin(xid)
values= (1, "test7409")
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (:1, :2)
 """,
values,
 )
awaitself.cursor.execute(
"select IntCol, StringCol1 from TestTempTable"
 )
awaitself.conn.tpc_commit(xid, one_phase=True)
self.assertEqual(awaitself.cursor.fetchall(), [values])

asyncdeftest_7410(self):
"7410 - test negative cases for tpc_commit()"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(7410, "txn7410", "branch1")
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute(
"""
 insert into TestTempTable (IntCol, StringCol1)
 values (1, 'test7410')
 """
 )
withself.assertRaises(TypeError):
awaitself.conn.tpc_commit("invalid xid")
awaitself.conn.tpc_prepare(xid)
withself.assertRaisesFullCode("ORA-02053"):
awaitself.conn.tpc_commit(xid, one_phase=True)
withself.assertRaisesFullCode("ORA-24756"):
awaitself.conn.tpc_commit(xid)

asyncdeftest_7411(self):
"7411 - test starting an already created transaction"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(7411, "txn7411", "branch1")
awaitself.conn.tpc_begin(xid)
withself.assertRaisesFullCode("ORA-24757"):
awaitself.conn.tpc_begin(xid, oracledb.TPC_BEGIN_NEW)
withself.assertRaisesFullCode("ORA-24797"):
awaitself.conn.tpc_begin(xid, oracledb.TPC_BEGIN_PROMOTE)
awaitself.conn.tpc_end()
forflagin [oracledb.TPC_BEGIN_NEW, oracledb.TPC_BEGIN_PROMOTE]:
withself.assertRaisesFullCode("ORA-24757"):
awaitself.conn.tpc_begin(xid, flag)
awaitself.conn.tpc_rollback(xid)

asyncdeftest_7412(self):
"7412 - test resuming a prepared transaction"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(7412, "txn7412", "branch1")
awaitself.conn.tpc_begin(xid)
awaitself.conn.tpc_prepare(xid)
withself.assertRaisesFullCode("ORA-24756"):
awaitself.conn.tpc_begin(xid, oracledb.TPC_BEGIN_RESUME)

asyncdeftest_7413(self):
"7413 - test tpc_begin and tpc_end with invalid parameters"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(7413, "txn7413", "branch1")
test_values= [
 (self.conn.tpc_begin, "DPY-2050"),
 (self.conn.tpc_end, "DPY-2051"),
 ]
fortpc_function, error_codeintest_values:
withself.assertRaises(TypeError):
awaittpc_function("invalid xid")
withself.assertRaisesFullCode(error_code):
awaittpc_function(xid, "invalid flag")
withself.assertRaisesFullCode(error_code):
awaittpc_function(xid, 70)

asyncdeftest_7414(self):
"7414 - test commiting transaction without tpc_commit"
xid=self.conn.xid(7414, "txn7409", "branch1")
awaitself.conn.tpc_begin(xid)
withself.assertRaisesFullCode("ORA-02089"):
awaitself.cursor.execute("truncate table TestTempTable")

asyncdeftest_7415(self):
"7415 - test tpc_commit when a commit is not needed"
xid=self.conn.xid(7416, "txn7416", "branch1")
awaitself.conn.tpc_begin(xid)
awaitself.cursor.execute("select * from TestTempTable")
awaitself.conn.tpc_end(xid)
awaitself.conn.tpc_prepare(xid)
withself.assertRaisesFullCode("ORA-24756"):
awaitself.conn.tpc_commit(xid)

asyncdeftest_7416(self):
"7416 - test transaction_in_progress"
awaitself.cursor.execute("truncate table TestTempTable")
xid=self.conn.xid(7415, "txn7415", "branch1")
self.assertFalse(self.conn.transaction_in_progress)

awaitself.conn.tpc_begin(xid)
self.assertTrue(self.conn.transaction_in_progress)
awaitself.cursor.execute(
"insert into TestTempTable (IntCol) values (2)"
 )

awaitself.conn.tpc_end(xid)
self.assertFalse(self.conn.transaction_in_progress)

awaitself.conn.tpc_prepare(xid)
self.assertFalse(self.conn.transaction_in_progress)

awaitself.conn.tpc_commit(xid)
self.assertFalse(self.conn.transaction_in_progress)


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