Add Testcase demonstrating fail on join

When a join is attempted while the transaction is in COMMITFAILED
zope.sqlalchemy is left in an inconsistent state.

See #2

src/zope/sqlalchemy/tests.py

@@ -49,6 +49,9 @@ def b(s):
 import threading
 import time
 
+from transaction._transaction import Status as ZopeStatus
+from transaction.interfaces import TransactionFailedError
+
 import sqlalchemy as sa
 from sqlalchemy import orm, sql, exc
 from zope.sqlalchemy import datamanager as tx
@@ -540,6 +543,38 @@ def testBulkUpdate(self):
         results = engine.connect().execute(test_users.select(test_users.c.lastname == "smith"))
         self.assertEqual(len(results.fetchall()), 2)
 
+    def testFailedJoin(self):
+        # When a join is issued while the transaction is in COMMITFAILED, the
+        # session is never closed and the session id stays in _SESSION_STATE,
+        # which means the session won't be joined in the future either. This
+        # causes the session to stay open forever, potentially accumulating
+        # data, but never issuing a commit.
+        dummy = DummyDataManager(key='dummy.first')
+        transaction.get().join(dummy)
+        try:
+            transaction.commit()
+        except DummyException:
+            # Commit raised an error, we are now in COMMITFAILED
+            pass
+        self.assertEqual(transaction.get().status, ZopeStatus.COMMITFAILED)
+
+        session = Session()
+        # try to interact with the session while the transaction is still
+        # in COMMITFAILED
+        self.assertRaises(TransactionFailedError,
+                          session.query(User).all)
+        transaction.abort()
+
+        # start a new transaction everything should be ok now
+        transaction.begin()
+        session = Session()
+        self.assertEqual([], session.query(User).all())
+        session.add(User(id=1, firstname='udo', lastname='juergens'))
+
+        # abort transaction, session should be closed without commit
+        transaction.abort()
+        self.assertEqual([], session.query(User).all())
+
 
 class RetryTests(unittest.TestCase):