summaryrefslogtreecommitdiff
path: root/test/engine/test_reconnect.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2013-07-11 15:15:09 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2013-07-11 15:15:09 -0400
commitfd55be01ddc0ab41dd9469c6c7736d12d5a2f1ea (patch)
tree460612226d5710aaf151f7552fcba6c1f7cbcd42 /test/engine/test_reconnect.py
parentcd8a40284d0fb053f652f4a3d2745c22674603f8 (diff)
downloadsqlalchemy-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.py34
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