- rework the handle error on connect tests from test_parsconnect where

they don't really belong into a new suite in test_execute

1 files changed, 0 insertions, 236 deletions

diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index 8d659420d..e53a99e15 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -238,242 +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_handle_error_event_connect(self):
-        e = create_engine('sqlite://')
-        dbapi = MockDBAPI()
-        sqlite3 = e.dialect.dbapi
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-        dbapi.connect = Mock(
-            side_effect=sqlite3.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
-        )
-
-    @testing.requires.sqlite
-    def test_handle_error_event_revalidate(self):
-        e = create_engine('sqlite://')
-        dbapi = MockDBAPI()
-        sqlite3 = e.dialect.dbapi
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-
-        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, exc.ProgrammingError)
-            raise MySpecialException("failed operation")
-
-        conn = eng.connect()
-        conn.invalidate()
-
-        dbapi.connect = Mock(
-            side_effect=sqlite3.ProgrammingError("random error"))
-
-        assert_raises(
-            MySpecialException,
-            getattr, conn, 'connection'
-        )
-
-    @testing.requires.sqlite
-    def test_handle_error_event_implicit_revalidate(self):
-        e = create_engine('sqlite://')
-        dbapi = MockDBAPI()
-        sqlite3 = e.dialect.dbapi
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-
-        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, exc.ProgrammingError)
-            raise MySpecialException("failed operation")
-
-        conn = eng.connect()
-        conn.invalidate()
-
-        dbapi.connect = Mock(
-            side_effect=sqlite3.ProgrammingError("random error"))
-
-        assert_raises(
-            MySpecialException,
-            conn.execute, select([1])
-        )
-
-    @testing.requires.sqlite
-    def test_handle_error_custom_connect(self):
-        e = create_engine('sqlite://')
-
-        dbapi = MockDBAPI()
-        sqlite3 = e.dialect.dbapi
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-
-        class MySpecialException(Exception):
-            pass
-
-        def custom_connect():
-            raise sqlite3.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
-        )
-
-    @testing.requires.sqlite
-    def test_handle_error_event_connect_invalidate_flag(self):
-        e = create_engine('sqlite://')
-        dbapi = MockDBAPI()
-        sqlite3 = e.dialect.dbapi
-        dbapi.Error = sqlite3.Error,
-        dbapi.ProgrammingError = sqlite3.ProgrammingError
-        dbapi.connect = Mock(
-            side_effect=sqlite3.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
-
-    @testing.requires.sqlite
-    def test_cant_connect_stay_invalidated(self):
-        e = create_engine('sqlite://')
-        sqlite3 = e.dialect.dbapi
-
-        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=sqlite3.ProgrammingError(
-                "Cannot operate on a closed database."))
-
-        try:
-            conn.connection
-            assert False
-        except tsa.exc.DBAPIError:
-            assert conn.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``."""