summaryrefslogtreecommitdiff
path: root/test/engine
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2017-05-22 14:08:55 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2017-05-22 15:51:07 -0400
commit9f0fb6c601829cb7c9f449d57e12e8b95dab51f5 (patch)
tree1201e89aa89fac39316ccfa87567a88b9667fa4c /test/engine
parentda1bc9878b71f6f7b87e2fa7895e1631ae581609 (diff)
downloadsqlalchemy-9f0fb6c601829cb7c9f449d57e12e8b95dab51f5.tar.gz
Allow metadata.reflect() to recover from unreflectable tables
Added support for views that are unreflectable due to stale table definitions, when calling :meth:`.MetaData.reflect`; a warning is emitted for the table that cannot respond to ``DESCRIBE`` but the operation succeeds. The MySQL dialect now raises UnreflectableTableError which is in turn caught by MetaData.reflect(). Reflecting the view standalone raises this error directly. Change-Id: Id8005219d8e073c154cc84a873df911b4a6cf4d6 Fixes: #3871
Diffstat (limited to 'test/engine')
-rw-r--r--test/engine/test_reflection.py31
1 files changed, 31 insertions, 0 deletions
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index 9616c300d..80a804796 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -12,6 +12,8 @@ from sqlalchemy.testing import eq_, is_true, assert_raises, \
from sqlalchemy import testing
from sqlalchemy.util import ue
from sqlalchemy.testing import config
+from sqlalchemy.testing import mock
+from sqlalchemy.testing import expect_warnings
metadata, users = None, None
@@ -972,6 +974,35 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
m9.reflect()
self.assert_(not m9.tables)
+ @testing.provide_metadata
+ def test_reflect_all_unreflectable_table(self):
+ names = ['rt_%s' % name for name in ('a', 'b', 'c', 'd', 'e')]
+
+ for name in names:
+ Table(name, self.metadata,
+ Column('id', sa.Integer, primary_key=True))
+ self.metadata.create_all()
+
+ m = MetaData()
+
+ reflecttable = testing.db.dialect.reflecttable
+
+ def patched(conn, table, *arg, **kw):
+ if table.name == 'rt_c':
+ raise sa.exc.UnreflectableTableError("Can't reflect rt_c")
+ else:
+ return reflecttable(conn, table, *arg, **kw)
+
+ with mock.patch.object(testing.db.dialect, "reflecttable", patched):
+ with expect_warnings("Skipping table rt_c: Can't reflect rt_c"):
+ m.reflect(bind=testing.db)
+
+ assert_raises_message(
+ sa.exc.UnreflectableTableError,
+ "Can't reflect rt_c",
+ Table, 'rt_c', m, autoload_with=testing.db
+ )
+
def test_reflect_all_conn_closing(self):
m1 = MetaData()
c = testing.db.connect()