- A DBAPI that raises an error on ``connect()`` which is not a subclass

of dbapi.Error (such as ``TypeError``, ``NotImplementedError``, etc.)
will propagate the exception unchanged.  Previously,
the error handling specific to the ``connect()`` routine would both
inappropriately run the exception through the dialect's
:meth:`.Dialect.is_disconnect` routine as well as wrap it in
a :class:`sqlalchemy.exc.DBAPIError`.  It is now propagated unchanged
in the same way as occurs within the execute process. [ticket:2881]
- add tests for this in test_parseconnect, but also add tests in test_execute
to ensure the execute() behavior as well

1 files changed, 28 insertions, 7 deletions

diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index 0ae747b9c..c4d8b8edc 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -1,4 +1,4 @@
-from sqlalchemy.testing import assert_raises, eq_
+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
@@ -256,17 +256,38 @@ pool_timeout=10
 
     @testing.requires.sqlite
     def test_wraps_connect_in_dbapi(self):
-        # sqlite uses SingletonThreadPool which doesnt have max_overflow
+        e = create_engine('sqlite://')
+        sqlite3 = e.dialect.dbapi
 
-        assert_raises(TypeError, create_engine, 'sqlite://',
-                      max_overflow=5, module=mock_sqlite_dbapi)
-        e = create_engine('sqlite://', connect_args={'use_unicode'
-                          : True}, convert_unicode=True)
+        dbapi = MockDBAPI()
+        dbapi.Error = sqlite3.Error,
+        dbapi.ProgrammingError = sqlite3.ProgrammingError
+        dbapi.connect = Mock(side_effect=sqlite3.ProgrammingError("random error"))
         try:
-            e.connect()
+            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