beginnings of a TPC test harness

By James Henstridge on 2008-05-12.

Merged from lp:~jamesh/psycopg/two-phase-commit/revision/354

ChangeLog

@@ -8,6 +8,14 @@
 	* psycopg/connection_int.c: Fixed access to freed memory in
 	conn_get_isolation_level(). Bug reported by Anton Kovalev.
 
+2010-10-06  Daniele Varrazzo  <daniele.varrazzo@gmail.com>
+
+	* Merged James Henstridge work on two-phase commit support.
+
+	2008-05-12  James Henstridge  <james@jamesh.id.au>
+
+		* beginnings of a TPC test harness
+
 2010-10-05  Daniele Varrazzo  <daniele.varrazzo@gmail.com>
 
 	* psycopg/cursor_type.c: Common code in execute() and mogrify() merged.

tests/dbapi20_tpc.py

@@ -0,0 +1,138 @@
+""" Python DB API 2.0 driver Two Phase Commit compliance test suite.
+
+"""
+
+import unittest
+
+
+class TwoPhaseCommitTest(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 = TwoPhaseCommitTest._last_id
+        TwoPhaseCommitTest._last_id += 1
+        return con.xid(42, "%s%d" % (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.assertEuqals(xid[0], 42)
+        self.assertEuqals(xid[1], "global")
+        self.assertEquals(xid[2], "bqual")
+
+    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()