Repaired the usage of ``.prepare()`` in conjunction with

cx_Oracle so that a return value of ``False`` will result
in no call to ``connection.commit()``, hence avoiding
"no transaction" errors.   Two-phase transactions have
now been shown to work in a rudimental fashion with
SQLAlchemy and cx_oracle, however are subject to caveats
observed with the driver; check the documentation
for details.  Also in 0.7.10.
[ticket:2611]

1 files changed, 70 insertions, 0 deletions

diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py
index 3e7ebf012..7604bf928 100644
--- a/test/dialect/test_oracle.py
+++ b/test/dialect/test_oracle.py
@@ -781,6 +781,76 @@ class ConstraintTest(fixtures.TestBase):
                     onupdate='CASCADE'))
         assert_raises(exc.SAWarning, bat.create)
 
+
+class TwoPhaseTest(fixtures.TablesTest):
+    """test cx_oracle two phase, which remains in a semi-broken state
+    so requires a carefully written test."""
+
+    __only_on__ = 'oracle+cx_oracle'
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('datatable', metadata,
+                Column('id', Integer, primary_key=True),
+                Column('data', String(50))
+            )
+
+    def _connection(self):
+        conn = testing.db.connect()
+        conn.detach()
+        return conn
+
+    def _assert_data(self, rows):
+        eq_(
+            testing.db.scalar("select count(*) from datatable"),
+            rows
+        )
+    def test_twophase_prepare_false(self):
+        conn = self._connection()
+        for i in xrange(2):
+            trans = conn.begin_twophase()
+            conn.execute("select 1 from dual")
+            trans.prepare()
+            trans.commit()
+        conn.close()
+        self._assert_data(0)
+
+    def test_twophase_prepare_true(self):
+        conn = self._connection()
+        for i in xrange(2):
+            trans = conn.begin_twophase()
+            conn.execute("insert into datatable (id, data) "
+                    "values (%s, 'somedata')" % i)
+            trans.prepare()
+            trans.commit()
+        conn.close()
+        self._assert_data(2)
+
+    def test_twophase_rollback(self):
+        conn = self._connection()
+        trans = conn.begin_twophase()
+        conn.execute("insert into datatable (id, data) "
+                "values (%s, 'somedata')" % 1)
+        trans.rollback()
+
+        trans = conn.begin_twophase()
+        conn.execute("insert into datatable (id, data) "
+                "values (%s, 'somedata')" % 1)
+        trans.prepare()
+        trans.commit()
+
+        conn.close()
+        self._assert_data(1)
+
+    def test_not_prepared(self):
+        conn = self._connection()
+        trans = conn.begin_twophase()
+        conn.execute("insert into datatable (id, data) "
+                "values (%s, 'somedata')" % 1)
+        trans.commit()
+        conn.close()
+        self._assert_data(1)
+
 class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = oracle.OracleDialect()