from sqlalchemy.test.testing import eq_, assert_raises, \
    assert_raises_message
import sys
import time
import threading
from sqlalchemy import create_engine, MetaData, INT, VARCHAR, Sequence, \
    select, Integer, String, func, text, exc
from sqlalchemy.test.schema import Table
from sqlalchemy.test.schema import Column
from sqlalchemy.test import TestBase, testing


users, metadata = None, None
class TransactionTest(TestBase):
    @classmethod
    def setup_class(cls):
        global users, metadata
        metadata = MetaData()
        users = Table('query_users', metadata,
            Column('user_id', INT, primary_key = True),
            Column('user_name', VARCHAR(20)),
            test_needs_acid=True,
        )
        users.create(testing.db)

    def teardown(self):
        testing.db.execute(users.delete()).close()

    @classmethod
    def teardown_class(cls):
        users.drop(testing.db)

    def test_commits(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        transaction.commit()

        transaction = connection.begin()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction.commit()

        transaction = connection.begin()
        result = connection.execute("select * from query_users")
        assert len(result.fetchall()) == 3
        transaction.commit()
        connection.close()

    def test_rollback(self):
        """test a basic rollback"""
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction.rollback()

        result = connection.execute("select * from query_users")
        assert len(result.fetchall()) == 0
        connection.close()

    def test_raise(self):
        connection = testing.db.connect()

        transaction = connection.begin()
        try:
            connection.execute(users.insert(), user_id=1, user_name='user1')
            connection.execute(users.insert(), user_id=2, user_name='user2')
            connection.execute(users.insert(), user_id=1, user_name='user3')
            transaction.commit()
            assert False
        except Exception , e:
            print "Exception: ", e
            transaction.rollback()

        result = connection.execute("select * from query_users")
        assert len(result.fetchall()) == 0
        connection.close()
    
    def test_transaction_container(self):

        def go(conn, table, data):
            for d in data:
                conn.execute(table.insert(), d)

        testing.db.transaction(go, users, [dict(user_id=1,
                               user_name='user1')])
        eq_(testing.db.execute(users.select()).fetchall(), [(1, 'user1'
            )])
        assert_raises(exc.DBAPIError, testing.db.transaction, go,
                      users, [{'user_id': 2, 'user_name': 'user2'},
                      {'user_id': 1, 'user_name': 'user3'}])
        eq_(testing.db.execute(users.select()).fetchall(), [(1, 'user1'
            )])
        
    def test_nested_rollback(self):
        connection = testing.db.connect()
        try:
            transaction = connection.begin()
            try:
                connection.execute(users.insert(), user_id=1,
                                   user_name='user1')
                connection.execute(users.insert(), user_id=2,
                                   user_name='user2')
                connection.execute(users.insert(), user_id=3,
                                   user_name='user3')
                trans2 = connection.begin()
                try:
                    connection.execute(users.insert(), user_id=4,
                            user_name='user4')
                    connection.execute(users.insert(), user_id=5,
                            user_name='user5')
                    raise Exception('uh oh')
                    trans2.commit()
                except:
                    trans2.rollback()
                    raise
                transaction.rollback()
            except Exception, e:
                transaction.rollback()
                raise
        except Exception, e:
            try:
                assert str(e) == 'uh oh'  # and not "This transaction is
                                          # inactive"
            finally:
                connection.close()

    def test_retains_through_options(self):
        connection = testing.db.connect()
        try:
            transaction = connection.begin()
            connection.execute(users.insert(), user_id=1, user_name='user1')
            conn2 = connection.execution_options(dummy=True)
            conn2.execute(users.insert(), user_id=2, user_name='user2')
            transaction.rollback()
            eq_(connection.scalar("select count(1) from query_users"), 0)
        finally:
            connection.close()
        
    def test_nesting(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        trans2 = connection.begin()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        connection.execute(users.insert(), user_id=5, user_name='user5')
        trans2.commit()
        transaction.rollback()
        self.assert_(connection.scalar('select count(1) from '
                     'query_users') == 0)
        result = connection.execute('select * from query_users')
        assert len(result.fetchall()) == 0
        connection.close()

    def test_close(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        trans2 = connection.begin()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        connection.execute(users.insert(), user_id=5, user_name='user5')
        assert connection.in_transaction()
        trans2.close()
        assert connection.in_transaction()
        transaction.commit()
        assert not connection.in_transaction()
        self.assert_(connection.scalar('select count(1) from '
                     'query_users') == 5)
        result = connection.execute('select * from query_users')
        assert len(result.fetchall()) == 5
        connection.close()

    def test_close2(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        trans2 = connection.begin()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        connection.execute(users.insert(), user_id=5, user_name='user5')
        assert connection.in_transaction()
        trans2.close()
        assert connection.in_transaction()
        transaction.close()
        assert not connection.in_transaction()
        self.assert_(connection.scalar('select count(1) from '
                     'query_users') == 0)
        result = connection.execute('select * from query_users')
        assert len(result.fetchall()) == 0
        connection.close()

    @testing.requires.savepoints
    def test_nested_subtransaction_rollback(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        trans2 = connection.begin_nested()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        trans2.rollback()
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction.commit()
        eq_(connection.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (3, )])
        connection.close()

    @testing.requires.savepoints
    @testing.crashes('oracle+zxjdbc',
                     'Errors out and causes subsequent tests to '
                     'deadlock')
    def test_nested_subtransaction_commit(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        trans2 = connection.begin_nested()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        trans2.commit()
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction.commit()
        eq_(connection.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (2, ), (3, )])
        connection.close()

    @testing.requires.savepoints
    def test_rollback_to_subtransaction(self):
        connection = testing.db.connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        trans2 = connection.begin_nested()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        trans3 = connection.begin()
        connection.execute(users.insert(), user_id=3, user_name='user3')
        trans3.rollback()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        transaction.commit()
        eq_(connection.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (4, )])
        connection.close()

    @testing.requires.two_phase_transactions
    def test_two_phase_transaction(self):
        connection = testing.db.connect()
        transaction = connection.begin_twophase()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        transaction.prepare()
        transaction.commit()
        transaction = connection.begin_twophase()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        transaction.commit()
        transaction.close()
        transaction = connection.begin_twophase()
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction.rollback()
        transaction = connection.begin_twophase()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        transaction.prepare()
        transaction.rollback()
        transaction.close()
        eq_(connection.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (2, )])
        connection.close()

    @testing.requires.two_phase_transactions
    @testing.requires.savepoints
    def test_mixed_two_phase_transaction(self):
        connection = testing.db.connect()
        transaction = connection.begin_twophase()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        transaction2 = connection.begin()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        transaction3 = connection.begin_nested()
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction4 = connection.begin()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        transaction4.commit()
        transaction3.rollback()
        connection.execute(users.insert(), user_id=5, user_name='user5')
        transaction2.commit()
        transaction.prepare()
        transaction.commit()
        eq_(connection.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (2, ), (5, )])
        connection.close()

    @testing.requires.two_phase_transactions
    @testing.crashes('mysql+oursql',
                     'Times out in full test runs only, causing '
                     'subsequent tests to fail')
    @testing.crashes('mysql+zxjdbc',
                     'Deadlocks, causing subsequent tests to fail')
    @testing.fails_on('mysql', 'FIXME: unknown')
    def test_two_phase_recover(self):

        # MySQL recovery doesn't currently seem to work correctly
        # Prepared transactions disappear when connections are closed
        # and even when they aren't it doesn't seem possible to use the
        # recovery id.

        connection = testing.db.connect()
        transaction = connection.begin_twophase()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        transaction.prepare()
        connection.close()
        connection2 = testing.db.connect()
        eq_(connection2.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [])
        recoverables = connection2.recover_twophase()
        assert transaction.xid in recoverables
        connection2.commit_prepared(transaction.xid, recover=True)
        eq_(connection2.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, )])
        connection2.close()

    @testing.requires.two_phase_transactions
    def test_multiple_two_phase(self):
        conn = testing.db.connect()
        xa = conn.begin_twophase()
        conn.execute(users.insert(), user_id=1, user_name='user1')
        xa.prepare()
        xa.commit()
        xa = conn.begin_twophase()
        conn.execute(users.insert(), user_id=2, user_name='user2')
        xa.prepare()
        xa.rollback()
        xa = conn.begin_twophase()
        conn.execute(users.insert(), user_id=3, user_name='user3')
        xa.rollback()
        xa = conn.begin_twophase()
        conn.execute(users.insert(), user_id=4, user_name='user4')
        xa.prepare()
        xa.commit()
        result = \
            conn.execute(select([users.c.user_name]).
                order_by(users.c.user_id))
        eq_(result.fetchall(), [('user1', ), ('user4', )])
        conn.close()

class AutoRollbackTest(TestBase):

    @classmethod
    def setup_class(cls):
        global metadata
        metadata = MetaData()

    @classmethod
    def teardown_class(cls):
        metadata.drop_all(testing.db)

    def test_rollback_deadlock(self):
        """test that returning connections to the pool clears any object
        locks."""

        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        users = Table('deadlock_users', metadata, Column('user_id',
                      INT, primary_key=True), Column('user_name',
                      VARCHAR(20)), test_needs_acid=True)
        users.create(conn1)
        conn1.execute('select * from deadlock_users')
        conn1.close()

        # without auto-rollback in the connection pool's return() logic,
        # this deadlocks in PostgreSQL, because conn1 is returned to the
        # pool but still has a lock on "deadlock_users". comment out the
        # rollback in pool/ConnectionFairy._close() to see !

        users.drop(conn2)
        conn2.close()

class ExplicitAutoCommitTest(TestBase):

    """test the 'autocommit' flag on select() and text() objects.
    
    Requires PostgreSQL so that we may define a custom function which
    modifies the database. """

    __only_on__ = 'postgresql'

    @classmethod
    def setup_class(cls):
        global metadata, foo
        metadata = MetaData(testing.db)
        foo = Table('foo', metadata, Column('id', Integer,
                    primary_key=True), Column('data', String(100)))
        metadata.create_all()
        testing.db.execute("create function insert_foo(varchar) "
                           "returns integer as 'insert into foo(data) "
                           "values ($1);select 1;' language sql")

    def teardown(self):
        foo.delete().execute().close()

    @classmethod
    def teardown_class(cls):
        testing.db.execute('drop function insert_foo(varchar)')
        metadata.drop_all()

    def test_control(self):

        # test that not using autocommit does not commit

        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execute(select([func.insert_foo('data1')]))
        assert conn2.execute(select([foo.c.data])).fetchall() == []
        conn1.execute(text("select insert_foo('moredata')"))
        assert conn2.execute(select([foo.c.data])).fetchall() == []
        trans = conn1.begin()
        trans.commit()
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('data1', ), ('moredata', )]
        conn1.close()
        conn2.close()

    def test_explicit_compiled(self):
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execute(select([func.insert_foo('data1'
                      )]).execution_options(autocommit=True))
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('data1', )]
        conn1.close()
        conn2.close()

    def test_explicit_connection(self):
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execution_options(autocommit=True).\
            execute(select([func.insert_foo('data1'
                )]))
        eq_(conn2.execute(select([foo.c.data])).fetchall(), [('data1',
            )])

        # connection supercedes statement

        conn1.execution_options(autocommit=False).\
            execute(select([func.insert_foo('data2'
                )]).execution_options(autocommit=True))
        eq_(conn2.execute(select([foo.c.data])).fetchall(), [('data1',
            )])

        # ditto

        conn1.execution_options(autocommit=True).\
            execute(select([func.insert_foo('data3'
                )]).execution_options(autocommit=False))
        eq_(conn2.execute(select([foo.c.data])).fetchall(), [('data1',
            ), ('data2', ), ('data3', )])
        conn1.close()
        conn2.close()

    def test_explicit_text(self):
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execute(text("select insert_foo('moredata')"
                      ).execution_options(autocommit=True))
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('moredata', )]
        conn1.close()
        conn2.close()

    @testing.uses_deprecated(r'autocommit on select\(\) is deprecated',
                             r'autocommit\(\) is deprecated')
    def test_explicit_compiled_deprecated(self):
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execute(select([func.insert_foo('data1')],
                      autocommit=True))
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('data1', )]
        conn1.execute(select([func.insert_foo('data2')]).autocommit())
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('data1', ), ('data2', )]
        conn1.close()
        conn2.close()

    @testing.uses_deprecated(r'autocommit on text\(\) is deprecated')
    def test_explicit_text_deprecated(self):
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execute(text("select insert_foo('moredata')",
                      autocommit=True))
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('moredata', )]
        conn1.close()
        conn2.close()

    def test_implicit_text(self):
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        conn1.execute(text("insert into foo (data) values "
                      "('implicitdata')"))
        assert conn2.execute(select([foo.c.data])).fetchall() \
            == [('implicitdata', )]
        conn1.close()
        conn2.close()


tlengine = None


class TLTransactionTest(TestBase):

    @classmethod
    def setup_class(cls):
        global users, metadata, tlengine
        tlengine = create_engine(testing.db.url, strategy='threadlocal')
        metadata = MetaData()
        users = Table('query_users', metadata, Column('user_id', INT,
                      Sequence('query_users_id_seq', optional=True),
                      primary_key=True), Column('user_name',
                      VARCHAR(20)), test_needs_acid=True)
        metadata.create_all(tlengine)

    def teardown(self):
        tlengine.execute(users.delete()).close()

    @classmethod
    def teardown_class(cls):
        metadata.drop_all(tlengine)
        tlengine.dispose()

    def setup(self):

        # ensure tests start with engine closed

        tlengine.close()

    def test_connection_close(self):
        """test that when connections are closed for real, transactions
        are rolled back and disposed."""

        c = tlengine.contextual_connect()
        c.begin()
        assert c.in_transaction()
        c.close()
        assert not c.in_transaction()

    def test_transaction_close(self):
        c = tlengine.contextual_connect()
        t = c.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        t2 = c.begin()
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.execute(users.insert(), user_id=4, user_name='user4')
        t2.close()
        result = c.execute('select * from query_users')
        assert len(result.fetchall()) == 4
        t.close()
        external_connection = tlengine.connect()
        result = external_connection.execute('select * from query_users'
                )
        try:
            assert len(result.fetchall()) == 0
        finally:
            c.close()
            external_connection.close()

    def test_rollback(self):
        """test a basic rollback"""

        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.rollback()
        external_connection = tlengine.connect()
        result = external_connection.execute('select * from query_users'
                )
        try:
            assert len(result.fetchall()) == 0
        finally:
            external_connection.close()

    def test_commit(self):
        """test a basic commit"""

        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.commit()
        external_connection = tlengine.connect()
        result = external_connection.execute('select * from query_users'
                )
        try:
            assert len(result.fetchall()) == 3
        finally:
            external_connection.close()

    def test_commits(self):
        connection = tlengine.connect()
        assert connection.execute('select count(1) from query_users'
                                  ).scalar() == 0
        connection.close()
        connection = tlengine.contextual_connect()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        transaction.commit()
        transaction = connection.begin()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        transaction.commit()
        transaction = connection.begin()
        result = connection.execute('select * from query_users')
        l = result.fetchall()
        assert len(l) == 3, 'expected 3 got %d' % len(l)
        transaction.commit()
        connection.close()

    def test_rollback_off_conn(self):

        # test that a TLTransaction opened off a TLConnection allows
        # that TLConnection to be aware of the transactional context

        conn = tlengine.contextual_connect()
        trans = conn.begin()
        conn.execute(users.insert(), user_id=1, user_name='user1')
        conn.execute(users.insert(), user_id=2, user_name='user2')
        conn.execute(users.insert(), user_id=3, user_name='user3')
        trans.rollback()
        external_connection = tlengine.connect()
        result = external_connection.execute('select * from query_users'
                )
        try:
            assert len(result.fetchall()) == 0
        finally:
            conn.close()
            external_connection.close()

    def test_morerollback_off_conn(self):

        # test that an existing TLConnection automatically takes place
        # in a TLTransaction opened on a second TLConnection

        conn = tlengine.contextual_connect()
        conn2 = tlengine.contextual_connect()
        trans = conn2.begin()
        conn.execute(users.insert(), user_id=1, user_name='user1')
        conn.execute(users.insert(), user_id=2, user_name='user2')
        conn.execute(users.insert(), user_id=3, user_name='user3')
        trans.rollback()
        external_connection = tlengine.connect()
        result = external_connection.execute('select * from query_users'
                )
        try:
            assert len(result.fetchall()) == 0
        finally:
            conn.close()
            conn2.close()
            external_connection.close()

    def test_commit_off_connection(self):
        conn = tlengine.contextual_connect()
        trans = conn.begin()
        conn.execute(users.insert(), user_id=1, user_name='user1')
        conn.execute(users.insert(), user_id=2, user_name='user2')
        conn.execute(users.insert(), user_id=3, user_name='user3')
        trans.commit()
        external_connection = tlengine.connect()
        result = external_connection.execute('select * from query_users'
                )
        try:
            assert len(result.fetchall()) == 3
        finally:
            conn.close()
            external_connection.close()

    def test_nesting_rollback(self):
        """tests nesting of transactions, rollback at the end"""

        external_connection = tlengine.connect()
        self.assert_(external_connection.connection
                     is not tlengine.contextual_connect().connection)
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=4, user_name='user4')
        tlengine.execute(users.insert(), user_id=5, user_name='user5')
        tlengine.commit()
        tlengine.rollback()
        try:
            self.assert_(external_connection.scalar(
                        'select count(1) from query_users'
                         ) == 0)
        finally:
            external_connection.close()

    def test_nesting_commit(self):
        """tests nesting of transactions, commit at the end."""

        external_connection = tlengine.connect()
        self.assert_(external_connection.connection
                     is not tlengine.contextual_connect().connection)
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=4, user_name='user4')
        tlengine.execute(users.insert(), user_id=5, user_name='user5')
        tlengine.commit()
        tlengine.commit()
        try:
            self.assert_(external_connection.scalar(
                        'select count(1) from query_users'
                         ) == 5)
        finally:
            external_connection.close()

    def test_mixed_nesting(self):
        """tests nesting of transactions off the TLEngine directly
        inside of tranasctions off the connection from the TLEngine"""

        external_connection = tlengine.connect()
        self.assert_(external_connection.connection
                     is not tlengine.contextual_connect().connection)
        conn = tlengine.contextual_connect()
        trans = conn.begin()
        trans2 = conn.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=4, user_name='user4')
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=5, user_name='user5')
        tlengine.execute(users.insert(), user_id=6, user_name='user6')
        tlengine.execute(users.insert(), user_id=7, user_name='user7')
        tlengine.commit()
        tlengine.execute(users.insert(), user_id=8, user_name='user8')
        tlengine.commit()
        trans2.commit()
        trans.rollback()
        conn.close()
        try:
            self.assert_(external_connection.scalar(
                        'select count(1) from query_users'
                         ) == 0)
        finally:
            external_connection.close()

    def test_more_mixed_nesting(self):
        """tests nesting of transactions off the connection from the
        TLEngine inside of tranasctions off thbe TLEngine directly."""

        external_connection = tlengine.connect()
        self.assert_(external_connection.connection
                     is not tlengine.contextual_connect().connection)
        tlengine.begin()
        connection = tlengine.contextual_connect()
        connection.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.begin()
        connection.execute(users.insert(), user_id=2, user_name='user2')
        connection.execute(users.insert(), user_id=3, user_name='user3')
        trans = connection.begin()
        connection.execute(users.insert(), user_id=4, user_name='user4')
        connection.execute(users.insert(), user_id=5, user_name='user5')
        trans.commit()
        tlengine.commit()
        tlengine.rollback()
        connection.close()
        try:
            self.assert_(external_connection.scalar(
                        'select count(1) from query_users'
                         ) == 0)
        finally:
            external_connection.close()

    @testing.requires.savepoints
    def test_nested_subtransaction_rollback(self):
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.begin_nested()
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.rollback()
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.commit()
        tlengine.close()
        eq_(tlengine.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (3, )])
        tlengine.close()

    @testing.requires.savepoints
    @testing.crashes('oracle+zxjdbc',
                     'Errors out and causes subsequent tests to '
                     'deadlock')
    def test_nested_subtransaction_commit(self):
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.begin_nested()
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.commit()
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.commit()
        tlengine.close()
        eq_(tlengine.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (2, ), (3, )])
        tlengine.close()

    @testing.requires.savepoints
    def test_rollback_to_subtransaction(self):
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.begin_nested()
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.begin()
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.rollback()
        tlengine.rollback()
        tlengine.execute(users.insert(), user_id=4, user_name='user4')
        tlengine.commit()
        tlengine.close()
        eq_(tlengine.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (4, )])
        tlengine.close()

    def test_connections(self):
        """tests that contextual_connect is threadlocal"""

        c1 = tlengine.contextual_connect()
        c2 = tlengine.contextual_connect()
        assert c1.connection is c2.connection
        c2.close()
        assert not c1.closed
        assert not tlengine.closed

    @testing.requires.independent_cursors
    def test_result_closing(self):
        """tests that contextual_connect is threadlocal"""

        r1 = tlengine.execute(select([1]))
        r2 = tlengine.execute(select([1]))
        row1 = r1.fetchone()
        row2 = r2.fetchone()
        r1.close()
        assert r2.connection is r1.connection
        assert not r2.connection.closed
        assert not tlengine.closed

        # close again, nothing happens since resultproxy calls close()
        # only once

        r1.close()
        assert r2.connection is r1.connection
        assert not r2.connection.closed
        assert not tlengine.closed
        r2.close()
        assert r2.connection.closed
        assert tlengine.closed

    def test_dispose(self):
        eng = create_engine(testing.db.url, strategy='threadlocal')
        result = eng.execute(select([1]))
        eng.dispose()
        eng.execute(select([1]))

    @testing.requires.two_phase_transactions
    def test_two_phase_transaction(self):
        tlengine.begin_twophase()
        tlengine.execute(users.insert(), user_id=1, user_name='user1')
        tlengine.prepare()
        tlengine.commit()
        tlengine.begin_twophase()
        tlengine.execute(users.insert(), user_id=2, user_name='user2')
        tlengine.commit()
        tlengine.begin_twophase()
        tlengine.execute(users.insert(), user_id=3, user_name='user3')
        tlengine.rollback()
        tlengine.begin_twophase()
        tlengine.execute(users.insert(), user_id=4, user_name='user4')
        tlengine.prepare()
        tlengine.rollback()
        eq_(tlengine.execute(select([users.c.user_id]).
            order_by(users.c.user_id)).fetchall(),
            [(1, ), (2, )])

counters = None


class ForUpdateTest(TestBase):

    @classmethod
    def setup_class(cls):
        global counters, metadata
        metadata = MetaData()
        counters = Table('forupdate_counters', metadata,
                         Column('counter_id', INT, primary_key=True),
                         Column('counter_value', INT),
                         test_needs_acid=True)
        counters.create(testing.db)

    def teardown(self):
        testing.db.execute(counters.delete()).close()

    @classmethod
    def teardown_class(cls):
        counters.drop(testing.db)

    def increment(
        self,
        count,
        errors,
        update_style=True,
        delay=0.005,
        ):
        con = testing.db.connect()
        sel = counters.select(for_update=update_style,
                              whereclause=counters.c.counter_id == 1)
        for i in xrange(count):
            trans = con.begin()
            try:
                existing = con.execute(sel).first()
                incr = existing['counter_value'] + 1
                time.sleep(delay)
                con.execute(counters.update(counters.c.counter_id == 1,
                            values={'counter_value': incr}))
                time.sleep(delay)
                readback = con.execute(sel).first()
                if readback['counter_value'] != incr:
                    raise AssertionError('Got %s post-update, expected '
                            '%s' % (readback['counter_value'], incr))
                trans.commit()
            except Exception, e:
                trans.rollback()
                errors.append(e)
                break
        con.close()

    @testing.crashes('mssql', 'FIXME: unknown')
    @testing.crashes('firebird', 'FIXME: unknown')
    @testing.crashes('sybase', 'FIXME: unknown')
    @testing.crashes('access', 'FIXME: unknown')
    @testing.requires.independent_connections
    def test_queued_update(self):
        """Test SELECT FOR UPDATE with concurrent modifications.
        
        Runs concurrent modifications on a single row in the users
        table, with each mutator trying to increment a value stored in
        user_name.
        
        """

        db = testing.db
        db.execute(counters.insert(), counter_id=1, counter_value=0)
        iterations, thread_count = 10, 5
        threads, errors = [], []
        for i in xrange(thread_count):
            thrd = threading.Thread(target=self.increment,
                                    args=(iterations, ),
                                    kwargs={'errors': errors,
                                    'update_style': True})
            thrd.start()
            threads.append(thrd)
        for thrd in threads:
            thrd.join()
        for e in errors:
            sys.stdout.write('Failure: %s\n' % e)
        self.assert_(len(errors) == 0)
        sel = counters.select(whereclause=counters.c.counter_id == 1)
        final = db.execute(sel).first()
        self.assert_(final['counter_value'] == iterations
                     * thread_count)

    def overlap(
        self,
        ids,
        errors,
        update_style,
        ):
        sel = counters.select(for_update=update_style,
                              whereclause=counters.c.counter_id.in_(ids))
        con = testing.db.connect()
        trans = con.begin()
        try:
            rows = con.execute(sel).fetchall()
            time.sleep(0.25)
            trans.commit()
        except Exception, e:
            trans.rollback()
            errors.append(e)
        con.close()

    def _threaded_overlap(
        self,
        thread_count,
        groups,
        update_style=True,
        pool=5,
        ):
        db = testing.db
        for cid in range(pool - 1):
            db.execute(counters.insert(), counter_id=cid + 1,
                       counter_value=0)
        errors, threads = [], []
        for i in xrange(thread_count):
            thrd = threading.Thread(target=self.overlap,
                                    args=(groups.pop(0), errors,
                                    update_style))
            thrd.start()
            threads.append(thrd)
        for thrd in threads:
            thrd.join()
        return errors

    @testing.crashes('mssql', 'FIXME: unknown')
    @testing.crashes('firebird', 'FIXME: unknown')
    @testing.crashes('sybase', 'FIXME: unknown')
    @testing.crashes('access', 'FIXME: unknown')
    @testing.requires.independent_connections
    def test_queued_select(self):
        """Simple SELECT FOR UPDATE conflict test"""

        errors = self._threaded_overlap(2, [(1, 2, 3), (3, 4, 5)])
        for e in errors:
            sys.stderr.write('Failure: %s\n' % e)
        self.assert_(len(errors) == 0)

    @testing.crashes('mssql', 'FIXME: unknown')
    @testing.fails_on('mysql', 'No support for NOWAIT')
    @testing.crashes('firebird', 'FIXME: unknown')
    @testing.crashes('sybase', 'FIXME: unknown')
    @testing.crashes('access', 'FIXME: unknown')
    @testing.requires.independent_connections
    def test_nowait_select(self):
        """Simple SELECT FOR UPDATE NOWAIT conflict test"""

        errors = self._threaded_overlap(2, [(1, 2, 3), (3, 4, 5)],
                update_style='nowait')
        self.assert_(len(errors) != 0)