test single and double quote inspection scenarios

Applied a sweep through all included dialects to ensure names that contain
single or double quotes are properly escaped when querying system tables,
for all :class:`.Inspector` methods that accept object names as an argument
(e.g. table names, view names, etc).   SQLite and MSSQL contained two
quoting issues that were repaired.

Fixes: #5456
Change-Id: I3bc98806f5166f3d82275650079ff561446f2aef

2 files changed, 20 insertions, 20 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 06ea80b9e..35e6799c5 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2887,9 +2887,12 @@ class MSDialect(default.DefaultDialect):
         for col in cols:
             colmap[col["name"]] = col
         # We also run an sp_columns to check for identity columns:
-        cursor = connection.exec_driver_sql(
-            "sp_columns @table_name = '%s', "
-            "@table_owner = '%s'" % (tablename, owner)
+        cursor = connection.execute(
+            sql.text(
+                "sp_columns @table_name = :table_name, "
+                "@table_owner = :table_owner",
+            ),
+            {"table_name": tablename, "table_owner": owner},
         )
         ic = None
         while True:
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index a203e786e..2868eabba 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1664,27 +1664,26 @@ class SQLiteDialect(default.DefaultDialect):
         if schema is not None:
             qschema = self.identifier_preparer.quote_identifier(schema)
             master = "%s.sqlite_master" % qschema
-            s = ("SELECT sql FROM %s WHERE name = '%s'" "AND type='view'") % (
+            s = ("SELECT sql FROM %s WHERE name = ? AND type='view'") % (
                 master,
-                view_name,
             )
-            rs = connection.exec_driver_sql(s)
+            rs = connection.exec_driver_sql(s, (view_name,))
         else:
             try:
                 s = (
                     "SELECT sql FROM "
                     " (SELECT * FROM sqlite_master UNION ALL "
                     "  SELECT * FROM sqlite_temp_master) "
-                    "WHERE name = '%s' "
+                    "WHERE name = ? "
                     "AND type='view'"
-                ) % view_name
-                rs = connection.exec_driver_sql(s)
+                )
+                rs = connection.exec_driver_sql(s, (view_name,))
             except exc.DBAPIError:
                 s = (
-                    "SELECT sql FROM sqlite_master WHERE name = '%s' "
+                    "SELECT sql FROM sqlite_master WHERE name = ? "
                     "AND type='view'"
-                ) % view_name
-                rs = connection.exec_driver_sql(s)
+                )
+                rs = connection.exec_driver_sql(s, (view_name,))
 
         result = rs.fetchall()
         if result:
@@ -2132,19 +2131,17 @@ class SQLiteDialect(default.DefaultDialect):
                 "SELECT sql FROM "
                 " (SELECT * FROM %(schema)ssqlite_master UNION ALL "
                 "  SELECT * FROM %(schema)ssqlite_temp_master) "
-                "WHERE name = '%(table)s' "
-                "AND type = 'table'"
-                % {"schema": schema_expr, "table": table_name}
+                "WHERE name = ? "
+                "AND type = 'table'" % {"schema": schema_expr}
             )
-            rs = connection.exec_driver_sql(s)
+            rs = connection.exec_driver_sql(s, (table_name,))
         except exc.DBAPIError:
             s = (
                 "SELECT sql FROM %(schema)ssqlite_master "
-                "WHERE name = '%(table)s' "
-                "AND type = 'table'"
-                % {"schema": schema_expr, "table": table_name}
+                "WHERE name = ? "
+                "AND type = 'table'" % {"schema": schema_expr}
             )
-            rs = connection.exec_driver_sql(s)
+            rs = connection.exec_driver_sql(s, (table_name,))
         return rs.scalar()
 
     def _get_table_pragma(self, connection, pragma, table_name, schema=None):