diff options
Diffstat (limited to 'test/engine')
| -rw-r--r-- | test/engine/test_execute.py | 22 | ||||
| -rw-r--r-- | test/engine/test_reconnect.py | 61 |
2 files changed, 62 insertions, 21 deletions
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py index 54a85bf9f..eff1026cd 100644 --- a/test/engine/test_execute.py +++ b/test/engine/test_execute.py @@ -1,7 +1,7 @@ # coding: utf-8 from sqlalchemy.testing import eq_, assert_raises, assert_raises_message, \ - config, is_, is_not_, le_ + config, is_, is_not_, le_, expect_warnings import re from sqlalchemy.testing.util import picklers from sqlalchemy.interfaces import ConnectionProxy @@ -1834,6 +1834,26 @@ class HandleErrorTest(fixtures.TestBase): ) eq_(patched.call_count, 1) + def test_exception_autorollback_fails(self): + engine = engines.testing_engine() + conn = engine.connect() + + def boom(connection): + raise engine.dialect.dbapi.OperationalError("rollback failed") + + with expect_warnings( + r"An exception has occurred during handling of a previous " + r"exception. The previous exception is.*i_dont_exist", + py2konly=True + ): + with patch.object(conn.dialect, "do_rollback", boom) as patched: + assert_raises_message( + tsa.exc.OperationalError, + "rollback failed", + conn.execute, + "insert into i_dont_exist (x) values ('y')" + ) + def test_exception_event_ad_hoc_context(self): """test that handle_error is called with a context in cases where _handle_dbapi_error() is normally called without diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py index be60056a5..f798ff845 100644 --- a/test/engine/test_reconnect.py +++ b/test/engine/test_reconnect.py @@ -1,4 +1,5 @@ -from sqlalchemy.testing import eq_, ne_, assert_raises, assert_raises_message +from sqlalchemy.testing import eq_, ne_, assert_raises, \ + expect_warnings, assert_raises_message import time from sqlalchemy import ( select, MetaData, Integer, String, create_engine, pool, exc, util) @@ -408,11 +409,17 @@ class MockReconnectTest(fixtures.TestBase): self.dbapi.shutdown("rollback_no_disconnect") # raises error - assert_raises_message( - tsa.exc.DBAPIError, - "something broke on rollback but we didn't lose the connection", - conn.execute, select([1]) - ) + with expect_warnings( + "An exception has occurred during handling .*" + "something broke on execute but we didn't lose the connection", + py2konly=True + ): + assert_raises_message( + tsa.exc.DBAPIError, + "something broke on rollback but we didn't " + "lose the connection", + conn.execute, select([1]) + ) assert conn.closed assert not conn.invalidated @@ -433,11 +440,16 @@ class MockReconnectTest(fixtures.TestBase): self.dbapi.shutdown("rollback") # raises error - assert_raises_message( - tsa.exc.DBAPIError, - "Lost the DB connection on rollback", - conn.execute, select([1]) - ) + with expect_warnings( + "An exception has occurred during handling .*" + "something broke on execute but we didn't lose the connection", + py2konly=True + ): + assert_raises_message( + tsa.exc.DBAPIError, + "Lost the DB connection on rollback", + conn.execute, select([1]) + ) assert not conn.closed assert conn.invalidated @@ -448,11 +460,16 @@ class MockReconnectTest(fixtures.TestBase): self.dbapi.shutdown("rollback") # raises error - assert_raises_message( - tsa.exc.DBAPIError, - "Lost the DB connection on rollback", - conn.execute, select([1]) - ) + with expect_warnings( + "An exception has occurred during handling .*" + "something broke on execute but we didn't lose the connection", + py2konly=True + ): + assert_raises_message( + tsa.exc.DBAPIError, + "Lost the DB connection on rollback", + conn.execute, select([1]) + ) assert conn.closed assert conn.invalidated @@ -765,10 +782,14 @@ class RealReconnectTest(fixtures.TestBase): self.engine.dialect.is_disconnect = is_disconnect conn = self.engine.connect() self.engine.test_shutdown() - assert_raises( - tsa.exc.DBAPIError, - conn.execute, select([1]) - ) + with expect_warnings( + "An exception has occurred during handling .*", + py2konly=True + ): + assert_raises( + tsa.exc.DBAPIError, + conn.execute, select([1]) + ) def test_rollback_on_invalid_plain(self): conn = self.engine.connect() |