Drop in dbapi tests for two-phase commit support

diff --git a/tests/dbapi20_tpc.py b/tests/dbapi20_tpc.py
new file mode 100644 (file)
index 0000000..d687032
--- /dev/null
+++ b/tests/dbapi20_tpc.py
@@ -0,0 +1,147 @@
+# flake8: noqa
+# fmt: off
+
+""" Python DB API 2.0 driver Two Phase Commit compliance test suite.
+
+"""
+
+import unittest
+
+
+class TwoPhaseCommitTests(unittest.TestCase):
+
+    driver = None
+
+    def connect(self):
+        """Make a database connection."""
+        raise NotImplementedError
+
+    _last_id = 0
+    _global_id_prefix = "dbapi20_tpc:"
+
+    def make_xid(self, con):
+        id = TwoPhaseCommitTests._last_id
+        TwoPhaseCommitTests._last_id += 1
+        return con.xid(42, f"{self._global_id_prefix}{id}", "qualifier")
+
+    def test_xid(self):
+        con = self.connect()
+        try:
+            xid = con.xid(42, "global", "bqual")
+        except self.driver.NotSupportedError:
+            self.fail("Driver does not support transaction IDs.")
+
+        self.assertEquals(xid[0], 42)
+        self.assertEquals(xid[1], "global")
+        self.assertEquals(xid[2], "bqual")
+
+        # Try some extremes for the transaction ID:
+        xid = con.xid(0, "", "")
+        self.assertEquals(tuple(xid), (0, "", ""))
+        xid = con.xid(0x7fffffff, "a" * 64, "b" * 64)
+        self.assertEquals(tuple(xid), (0x7fffffff, "a" * 64, "b" * 64))
+
+    def test_tpc_begin(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            try:
+                con.tpc_begin(xid)
+            except self.driver.NotSupportedError:
+                self.fail("Driver does not support tpc_begin()")
+        finally:
+            con.close()
+
+    def test_tpc_commit_without_prepare(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            con.tpc_begin(xid)
+            cursor = con.cursor()
+            cursor.execute("SELECT 1")
+            con.tpc_commit()
+        finally:
+            con.close()
+
+    def test_tpc_rollback_without_prepare(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            con.tpc_begin(xid)
+            cursor = con.cursor()
+            cursor.execute("SELECT 1")
+            con.tpc_rollback()
+        finally:
+            con.close()
+
+    def test_tpc_commit_with_prepare(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            con.tpc_begin(xid)
+            cursor = con.cursor()
+            cursor.execute("SELECT 1")
+            con.tpc_prepare()
+            con.tpc_commit()
+        finally:
+            con.close()
+
+    def test_tpc_rollback_with_prepare(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            con.tpc_begin(xid)
+            cursor = con.cursor()
+            cursor.execute("SELECT 1")
+            con.tpc_prepare()
+            con.tpc_rollback()
+        finally:
+            con.close()
+
+    def test_tpc_begin_in_transaction_fails(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+
+            cursor = con.cursor()
+            cursor.execute("SELECT 1")
+            self.assertRaises(self.driver.ProgrammingError,
+                              con.tpc_begin, xid)
+        finally:
+            con.close()
+
+    def test_tpc_begin_in_tpc_transaction_fails(self):
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+
+            cursor = con.cursor()
+            cursor.execute("SELECT 1")
+            self.assertRaises(self.driver.ProgrammingError,
+                              con.tpc_begin, xid)
+        finally:
+            con.close()
+
+    def test_commit_in_tpc_fails(self):
+        # calling commit() within a TPC transaction fails with
+        # ProgrammingError.
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            con.tpc_begin(xid)
+
+            self.assertRaises(self.driver.ProgrammingError, con.commit)
+        finally:
+            con.close()
+
+    def test_rollback_in_tpc_fails(self):
+        # calling rollback() within a TPC transaction fails with
+        # ProgrammingError.
+        con = self.connect()
+        try:
+            xid = self.make_xid(con)
+            con.tpc_begin(xid)
+
+            self.assertRaises(self.driver.ProgrammingError, con.rollback)
+        finally:
+            con.close()

diff --git a/tests/test_psycopg_dbapi20.py b/tests/test_psycopg_dbapi20.py
index 3d284bcd80b22cbc2a92ebe7b031de603311501b..04efa1ecb4ff404b1f923dd3405756e8c9554158 100644 (file)
--- a/tests/test_psycopg_dbapi20.py
+++ b/tests/test_psycopg_dbapi20.py
@@ -6,6 +6,7 @@ import psycopg
 from psycopg.conninfo import conninfo_to_dict
 
 from . import dbapi20
+from . import dbapi20_tpc
 
 
 @pytest.fixture(scope="class")
@@ -28,8 +29,19 @@ class PsycopgTests(dbapi20.DatabaseAPI20Test):
         pass
 
 
+# @skip_if_tpc_disabled
+@pytest.mark.usefixtures("with_dsn")
+class PsycopgTPCTests(dbapi20_tpc.TwoPhaseCommitTests):
+    driver = psycopg
+    # connect_args = () # set by the fixture
+
+    def connect(self):
+        return psycopg.connect(*self.connect_args)
+
+
 # Shut up warnings
 PsycopgTests.failUnless = PsycopgTests.assertTrue  # type: ignore[assignment]
+PsycopgTPCTests.assertEquals = PsycopgTPCTests.assertEqual
 
 
 @pytest.mark.parametrize(