Add support for SQLTablePrivileges

We're limited in what we can test, as we don't know that we'll be
running with GRANT permissions in the pipeline test harness, but
we can at least make sure the function returns a results set with
the right shape and column names. No point in testing on SQLite,
which doesn't implement a PRIVILEGES model.

I also suppressed .DS_Store in .gitignore.

Closes #1340

Author: bkline

.gitignore

@@ -58,6 +58,7 @@ pyodbc.conf
 tmp
 tags
 xxx_*
+.DS_Store
 
 # The Access unit tests copy empty.accdb and empty.mdb to these names and use them.
 test.accdb

src/cursor.cpp

@@ -1411,6 +1411,151 @@ static PyObject* Cursor_tables(PyObject* self, PyObject* args, PyObject* kwargs)
 }
 
 
+static char tablePrivileges_doc[] =
+    "C.tablePrivileges(table=None, catalog=None, schema=None) --> self\n"
+    "\n"
+    "Executes SQLTablePrivileges and creates a results set of tables and the\n"
+    "privileges associated with those tables.\n"
+    "\n"
+    "Pattern strings can be provided to filter the results set.  Some data\n"
+    "sources add additional filtering to suppress rows based on the rights\n"
+    "of the current user.\n"
+    "\n"
+    "For the table and schema values the '_' and '%' characters are interpreted\n"
+    "as wildcards.  The escape character is driver specific, so you should use\n"
+    "the Connection.searchescape property if you need to include one of the\n"
+    "wildcard characters as part of the name being searched.\n"
+    "\n"
+    "If the SQL_ATTR_METADATA_ID statement attribute is set to SQL_TRUE, the\n"
+    "arguments are treated as identifiers and their case is not significant.\n"
+    "Otherwise they are treated literally, and case is significant.\n"
+    "\n"
+    "table\n"
+    "  Optional string search pattern for table names.\n\n"
+    "catalog\n"
+    "  Table catalog name.  If a driver supports catalogs for some tables but\n"
+    "  not for others, such as when the driver retrieves data from different\n"
+    "  DBMSs, an empty string ('') denotes those tables that do not have\n"
+    "  catalogs.  This argument cannot contain a search string pattern.\n\n"
+    "schema\n"
+    "  Search string pattern for schema names.  If a driver supports schemas\n"
+    "  for some tables but not others, an empty string ('') is used to restrict\n"
+    "  the results to tables that do not have schemas.\n\n"
+    "Each row fetched has the following columns:\n"
+    "  0) table_cat\n"
+    "  1) table_schem\n"
+    "  2) table_name\n"
+    "  3) grantor\n"
+    "  4) grantee\n"
+    "  5) privilege\n"
+    "  6) is_grantable";
+
+
+static PyObject* Cursor_tablePrivileges(PyObject* self, PyObject* args, PyObject* kwargs)
+{
+    PyObject* pCatalog = 0;
+    PyObject* pSchema = 0;
+    PyObject* pTable = 0;
+
+    char* kwnames[] = { "table", "catalog", "schema", 0 };
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OOO", kwnames, &pTable, &pCatalog, &pSchema))
+        return 0;
+
+    Cursor* cur = Cursor_Validate(self, CURSOR_REQUIRE_OPEN);
+
+    if (!free_results(cur, FREE_STATEMENT | FREE_PREPARED))
+        return 0;
+
+    SQLRETURN ret = 0;
+
+    // Use the cursor's encoding.
+    const TextEnc* penc = &cur->cnxn->unicode_enc;
+    bool isWide = penc->ctype == SQL_C_WCHAR;
+    Object oTable;
+    Object oCatalog;
+    Object oSchema;
+    if (pTable && pTable != Py_None)
+    {
+        oTable = penc->Encode(pTable);
+        if (!oTable)
+            return 0;
+    }
+    if (pCatalog && pCatalog != Py_None)
+    {
+        oCatalog = penc->Encode(pCatalog);
+        if (!oCatalog)
+            return 0;
+    }
+    if (pSchema && pSchema != Py_None)
+    {
+        oSchema = penc->Encode(pSchema);
+        if (!oSchema)
+            return 0;
+    }
+    char* szTable = 0;
+    SQLSMALLINT cchTable = SQL_NTS;
+    if (oTable)
+    {
+        szTable = PyBytes_AS_STRING(oTable.Get());
+        if (isWide)
+            cchTable = (SQLSMALLINT)(PyBytes_GET_SIZE(oTable.Get()) / sizeof(uint16_t));
+    }
+    char* szCatalog = 0;
+    SQLSMALLINT cchCatalog = SQL_NTS;
+    if (oCatalog)
+    {
+        szCatalog = PyBytes_AS_STRING(oCatalog.Get());
+        if (isWide)
+            cchCatalog = (SQLSMALLINT)(PyBytes_GET_SIZE(oCatalog.Get()) / sizeof(uint16_t));
+    }
+    char* szSchema = 0;
+    SQLSMALLINT cchSchema = SQL_NTS;
+    if (oSchema)
+    {
+        szSchema = PyBytes_AS_STRING(oSchema.Get());
+        if (isWide)
+            cchSchema = (SQLSMALLINT)(PyBytes_GET_SIZE(oSchema.Get()) / sizeof(uint16_t));
+    }
+
+    Py_BEGIN_ALLOW_THREADS
+    if (isWide)
+        ret = SQLTablePrivilegesW(
+            cur->hstmt,
+            (SQLWCHAR*)szCatalog, cchCatalog,
+            (SQLWCHAR*)szSchema, cchSchema,
+            (SQLWCHAR*)szTable, cchTable
+        );
+    else
+        ret = SQLTablePrivileges(
+            cur->hstmt,
+            (SQLCHAR*)szCatalog, cchCatalog,
+            (SQLCHAR*)szSchema, cchSchema,
+            (SQLCHAR*)szTable, cchTable
+        );
+    Py_END_ALLOW_THREADS
+
+    if (!SQL_SUCCEEDED(ret))
+        return RaiseErrorFromHandle(cur->cnxn, "SQLTablePrivileges", cur->cnxn->hdbc, cur->hstmt);
+
+    SQLSMALLINT cCols;
+    Py_BEGIN_ALLOW_THREADS
+    ret = SQLNumResultCols(cur->hstmt, &cCols);
+    Py_END_ALLOW_THREADS
+    if (!SQL_SUCCEEDED(ret))
+        return RaiseErrorFromHandle(cur->cnxn, "SQLNumResultCols", cur->cnxn->hdbc, cur->hstmt);
+
+    if (!PrepareResults(cur, cCols))
+        return 0;
+
+    if (!create_name_map(cur, cCols, true))
+        return 0;
+
+    // Return the cursor so the results can be iterated over directly.
+    Py_INCREF(cur);
+    return (PyObject*)cur;
+}
+
+
 static char columns_doc[] =
     "C.columns(table=None, catalog=None, schema=None, column=None)\n\n"
     "Creates a results set of column names in specified tables by executing the ODBC SQLColumns function.\n"
@@ -2396,6 +2541,7 @@ static PyMethodDef Cursor_methods[] =
     { "fetchmany",        (PyCFunction)Cursor_fetchmany,        METH_VARARGS,               fetchmany_doc        },
     { "nextset",          (PyCFunction)Cursor_nextset,          METH_NOARGS,                nextset_doc          },
     { "tables",           (PyCFunction)Cursor_tables,           METH_VARARGS|METH_KEYWORDS, tables_doc           },
+    { "tablePrivileges",  (PyCFunction)Cursor_tablePrivileges,  METH_VARARGS|METH_KEYWORDS, tablePrivileges_doc  },
     { "columns",          (PyCFunction)Cursor_columns,          METH_VARARGS|METH_KEYWORDS, columns_doc          },
     { "statistics",       (PyCFunction)Cursor_statistics,       METH_VARARGS|METH_KEYWORDS, statistics_doc       },
     { "rowIdColumns",     (PyCFunction)Cursor_rowIdColumns,     METH_VARARGS|METH_KEYWORDS, rowIdColumns_doc     },

tests/mysql_test.py

@@ -353,6 +353,19 @@ def test_row_description(cursor: pyodbc.Cursor):
     assert cursor.description == row.cursor_description
 
 
+def test_table_privileges(cursor: pyodbc.Cursor):
+    # Confirm exposure of SQLTablePrivileges.  We're limited in what we can test, as
+    # we can't control whether we're running with permission to create users or grant
+    # permissions.  We can at least verify that the method generates a results set
+    # with the right columns.
+    cols = ["table_cat", "table_schem", "table_name", "grantor",
+            "grantee", "privilege", "is_grantable"]
+    cursor.tablePrivileges()
+    names = [col[0] for col in cursor.description]
+    assert len(cols) == len(names), "privileges results set has the wrong shape"
+    assert cols == names, "unexpected column names for privileges results set"
+
+
 def test_executemany(cursor: pyodbc.Cursor):
     cursor.execute("create table t1(a int, b varchar(10))")
 

tests/postgresql_test.py

@@ -490,6 +490,20 @@ def _get_column_size(row):
     assert row.type_name == 'varchar'
     assert _get_column_size(row) == 3
 
+
+def test_table_privileges(cursor: pyodbc.Cursor):
+    # Confirm exposure of SQLTablePrivileges.  We're limited in what we can test, as
+    # we can't control whether we're running with permission to create users or grant
+    # permissions.  We can at least verify that the method generates a results set
+    # with the right columns.
+    cols = ["table_cat", "table_schem", "table_name", "grantor",
+            "grantee", "privilege", "is_grantable"]
+    cursor.tablePrivileges()
+    names = [col[0] for col in cursor.description]
+    assert len(cols) == len(names), "privileges results set has the wrong shape"
+    assert cols == names, "unexpected column names for privileges results set"
+
+
 def test_cancel(cursor: pyodbc.Cursor):
     # I'm not sure how to reliably cause a hang to cancel, so for now we'll settle with
     # making sure SQLCancel is called correctly.

tests/sqlserver_test.py

@@ -1426,6 +1426,19 @@ def test_columns(cursor: pyodbc.Cursor):
         cursor.execute(f"drop table {table_name}")
 
 
+def test_table_privileges(cursor: pyodbc.Cursor):
+    # Confirm exposure of SQLTablePrivileges.  We're limited in what we can test, as
+    # we can't control whether we're running with permission to create users or grant
+    # permissions.  We can at least verify that the method generates a results set
+    # with the right columns.
+    cols = ["table_cat", "table_schem", "table_name", "grantor",
+            "grantee", "privilege", "is_grantable"]
+    cursor.tablePrivileges()
+    names = [col[0] for col in cursor.description]
+    assert len(cols) == len(names), "privileges results set has the wrong shape"
+    assert cols == names, "unexpected column names for privileges results set"
+
+
 def test_cancel(cursor: pyodbc.Cursor):
     # I'm not sure how to reliably cause a hang to cancel, so for now we'll settle with
     # making sure SQLCancel is called correctly.