diff options
Diffstat (limited to 'test/engine/test_transaction.py')
| -rw-r--r-- | test/engine/test_transaction.py | 119 |
1 files changed, 89 insertions, 30 deletions
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py index b3b17e75a..b662c7fcd 100644 --- a/test/engine/test_transaction.py +++ b/test/engine/test_transaction.py @@ -1,8 +1,6 @@ from sqlalchemy.testing import eq_, assert_raises, \ - assert_raises_message, ne_ + assert_raises_message, ne_, expect_warnings import sys -import time -import threading from sqlalchemy import event from sqlalchemy.testing.engines import testing_engine from sqlalchemy import create_engine, MetaData, INT, VARCHAR, Sequence, \ @@ -1240,7 +1238,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 +1246,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 +1268,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 +1280,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 +1296,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 +1316,36 @@ 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_connection_invalidated(self): + eng = testing_engine() + conn = eng.connect() + c2 = conn.execution_options( + isolation_level=self._non_default_isolation_level()) + c2.invalidate() + c2.connection + + # TODO: do we want to rebuild the previous isolation? + # for now, this is current behavior so we will leave it. + eq_(c2.get_isolation_level(), self._default_isolation_level()) 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), @@ -1358,6 +1371,30 @@ class IsolationLevelTest(fixtures.TestBase): c3.close() c4.close() + def test_warning_in_transaction(self): + eng = testing_engine() + c1 = eng.connect() + with expect_warnings( + "Connection is already established with a Transaction; " + "setting isolation_level may implicitly rollback or commit " + "the existing transaction, or have no effect until next " + "transaction" + ): + with c1.begin(): + c1 = c1.execution_options( + isolation_level=self._non_default_isolation_level() + ) + + eq_( + eng.dialect.get_isolation_level(c1.connection), + self._non_default_isolation_level() + ) + # stays outside of transaction + eq_( + eng.dialect.get_isolation_level(c1.connection), + self._non_default_isolation_level() + ) + def test_per_statement_bzzt(self): assert_raises_message( exc.ArgumentError, @@ -1366,19 +1403,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()) + |