mirror of
https://github.com/psycopg/psycopg2.git
synced 2025-11-04 09:47:30 +03:00
8bfd34faf2
By James Henstridge on 2008-07-24. Merged from lp:~jamesh/psycopg/two-phase-commit/revision/359
145 lines
4.0 KiB
Python
145 lines
4.0 KiB
Python
""" 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, "%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.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()
|