from ...testing import assert_raises
from ...testing import config
from ...testing import eq_
+from ...testing import is_
from ...testing import is_false
+from ...testing import is_not_
from ...testing import is_true
+from ...testing import ne_
from ...testing.fixtures import TestBase
is_true(self.conn.in_transaction())
with context.autocommit_block():
- is_false(self.conn.in_transaction())
-
+ # in 1.x, self.conn is separate due to the
+ # execution_options call. however for future they are the
+ # same connection and there is a "transaction" block
+ # despite autocommit
+ if self.is_sqlalchemy_future:
+ is_(context.connection, self.conn)
+ else:
+ is_not_(context.connection, self.conn)
+ is_false(self.conn.in_transaction())
+
+ eq_(
+ context.connection._execution_options[
+ "isolation_level"
+ ],
+ "AUTOCOMMIT",
+ )
+
+ ne_(
+ context.connection._execution_options.get(
+ "isolation_level", None
+ ),
+ "AUTOCOMMIT",
+ )
is_true(self.conn.in_transaction())
is_false(self.conn.in_transaction())
is_false(self.conn.in_transaction())
with context.autocommit_block():
- is_false(self.conn.in_transaction())
+ is_true(context.connection.in_transaction())
+
+ # in 1.x, self.conn is separate due to the execution_options
+ # call. however for future they are the same connection and there
+ # is a "transaction" block despite autocommit
+ if self.is_sqlalchemy_future:
+ is_(context.connection, self.conn)
+ else:
+ is_not_(context.connection, self.conn)
+ is_false(self.conn.in_transaction())
+
+ eq_(
+ context.connection._execution_options["isolation_level"],
+ "AUTOCOMMIT",
+ )
+
+ ne_(
+ context.connection._execution_options.get("isolation_level", None),
+ "AUTOCOMMIT",
+ )
+
is_false(self.conn.in_transaction())
def test_autocommit_block_transactional_ddl_sqlmode(self):