diff options
Diffstat (limited to 'test/engine/test_parseconnect.py')
| -rw-r--r-- | test/engine/test_parseconnect.py | 35 |
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 |