Take schema name into account when querying sqlite_master

Fixed bug where SQLite CHECK constraint reflection would fail
if the referenced table were in a remote schema, e.g. on SQLite a
remote database referred to by ATTACH.

Also add suite support for general CHECK constraint reflection.

Change-Id: I073a72cb47dc4f8c5683000d708768523759332f
Fixes: #4099

3 files changed, 61 insertions, 6 deletions

diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 4bc60be62..d8ce7f394 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1545,14 +1545,19 @@ class SQLiteDialect(default.DefaultDialect):
     def _get_table_sql(self, connection, table_name, schema=None, **kw):
         try:
             s = ("SELECT sql FROM "
-                 " (SELECT * FROM sqlite_master UNION ALL "
-                 "  SELECT * FROM sqlite_temp_master) "
-                 "WHERE name = '%s' "
-                 "AND type = 'table'") % table_name
+                 " (SELECT * FROM %(schema)ssqlite_master UNION ALL "
+                 "  SELECT * FROM %(schema)ssqlite_temp_master) "
+                 "WHERE name = '%(table)s' "
+                 "AND type = 'table'" % {
+                     "schema": ("%s." % schema) if schema else "",
+                     "table": table_name})
             rs = connection.execute(s)
         except exc.DBAPIError:
-            s = ("SELECT sql FROM sqlite_master WHERE name = '%s' "
-                 "AND type = 'table'") % table_name
+            s = ("SELECT sql FROM %(schema)ssqlite_master "
+                 "WHERE name = '%(table)s' "
+                 "AND type = 'table'" % {
+                     "schema": ("%s." % schema) if schema else "",
+                     "table": table_name})
             rs = connection.execute(s)
         return rs.scalar()
 
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index b3ad29a3b..a66b091aa 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -430,6 +430,11 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def check_constraint_reflection(self):
+        """target dialect supports reflection of check constraints"""
+        return exclusions.closed()
+
+    @property
     def duplicate_key_raises_integrity_error(self):
         """target dialect raises IntegrityError when reporting an INSERT
         with a primary key violation.  (hint: it should)
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 7674338b4..f541165e3 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -16,6 +16,7 @@ from sqlalchemy.schema import DDL, Index
 from sqlalchemy import event
 from sqlalchemy.sql.elements import quoted_name
 from sqlalchemy import ForeignKey
+import re
 
 metadata, users = None, None
 
@@ -760,6 +761,50 @@ class ComponentReflectionTest(fixtures.TablesTest):
             eq_(names_that_duplicate_index, idx_names)
             eq_(uq_names, set())
 
+    @testing.requires.check_constraint_reflection
+    def test_get_check_constraints(self):
+        self._test_get_check_constraints()
+
+    @testing.requires.check_constraint_reflection
+    @testing.requires.schemas
+    def test_get_check_constraints_schema(self):
+        self._test_get_check_constraints(schema=testing.config.test_schema)
+
+    @testing.provide_metadata
+    def _test_get_check_constraints(self, schema=None):
+        orig_meta = self.metadata
+        Table(
+            'sa_cc', orig_meta,
+            Column('a', Integer()),
+            sa.CheckConstraint('a > 1 AND a < 5', name='cc1'),
+            sa.CheckConstraint('a = 1 OR (a > 2 AND a < 5)', name='cc2'),
+            schema=schema
+        )
+
+        orig_meta.create_all()
+
+        inspector = inspect(orig_meta.bind)
+        reflected = sorted(
+            inspector.get_check_constraints('sa_cc', schema=schema),
+            key=operator.itemgetter('name')
+        )
+
+        reflected = [
+            {"name": item["name"],
+             # trying to minimize effect of quoting, parenthesis, etc.
+             # may need to add more to this as new dialects get CHECK
+             # constraint reflection support
+             "sqltext": re.sub(r"[`'\(\)]", '', item["sqltext"].lower())}
+            for item in reflected
+        ]
+        eq_(
+            reflected,
+            [
+                {'name': 'cc1', 'sqltext': 'a > 1 and a < 5'},
+                {'name': 'cc2', 'sqltext': 'a = 1 or a > 2 and a < 5'}
+            ]
+        )
+
     @testing.provide_metadata
     def _test_get_view_definition(self, schema=None):
         meta = self.metadata