diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2013-07-11 15:15:09 -0400 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2013-07-11 15:15:09 -0400 |
commit | fd55be01ddc0ab41dd9469c6c7736d12d5a2f1ea (patch) | |
tree | 460612226d5710aaf151f7552fcba6c1f7cbcd42 /test/engine/test_reconnect.py | |
parent | cd8a40284d0fb053f652f4a3d2745c22674603f8 (diff) | |
download | sqlalchemy-fd55be01ddc0ab41dd9469c6c7736d12d5a2f1ea.tar.gz |
Dialect.initialize() is not called a second time if an :class:`.Engine`
is recreated, due to a disconnect error. This fixes a particular
issue in the Oracle 8 dialect, but in general the dialect.initialize()
phase should only be once per dialect. Also in 0.8.3. [ticket:2776]
Diffstat (limited to 'test/engine/test_reconnect.py')
-rw-r--r-- | test/engine/test_reconnect.py | 34 |
1 files changed, 31 insertions, 3 deletions
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py index 86003bec6..0a964cf63 100644 --- a/test/engine/test_reconnect.py +++ b/test/engine/test_reconnect.py @@ -346,6 +346,28 @@ class MockReconnectTest(fixtures.TestBase): list, result ) + def test_dialect_initialize_once(self): + from sqlalchemy.engine.base import Engine + from sqlalchemy.engine.url import URL + from sqlalchemy.engine.default import DefaultDialect + from sqlalchemy.pool import QueuePool + dbapi = self.dbapi + + mock_dialect = Mock() + class MyURL(URL): + def get_dialect(self): + return Dialect + class Dialect(DefaultDialect): + initialize = Mock() + + engine = create_engine(MyURL("foo://"), module=dbapi) + c1 = engine.connect() + engine.dispose() + c2 = engine.connect() + eq_(Dialect.initialize.call_count, 1) + + + class CursorErrTest(fixtures.TestBase): def setup(self): @@ -494,9 +516,15 @@ class RealReconnectTest(fixtures.TestBase): # raises a DBAPIError, not an AttributeError assert_raises(exc.DBAPIError, engine.connect) - # dispose connections so we get a new one on - # next go - engine.dispose() + @testing.skip_if( + [lambda: util.py3k, "oracle+cx_oracle"], + "Crashes on py3k+cx_oracle") + def test_explode_in_initializer_disconnect(self): + engine = engines.testing_engine() + def broken_initialize(connection): + connection.execute("select fake_stuff from _fake_table") + + engine.dialect.initialize = broken_initialize p1 = engine.pool |