Detect VACUUM in SQL analysis

Refs https://github.com/simonw/datasette/pull/2749#issuecomment-4559073803

tests/test_queries.py

@@ -2011,6 +2011,37 @@ async def test_execute_write_rejects_function_operations():
     assert (await db.execute("select name from dogs")).dicts() == []
 
 
+@pytest.mark.asyncio
+async def test_execute_write_rejects_vacuum_operation():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    }
+                }
+            }
+        },
+    )
+    ds.add_memory_database("execute_write_vacuum_operation", name="data")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={"sql": "vacuum"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Unsupported SQL operation: vacuum database"
+    ]
+
+
 @pytest.mark.asyncio
 async def test_execute_write_create_table_uses_create_table_permission():
     ds = Datasette(

tests/test_utils_sql_analysis.py

@@ -129,6 +129,50 @@ def test_analyze_create_table_operation():
     ]
 
 
+def test_analyze_vacuum_operation():
+    conn = sqlite3.connect(":memory:")
+    try:
+        analysis = analyze_sql_tables(conn, "vacuum", database_name="data")
+    finally:
+        conn.close()
+
+    assert [operation_dict(operation) for operation in analysis.operations] == [
+        {
+            "operation": "vacuum",
+            "target_type": "database",
+            "database": "data",
+            "sqlite_schema": "main",
+            "table": None,
+            "target": "data",
+            "columns": (),
+            "source": None,
+            "internal": False,
+        }
+    ]
+
+
+def test_analyze_statement_with_no_authorizer_callbacks_is_unknown():
+    conn = sqlite3.connect(":memory:")
+    try:
+        analysis = analyze_sql_tables(conn, "reindex", database_name="data")
+    finally:
+        conn.close()
+
+    assert [operation_dict(operation) for operation in analysis.operations] == [
+        {
+            "operation": "unknown",
+            "target_type": "statement",
+            "database": "data",
+            "sqlite_schema": None,
+            "table": None,
+            "target": None,
+            "columns": (),
+            "source": None,
+            "internal": False,
+        }
+    ]
+
+
 def test_analyze_transaction_operation(conn):
     analysis = analyze_sql_tables(conn, "commit", database_name="data")