Merge branch 'master' into oracle-servicename-optionpr/152

3 files changed, 415 insertions, 167 deletions

diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 5c3279ba9..730ef4446 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -174,7 +174,7 @@ class ExecuteTest(fixtures.TestBase):
     @testing.skip_if(
         lambda: testing.against('mysql+mysqldb'), 'db-api flaky')
     @testing.fails_on_everything_except(
-        'postgresql+psycopg2',
+        'postgresql+psycopg2', 'postgresql+psycopg2cffi',
         'postgresql+pypostgresql', 'mysql+mysqlconnector',
         'mysql+pymysql', 'mysql+cymysql')
     def test_raw_python(self):
@@ -639,21 +639,21 @@ class ConvenienceExecuteTest(fixtures.TablesTest):
 
     def test_transaction_connection_ctx_commit(self):
         fn = self._trans_fn(True)
-        conn = testing.db.connect()
-        ctx = conn.begin()
-        testing.run_as_contextmanager(ctx, fn, 5, value=8)
-        self._assert_fn(5, value=8)
+        with testing.db.connect() as conn:
+            ctx = conn.begin()
+            testing.run_as_contextmanager(ctx, fn, 5, value=8)
+            self._assert_fn(5, value=8)
 
     def test_transaction_connection_ctx_rollback(self):
         fn = self._trans_rollback_fn(True)
-        conn = testing.db.connect()
-        ctx = conn.begin()
-        assert_raises_message(
-            Exception,
-            "breakage",
-            testing.run_as_contextmanager, ctx, fn, 5, value=8
-        )
-        self._assert_no_data()
+        with testing.db.connect() as conn:
+            ctx = conn.begin()
+            assert_raises_message(
+                Exception,
+                "breakage",
+                testing.run_as_contextmanager, ctx, fn, 5, value=8
+            )
+            self._assert_no_data()
 
     def test_connection_as_ctx(self):
         fn = self._trans_fn()
@@ -666,10 +666,12 @@ class ConvenienceExecuteTest(fixtures.TablesTest):
     def test_connect_as_ctx_noautocommit(self):
         fn = self._trans_fn()
         self._assert_no_data()
-        ctx = testing.db.connect().execution_options(autocommit=False)
-        testing.run_as_contextmanager(ctx, fn, 5, value=8)
-        # autocommit is off
-        self._assert_no_data()
+
+        with testing.db.connect() as conn:
+            ctx = conn.execution_options(autocommit=False)
+            testing.run_as_contextmanager(ctx, fn, 5, value=8)
+            # autocommit is off
+            self._assert_no_data()
 
     def test_transaction_engine_fn_commit(self):
         fn = self._trans_fn()
@@ -687,17 +689,17 @@ class ConvenienceExecuteTest(fixtures.TablesTest):
 
     def test_transaction_connection_fn_commit(self):
         fn = self._trans_fn()
-        conn = testing.db.connect()
-        conn.transaction(fn, 5, value=8)
-        self._assert_fn(5, value=8)
+        with testing.db.connect() as conn:
+            conn.transaction(fn, 5, value=8)
+            self._assert_fn(5, value=8)
 
     def test_transaction_connection_fn_rollback(self):
         fn = self._trans_rollback_fn()
-        conn = testing.db.connect()
-        assert_raises(
-            Exception,
-            conn.transaction, fn, 5, value=8
-        )
+        with testing.db.connect() as conn:
+            assert_raises(
+                Exception,
+                conn.transaction, fn, 5, value=8
+            )
         self._assert_no_data()
 
 
@@ -1900,6 +1902,272 @@ class HandleErrorTest(fixtures.TestBase):
         self._test_alter_disconnect(True, False)
         self._test_alter_disconnect(False, False)
 
+    def test_handle_error_event_connect_isolation_level(self):
+        engine = engines.testing_engine()
+
+        class MySpecialException(Exception):
+            pass
+
+        @event.listens_for(engine, "handle_error")
+        def handle_error(ctx):
+            raise MySpecialException("failed operation")
+
+        ProgrammingError = engine.dialect.dbapi.ProgrammingError
+        with engine.connect() as conn:
+            with patch.object(
+                conn.dialect, "get_isolation_level",
+                Mock(side_effect=ProgrammingError("random error"))
+            ):
+                assert_raises(
+                    MySpecialException,
+                    conn.get_isolation_level
+                )
+
+
+class HandleInvalidatedOnConnectTest(fixtures.TestBase):
+    __requires__ = ('sqlite', )
+
+    def setUp(self):
+        e = create_engine('sqlite://')
+
+        connection = Mock(
+            get_server_version_info=Mock(return_value='5.0'))
+
+        def connect(*args, **kwargs):
+            return connection
+        dbapi = Mock(
+            sqlite_version_info=(99, 9, 9,),
+            version_info=(99, 9, 9,),
+            sqlite_version='99.9.9',
+            paramstyle='named',
+            connect=Mock(side_effect=connect)
+        )
+
+        sqlite3 = e.dialect.dbapi
+        dbapi.Error = sqlite3.Error,
+        dbapi.ProgrammingError = sqlite3.ProgrammingError
+
+        self.dbapi = dbapi
+        self.ProgrammingError = sqlite3.ProgrammingError
+
+    def test_wraps_connect_in_dbapi(self):
+        dbapi = self.dbapi
+        dbapi.connect = Mock(
+            side_effect=self.ProgrammingError("random error"))
+        try:
+            create_engine('sqlite://', module=dbapi).connect()
+            assert False
+        except tsa.exc.DBAPIError as de:
+            assert not de.connection_invalidated
+
+    def test_handle_error_event_connect(self):
+        dbapi = self.dbapi
+        dbapi.connect = Mock(
+            side_effect=self.ProgrammingError("random error"))
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is None
+            raise MySpecialException("failed operation")
+
+        assert_raises(
+            MySpecialException,
+            eng.connect
+        )
+
+    def test_handle_error_event_revalidate(self):
+        dbapi = self.dbapi
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi, _initialize=False)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is conn
+            assert isinstance(ctx.sqlalchemy_exception, tsa.exc.ProgrammingError)
+            raise MySpecialException("failed operation")
+
+        conn = eng.connect()
+        conn.invalidate()
+
+        dbapi.connect = Mock(
+            side_effect=self.ProgrammingError("random error"))
+
+        assert_raises(
+            MySpecialException,
+            getattr, conn, 'connection'
+        )
+
+    def test_handle_error_event_implicit_revalidate(self):
+        dbapi = self.dbapi
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi, _initialize=False)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is conn
+            assert isinstance(
+                ctx.sqlalchemy_exception, tsa.exc.ProgrammingError)
+            raise MySpecialException("failed operation")
+
+        conn = eng.connect()
+        conn.invalidate()
+
+        dbapi.connect = Mock(
+            side_effect=self.ProgrammingError("random error"))
+
+        assert_raises(
+            MySpecialException,
+            conn.execute, select([1])
+        )
+
+    def test_handle_error_custom_connect(self):
+        dbapi = self.dbapi
+
+        class MySpecialException(Exception):
+            pass
+
+        def custom_connect():
+            raise self.ProgrammingError("random error")
+
+        eng = create_engine('sqlite://', module=dbapi, creator=custom_connect)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is None
+            raise MySpecialException("failed operation")
+
+        assert_raises(
+            MySpecialException,
+            eng.connect
+        )
+
+    def test_handle_error_event_connect_invalidate_flag(self):
+        dbapi = self.dbapi
+        dbapi.connect = Mock(
+            side_effect=self.ProgrammingError(
+                "Cannot operate on a closed database."))
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.is_disconnect
+            ctx.is_disconnect = False
+
+        try:
+            eng.connect()
+            assert False
+        except tsa.exc.DBAPIError as de:
+            assert not de.connection_invalidated
+
+    def test_cant_connect_stay_invalidated(self):
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://')
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.is_disconnect
+
+        conn = eng.connect()
+
+        conn.invalidate()
+
+        eng.pool._creator = Mock(
+            side_effect=self.ProgrammingError(
+                "Cannot operate on a closed database."))
+
+        try:
+            conn.connection
+            assert False
+        except tsa.exc.DBAPIError:
+            assert conn.invalidated
+
+    def _test_dont_touch_non_dbapi_exception_on_connect(self, connect_fn):
+        dbapi = self.dbapi
+        dbapi.connect = Mock(side_effect=TypeError("I'm not a DBAPI error"))
+
+        e = create_engine('sqlite://', module=dbapi)
+        e.dialect.is_disconnect = is_disconnect = Mock()
+        assert_raises_message(
+            TypeError,
+            "I'm not a DBAPI error",
+            connect_fn, e
+        )
+        eq_(is_disconnect.call_count, 0)
+
+    def test_dont_touch_non_dbapi_exception_on_connect(self):
+        self._test_dont_touch_non_dbapi_exception_on_connect(
+            lambda engine: engine.connect())
+
+    def test_dont_touch_non_dbapi_exception_on_contextual_connect(self):
+        self._test_dont_touch_non_dbapi_exception_on_connect(
+            lambda engine: engine.contextual_connect())
+
+    def test_ensure_dialect_does_is_disconnect_no_conn(self):
+        """test that is_disconnect() doesn't choke if no connection,
+        cursor given."""
+        dialect = testing.db.dialect
+        dbapi = dialect.dbapi
+        assert not dialect.is_disconnect(
+            dbapi.OperationalError("test"), None, None)
+
+    def _test_invalidate_on_connect(self, connect_fn):
+        """test that is_disconnect() is called during connect.
+
+        interpretation of connection failures are not supported by
+        every backend.
+
+        """
+
+        dbapi = self.dbapi
+        dbapi.connect = Mock(
+            side_effect=self.ProgrammingError(
+                "Cannot operate on a closed database."))
+        try:
+            connect_fn(create_engine('sqlite://', module=dbapi))
+            assert False
+        except tsa.exc.DBAPIError as de:
+            assert de.connection_invalidated
+
+    def test_invalidate_on_connect(self):
+        """test that is_disconnect() is called during connect.
+
+        interpretation of connection failures are not supported by
+        every backend.
+
+        """
+        self._test_invalidate_on_connect(lambda engine: engine.connect())
+
+    def test_invalidate_on_contextual_connect(self):
+        """test that is_disconnect() is called during connect.
+
+        interpretation of connection failures are not supported by
+        every backend.
+
+        """
+        self._test_invalidate_on_connect(
+            lambda engine: engine.contextual_connect())
+
 
 class ProxyConnectionTest(fixtures.TestBase):
 
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index 391b92144..e53a99e15 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -1,12 +1,15 @@
 from sqlalchemy.testing import assert_raises, eq_, assert_raises_message
-from sqlalchemy.util.compat import configparser, StringIO
 import sqlalchemy.engine.url as url
 from sqlalchemy import create_engine, engine_from_config, exc, pool
 from sqlalchemy.engine.default import DefaultDialect
 import sqlalchemy as tsa
 from sqlalchemy.testing import fixtures
 from sqlalchemy import testing
-from sqlalchemy.testing.mock import Mock, MagicMock, patch
+from sqlalchemy.testing.mock import Mock, MagicMock
+from sqlalchemy import event
+from sqlalchemy import select
+
+dialect = None
 
 
 class ParseConnectTest(fixtures.TestBase):
@@ -31,21 +34,25 @@ class ParseConnectTest(fixtures.TestBase):
             'dbtype://username:password@/database',
             'dbtype:////usr/local/_xtest@example.com/members.db',
             'dbtype://username:apples%2Foranges@hostspec/database',
-            'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]/database?foo=bar',
-            'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]:80/database?foo=bar'
-            ):
+            'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]'
+            '/database?foo=bar',
+            'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]:80'
+            '/database?foo=bar'
+        ):
             u = url.make_url(text)
 
             assert u.drivername in ('dbtype', 'dbtype+apitype')
             assert u.username in ('username', None)
             assert u.password in ('password', 'apples/oranges', None)
-            assert u.host in ('hostspec', '127.0.0.1',
-                        '2001:da8:2004:1000:202:116:160:90', '', None), u.host
-            assert u.database in ('database',
-                    '/usr/local/_xtest@example.com/members.db',
-                    '/usr/db_file.db', ':memory:', '',
-                    'foo/bar/im/a/file',
-                    'E:/work/src/LEM/db/hello.db', None), u.database
+            assert u.host in (
+                'hostspec', '127.0.0.1',
+                '2001:da8:2004:1000:202:116:160:90', '', None), u.host
+            assert u.database in (
+                'database',
+                '/usr/local/_xtest@example.com/members.db',
+                '/usr/db_file.db', ':memory:', '',
+                'foo/bar/im/a/file',
+                'E:/work/src/LEM/db/hello.db', None), u.database
             eq_(str(u), text)
 
     def test_rfc1738_password(self):
@@ -53,13 +60,17 @@ class ParseConnectTest(fixtures.TestBase):
         eq_(u.password, "pass word + other:words")
         eq_(str(u), "dbtype://user:pass word + other%3Awords@host/dbname")
 
-        u = url.make_url('dbtype://username:apples%2Foranges@hostspec/database')
+        u = url.make_url(
+            'dbtype://username:apples%2Foranges@hostspec/database')
         eq_(u.password, "apples/oranges")
         eq_(str(u), 'dbtype://username:apples%2Foranges@hostspec/database')
 
-        u = url.make_url('dbtype://username:apples%40oranges%40%40@hostspec/database')
+        u = url.make_url(
+            'dbtype://username:apples%40oranges%40%40@hostspec/database')
         eq_(u.password, "apples@oranges@@")
-        eq_(str(u), 'dbtype://username:apples%40oranges%40%40@hostspec/database')
+        eq_(
+            str(u),
+            'dbtype://username:apples%40oranges%40%40@hostspec/database')
 
         u = url.make_url('dbtype://username%40:@hostspec/database')
         eq_(u.password, '')
@@ -70,23 +81,23 @@ class ParseConnectTest(fixtures.TestBase):
         eq_(u.password, 'pass/word')
         eq_(str(u), 'dbtype://username:pass%2Fword@hostspec/database')
 
+
 class DialectImportTest(fixtures.TestBase):
     def test_import_base_dialects(self):
-
         # the globals() somehow makes it for the exec() + nose3.
 
         for name in (
-            'mysql',
-            'firebird',
-            'postgresql',
-            'sqlite',
-            'oracle',
-            'mssql',
-            ):
+                'mysql',
+                'firebird',
+                'postgresql',
+                'sqlite',
+                'oracle',
+                'mssql'):
             exec ('from sqlalchemy.dialects import %s\ndialect = '
                   '%s.dialect()' % (name, name), globals())
             eq_(dialect.name, name)
 
+
 class CreateEngineTest(fixtures.TestBase):
     """test that create_engine arguments of different types get
     propagated properly"""
@@ -97,26 +108,28 @@ class CreateEngineTest(fixtures.TestBase):
             create_engine('postgresql://scott:tiger@somehost/test?foobe'
                           'r=12&lala=18&fooz=somevalue', module=dbapi,
                           _initialize=False)
-        c = e.connect()
+        e.connect()
 
     def test_kwargs(self):
         dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'},
                           fooz='somevalue')
         e = \
-            create_engine('postgresql://scott:tiger@somehost/test?fooz='
-                          'somevalue', connect_args={'foober': 12,
-                          'lala': 18, 'hoho': {'this': 'dict'}},
-                          module=dbapi, _initialize=False)
-        c = e.connect()
-
+            create_engine(
+                'postgresql://scott:tiger@somehost/test?fooz='
+                'somevalue', connect_args={
+                    'foober': 12,
+                    'lala': 18, 'hoho': {'this': 'dict'}},
+                module=dbapi, _initialize=False)
+        e.connect()
 
     def test_engine_from_config(self):
         dbapi = mock_dbapi
 
-        config = \
-            {'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test'\
-             '?fooz=somevalue', 'sqlalchemy.pool_recycle': '50',
-             'sqlalchemy.echo': 'true'}
+        config = {
+            'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test'
+            '?fooz=somevalue',
+            'sqlalchemy.pool_recycle': '50',
+            'sqlalchemy.echo': 'true'}
 
         e = engine_from_config(config, module=dbapi, _initialize=False)
         assert e.pool._recycle == 50
@@ -125,7 +138,6 @@ class CreateEngineTest(fixtures.TestBase):
                             'z=somevalue')
         assert e.echo is True
 
-
     def test_engine_from_config_custom(self):
         from sqlalchemy import util
         from sqlalchemy.dialects import registry
@@ -143,8 +155,9 @@ class CreateEngineTest(fixtures.TestBase):
 
         global dialect
         dialect = MyDialect
-        registry.register("mockdialect.barb",
-                    ".".join(tokens[0:-1]), tokens[-1])
+        registry.register(
+            "mockdialect.barb",
+            ".".join(tokens[0:-1]), tokens[-1])
 
         config = {
             "sqlalchemy.url": "mockdialect+barb://",
@@ -155,7 +168,6 @@ class CreateEngineTest(fixtures.TestBase):
         eq_(e.dialect.foobar, 5)
         eq_(e.dialect.bathoho, False)
 
-
     def test_custom(self):
         dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'},
                           fooz='somevalue')
@@ -169,7 +181,7 @@ class CreateEngineTest(fixtures.TestBase):
 
         e = create_engine('postgresql://', creator=connect,
                           module=dbapi, _initialize=False)
-        c = e.connect()
+        e.connect()
 
     def test_recycle(self):
         dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'},
@@ -188,8 +200,9 @@ class CreateEngineTest(fixtures.TestBase):
             (True, pool.reset_rollback),
             (False, pool.reset_none),
         ]:
-            e = create_engine('postgresql://', pool_reset_on_return=value,
-                          module=dbapi, _initialize=False)
+            e = create_engine(
+                'postgresql://', pool_reset_on_return=value,
+                module=dbapi, _initialize=False)
             assert e.pool._reset_on_return is expected
 
         assert_raises(
@@ -217,7 +230,7 @@ class CreateEngineTest(fixtures.TestBase):
             lala=5,
             use_ansi=True,
             module=mock_dbapi,
-            )
+        )
         assert_raises(TypeError, create_engine, 'postgresql://',
                       lala=5, module=mock_dbapi)
         assert_raises(TypeError, create_engine, 'sqlite://', lala=5,
@@ -225,69 +238,6 @@ class CreateEngineTest(fixtures.TestBase):
         assert_raises(TypeError, create_engine, 'mysql+mysqldb://',
                       use_unicode=True, module=mock_dbapi)
 
-    @testing.requires.sqlite
-    def test_wraps_connect_in_dbapi(self):
-        e = create_engine('sqlite://')
-        sqlite3 = e.dialect.dbapi
-
-        dbapi = MockDBAPI()
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-        dbapi.connect = Mock(side_effect=sqlite3.ProgrammingError("random error"))
-        try:
-            create_engine('sqlite://', module=dbapi).connect()
-            assert False
-        except tsa.exc.DBAPIError as de:
-            assert not de.connection_invalidated
-
-
-    @testing.requires.sqlite
-    def test_dont_touch_non_dbapi_exception_on_connect(self):
-        e = create_engine('sqlite://')
-        sqlite3 = e.dialect.dbapi
-
-        dbapi = MockDBAPI()
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-        dbapi.connect = Mock(side_effect=TypeError("I'm not a DBAPI error"))
-        e = create_engine('sqlite://', module=dbapi)
-        e.dialect.is_disconnect = is_disconnect = Mock()
-        assert_raises_message(
-            TypeError,
-            "I'm not a DBAPI error",
-            e.connect
-        )
-        eq_(is_disconnect.call_count, 0)
-
-    def test_ensure_dialect_does_is_disconnect_no_conn(self):
-        """test that is_disconnect() doesn't choke if no connection, cursor given."""
-        dialect = testing.db.dialect
-        dbapi = dialect.dbapi
-        assert not dialect.is_disconnect(dbapi.OperationalError("test"), None, None)
-
-    @testing.requires.sqlite
-    def test_invalidate_on_connect(self):
-        """test that is_disconnect() is called during connect.
-
-        interpretation of connection failures are not supported by
-        every backend.
-
-        """
-
-        e = create_engine('sqlite://')
-        sqlite3 = e.dialect.dbapi
-
-        dbapi = MockDBAPI()
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-        dbapi.connect = Mock(side_effect=sqlite3.ProgrammingError(
-                                    "Cannot operate on a closed database."))
-        try:
-            create_engine('sqlite://', module=dbapi).connect()
-            assert False
-        except tsa.exc.DBAPIError as de:
-            assert de.connection_invalidated
-
     def test_urlattr(self):
         """test the url attribute on ``Engine``."""
 
@@ -313,7 +263,7 @@ class CreateEngineTest(fixtures.TestBase):
             echo_pool=None,
             module=mock_dbapi,
             _initialize=False,
-            )
+        )
         assert e.pool._recycle == 50
 
         # these args work for QueuePool
@@ -325,7 +275,7 @@ class CreateEngineTest(fixtures.TestBase):
             poolclass=tsa.pool.QueuePool,
             module=mock_dbapi,
             _initialize=False,
-            )
+        )
 
         # but not SingletonThreadPool
 
@@ -338,7 +288,8 @@ class CreateEngineTest(fixtures.TestBase):
             poolclass=tsa.pool.SingletonThreadPool,
             module=mock_sqlite_dbapi,
             _initialize=False,
-            )
+        )
+
 
 class TestRegNewDBAPI(fixtures.TestBase):
     def test_register_base(self):
@@ -361,7 +312,8 @@ class TestRegNewDBAPI(fixtures.TestBase):
 
         global dialect
         dialect = MockDialect
-        registry.register("mockdialect.foob", ".".join(tokens[0:-1]), tokens[-1])
+        registry.register(
+            "mockdialect.foob", ".".join(tokens[0:-1]), tokens[-1])
 
         e = create_engine("mockdialect+foob://")
         assert isinstance(e.dialect, MockDialect)
@@ -373,13 +325,16 @@ class TestRegNewDBAPI(fixtures.TestBase):
         e = create_engine("mysql+my_mock_dialect://")
         assert isinstance(e.dialect, MockDialect)
 
+
 class MockDialect(DefaultDialect):
     @classmethod
     def dbapi(cls, **kw):
         return MockDBAPI()
 
+
 def MockDBAPI(**assert_kwargs):
     connection = Mock(get_server_version_info=Mock(return_value='5.0'))
+
     def connect(*args, **kwargs):
         for k in assert_kwargs:
             assert k in kwargs, 'key %s not present in dictionary' % k
@@ -389,12 +344,12 @@ def MockDBAPI(**assert_kwargs):
         return connection
 
     return MagicMock(
-                sqlite_version_info=(99, 9, 9,),
-                version_info=(99, 9, 9,),
-                sqlite_version='99.9.9',
-                paramstyle='named',
-                connect=Mock(side_effect=connect)
-            )
+        sqlite_version_info=(99, 9, 9,),
+        version_info=(99, 9, 9,),
+        sqlite_version='99.9.9',
+        paramstyle='named',
+        connect=Mock(side_effect=connect)
+    )
 
 mock_dbapi = MockDBAPI()
 mock_sqlite_dbapi = msd = MockDBAPI()
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index b3b17e75a..0f5bb4cb5 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -1240,7 +1240,7 @@ class IsolationLevelTest(fixtures.TestBase):
 
         eng = testing_engine()
         isolation_level = eng.dialect.get_isolation_level(
-                                eng.connect().connection)
+            eng.connect().connection)
         level = self._non_default_isolation_level()
 
         ne_(isolation_level, level)
@@ -1248,7 +1248,7 @@ class IsolationLevelTest(fixtures.TestBase):
         eng = testing_engine(options=dict(isolation_level=level))
         eq_(
             eng.dialect.get_isolation_level(
-                                eng.connect().connection),
+                eng.connect().connection),
             level
         )
 
@@ -1270,7 +1270,7 @@ class IsolationLevelTest(fixtures.TestBase):
     def test_default_level(self):
         eng = testing_engine(options=dict())
         isolation_level = eng.dialect.get_isolation_level(
-                                        eng.connect().connection)
+            eng.connect().connection)
         eq_(isolation_level, self._default_isolation_level())
 
     def test_reset_level(self):
@@ -1282,8 +1282,8 @@ class IsolationLevelTest(fixtures.TestBase):
         )
 
         eng.dialect.set_isolation_level(
-                conn.connection, self._non_default_isolation_level()
-            )
+            conn.connection, self._non_default_isolation_level()
+        )
         eq_(
             eng.dialect.get_isolation_level(conn.connection),
             self._non_default_isolation_level()
@@ -1298,14 +1298,15 @@ class IsolationLevelTest(fixtures.TestBase):
         conn.close()
 
     def test_reset_level_with_setting(self):
-        eng = testing_engine(options=dict(
-                            isolation_level=
-                                self._non_default_isolation_level()))
+        eng = testing_engine(
+            options=dict(
+                isolation_level=self._non_default_isolation_level()))
         conn = eng.connect()
         eq_(eng.dialect.get_isolation_level(conn.connection),
             self._non_default_isolation_level())
-        eng.dialect.set_isolation_level(conn.connection,
-                self._default_isolation_level())
+        eng.dialect.set_isolation_level(
+            conn.connection,
+            self._default_isolation_level())
         eq_(eng.dialect.get_isolation_level(conn.connection),
             self._default_isolation_level())
         eng.dialect.reset_isolation_level(conn.connection)
@@ -1317,22 +1318,24 @@ class IsolationLevelTest(fixtures.TestBase):
         eng = testing_engine(options=dict(isolation_level='FOO'))
         assert_raises_message(
             exc.ArgumentError,
-                "Invalid value '%s' for isolation_level. "
-                "Valid isolation levels for %s are %s" %
-                ("FOO", eng.dialect.name,
-                ", ".join(eng.dialect._isolation_lookup)),
-            eng.connect)
+            "Invalid value '%s' for isolation_level. "
+            "Valid isolation levels for %s are %s" %
+            ("FOO",
+             eng.dialect.name, ", ".join(eng.dialect._isolation_lookup)),
+            eng.connect
+        )
 
     def test_per_connection(self):
         from sqlalchemy.pool import QueuePool
-        eng = testing_engine(options=dict(
-                                poolclass=QueuePool,
-                                pool_size=2, max_overflow=0))
+        eng = testing_engine(
+            options=dict(
+                poolclass=QueuePool,
+                pool_size=2, max_overflow=0))
 
         c1 = eng.connect()
         c1 = c1.execution_options(
-                    isolation_level=self._non_default_isolation_level()
-                )
+            isolation_level=self._non_default_isolation_level()
+        )
         c2 = eng.connect()
         eq_(
             eng.dialect.get_isolation_level(c1.connection),
@@ -1366,19 +1369,41 @@ class IsolationLevelTest(fixtures.TestBase):
             r"per-engine using the isolation_level "
             r"argument to create_engine\(\).",
             select([1]).execution_options,
-                    isolation_level=self._non_default_isolation_level()
+            isolation_level=self._non_default_isolation_level()
         )
 
-
     def test_per_engine(self):
         # new in 0.9
-        eng = create_engine(testing.db.url,
-                            execution_options={
-                                'isolation_level':
-                                    self._non_default_isolation_level()}
-                        )
+        eng = create_engine(
+            testing.db.url,
+            execution_options={
+                'isolation_level':
+                    self._non_default_isolation_level()}
+        )
         conn = eng.connect()
         eq_(
             eng.dialect.get_isolation_level(conn.connection),
             self._non_default_isolation_level()
         )
+
+    def test_isolation_level_accessors_connection_default(self):
+        eng = create_engine(
+            testing.db.url
+        )
+        with eng.connect() as conn:
+            eq_(conn.default_isolation_level, self._default_isolation_level())
+        with eng.connect() as conn:
+            eq_(conn.get_isolation_level(), self._default_isolation_level())
+
+    def test_isolation_level_accessors_connection_option_modified(self):
+        eng = create_engine(
+            testing.db.url
+        )
+        with eng.connect() as conn:
+            c2 = conn.execution_options(
+                isolation_level=self._non_default_isolation_level())
+            eq_(conn.default_isolation_level, self._default_isolation_level())
+            eq_(conn.get_isolation_level(),
+                self._non_default_isolation_level())
+            eq_(c2.get_isolation_level(), self._non_default_isolation_level())
+