Compare commits

...

5 Commits

Author SHA1 Message Date
Amin Ghadersohi
468ac195f0 refactor(mcp): rename undelete_dashboard to restore_dashboard
Matches the platform vocabulary: the REST API exposes
POST /api/v1/dashboard/<uuid>/restore, the command layer is
RestoreDashboardCommand/BaseRestoreCommand, and SoftDeleteMixin's
recovery method is restore(). "undelete" appeared nowhere else in
the codebase.
2026-07-10 02:53:37 +00:00
Amin Ghadersohi
af2392d0f4 test(mcp): cover generic restore-failure path in undelete_dashboard 2026-07-09 14:59:10 +00:00
Amin Ghadersohi
50985f89f1 test(mcp): address review feedback on delete/undelete dashboard tools
- Add test for the DashboardDeleteFailedError path (associated alerts/reports)
- Add sanitization tests for DeletedDashboardSummary, DeleteDashboardResponse,
  and UndeleteDashboardResponse schemas
- Annotate test fixture return types
2026-07-08 19:36:28 +00:00
Amin Ghadersohi
4bfbdecd11 feat(mcp): add undelete_dashboard tool and soft-delete-aware delete responses 2026-07-08 19:36:28 +00:00
Amin Ghadersohi
cfc0515931 feat(mcp): add delete_dashboard tool 2026-07-08 19:36:28 +00:00
8 changed files with 919 additions and 0 deletions

View File

@@ -135,6 +135,8 @@ Dashboard Management:
- add_chart_to_existing_dashboard: Add a chart to an existing dashboard (requires write access) - add_chart_to_existing_dashboard: Add a chart to an existing dashboard (requires write access)
- manage_native_filters: Add, update, remove, or reorder native filters on a dashboard (requires write access; supports filter_select and filter_time) - manage_native_filters: Add, update, remove, or reorder native filters on a dashboard (requires write access; supports filter_select and filter_time)
- remove_chart_from_dashboard: Remove a chart from an existing dashboard (requires write access) - remove_chart_from_dashboard: Remove a chart from an existing dashboard (requires write access)
- delete_dashboard: Delete a dashboard by ID; requires confirm=true; response reports whether the delete is recoverable (requires write access)
- restore_dashboard: Restore a soft-deleted dashboard by ID (requires write access; owners or admins only)
Annotation Layers: Annotation Layers:
- list_annotation_layers: List annotation layers with advanced filters (1-based pagination) - list_annotation_layers: List annotation layers with advanced filters (1-based pagination)
@@ -725,6 +727,7 @@ from superset.mcp_service.chart.tool import ( # noqa: F401, E402
) )
from superset.mcp_service.dashboard.tool import ( # noqa: F401, E402 from superset.mcp_service.dashboard.tool import ( # noqa: F401, E402
add_chart_to_existing_dashboard, add_chart_to_existing_dashboard,
delete_dashboard,
duplicate_dashboard, duplicate_dashboard,
generate_dashboard, generate_dashboard,
get_dashboard_datasets, get_dashboard_datasets,
@@ -733,6 +736,7 @@ from superset.mcp_service.dashboard.tool import ( # noqa: F401, E402
list_dashboards, list_dashboards,
manage_native_filters, manage_native_filters,
remove_chart_from_dashboard, remove_chart_from_dashboard,
restore_dashboard,
update_dashboard, update_dashboard,
) )
from superset.mcp_service.database.tool import ( # noqa: F401, E402 from superset.mcp_service.database.tool import ( # noqa: F401, E402

View File

@@ -676,6 +676,99 @@ class RemoveChartFromDashboardResponse(BaseModel):
return sanitize_for_llm_context(value, field_path=("error",)) return sanitize_for_llm_context(value, field_path=("error",))
class DeleteDashboardRequest(BaseModel):
"""Request schema for deleting a dashboard."""
dashboard_id: int = Field(..., description="ID of the dashboard to delete")
confirm: bool = Field(
...,
description=(
"Explicit confirmation of the deletion. Depending on the "
"deployment, deletion is either recoverable via "
"restore_dashboard (soft delete) or permanent (hard delete) — "
"the response's 'permanent' field reports which applied. The "
"tool refuses to delete unless this is set to true."
),
)
class DeletedDashboardSummary(BaseModel):
"""Summary of a dashboard targeted for deletion or restoration."""
id: int = Field(..., description="ID of the dashboard")
dashboard_title: str | None = Field(None, description="Title of the dashboard")
slug: str | None = Field(None, description="Slug of the dashboard")
uuid: str | None = Field(None, description="UUID of the dashboard")
@field_validator("dashboard_title", "slug")
@classmethod
def sanitize_text_for_llm_context(cls, value: str | None) -> str | None:
"""Wrap user-controlled dashboard text before LLM exposure."""
if value is None:
return value
return sanitize_for_llm_context(value, field_path=("dashboard",))
class DeleteDashboardResponse(BaseModel):
"""Response schema for deleting a dashboard."""
deleted: bool = Field(False, description="True when the dashboard was deleted")
permanent: bool = Field(
False,
description=(
"True when the deletion was a permanent hard delete. False when "
"the dashboard was soft-deleted and can be recovered with "
"restore_dashboard."
),
)
dashboard: DeletedDashboardSummary | None = Field(
None, description="Summary of the deleted (or targeted) dashboard"
)
error: str | None = Field(None, description="Error message, if operation failed")
@field_validator("error")
@classmethod
def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
"""Wrap error text before it is exposed to LLM context.
The error may echo the dashboard-controlled title — it must be wrapped
so the LLM treats it as data, not instructions.
"""
if value is None:
return value
return sanitize_for_llm_context(value, field_path=("error",))
class RestoreDashboardRequest(BaseModel):
"""Request schema for restoring a soft-deleted dashboard."""
dashboard_id: int = Field(
..., description="ID of the soft-deleted dashboard to restore"
)
class RestoreDashboardResponse(BaseModel):
"""Response schema for restoring a dashboard."""
restored: bool = Field(False, description="True when the dashboard was restored")
dashboard: DeletedDashboardSummary | None = Field(
None, description="Summary of the restored (or targeted) dashboard"
)
error: str | None = Field(None, description="Error message, if operation failed")
@field_validator("error")
@classmethod
def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
"""Wrap error text before it is exposed to LLM context.
The error may echo the dashboard-controlled title or slug — both must
be wrapped so the LLM treats them as data, not instructions.
"""
if value is None:
return value
return sanitize_for_llm_context(value, field_path=("error",))
class GenerateDashboardRequest(BaseModel): class GenerateDashboardRequest(BaseModel):
"""Request schema for generating a dashboard.""" """Request schema for generating a dashboard."""

View File

@@ -16,6 +16,7 @@
# under the License. # under the License.
from .add_chart_to_existing_dashboard import add_chart_to_existing_dashboard from .add_chart_to_existing_dashboard import add_chart_to_existing_dashboard
from .delete_dashboard import delete_dashboard
from .duplicate_dashboard import duplicate_dashboard from .duplicate_dashboard import duplicate_dashboard
from .generate_dashboard import generate_dashboard from .generate_dashboard import generate_dashboard
from .get_dashboard_datasets import get_dashboard_datasets from .get_dashboard_datasets import get_dashboard_datasets
@@ -24,6 +25,7 @@ from .get_dashboard_layout import get_dashboard_layout
from .list_dashboards import list_dashboards from .list_dashboards import list_dashboards
from .manage_native_filters import manage_native_filters from .manage_native_filters import manage_native_filters
from .remove_chart_from_dashboard import remove_chart_from_dashboard from .remove_chart_from_dashboard import remove_chart_from_dashboard
from .restore_dashboard import restore_dashboard
from .update_dashboard import update_dashboard from .update_dashboard import update_dashboard
__all__ = [ __all__ = [
@@ -36,5 +38,7 @@ __all__ = [
"add_chart_to_existing_dashboard", "add_chart_to_existing_dashboard",
"manage_native_filters", "manage_native_filters",
"remove_chart_from_dashboard", "remove_chart_from_dashboard",
"delete_dashboard",
"restore_dashboard",
"update_dashboard", "update_dashboard",
] ]

View File

@@ -0,0 +1,167 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
MCP tool: delete_dashboard
This tool deletes a dashboard. It requires an explicit ``confirm=true``
safety gate so callers must state destructive intent. When the
``SOFT_DELETE`` feature flag is enabled the dashboard is soft-deleted and
recoverable via ``restore_dashboard``; otherwise the delete is permanent.
"""
import logging
from fastmcp import Context
from superset_core.mcp.decorators import tool, ToolAnnotations
from superset.extensions import event_logger
from superset.mcp_service.dashboard.schemas import (
DeleteDashboardRequest,
DeleteDashboardResponse,
DeletedDashboardSummary,
)
logger = logging.getLogger(__name__)
@tool(
tags=["mutate"],
class_permission_name="Dashboard",
method_permission_name="write",
annotations=ToolAnnotations(
title="Delete dashboard",
readOnlyHint=False,
destructiveHint=True,
),
)
async def delete_dashboard(
request: DeleteDashboardRequest, ctx: Context
) -> DeleteDashboardResponse:
"""
Delete a dashboard by ID. The charts on the dashboard are NOT deleted —
only the dashboard itself. The tool refuses to run unless
``confirm=true`` is explicitly passed. The response's ``permanent``
field reports whether the delete was a recoverable soft delete
(restore with ``restore_dashboard``) or a permanent hard delete.
"""
from superset import is_feature_enabled
from superset.commands.dashboard.delete import DeleteDashboardCommand
from superset.commands.dashboard.exceptions import (
DashboardDeleteFailedError,
DashboardForbiddenError,
DashboardNotFoundError,
)
from superset.daos.dashboard import DashboardDAO
soft_delete_enabled = is_feature_enabled("SOFT_DELETE")
if not request.confirm:
await ctx.warning(
"Deletion of dashboard %s not confirmed" % (request.dashboard_id,)
)
recovery_note = (
"The dashboard can be restored afterwards with restore_dashboard."
if soft_delete_enabled
else "The deletion is permanent and cannot be undone."
)
return DeleteDashboardResponse(
deleted=False,
permanent=False,
dashboard=None,
error=(
f"Deletion not confirmed. This will delete dashboard "
f"{request.dashboard_id}. {recovery_note} "
"Re-run with confirm=true to proceed."
),
)
summary: DeletedDashboardSummary | None = None
try:
with event_logger.log_context(action="mcp.delete_dashboard.validation"):
dashboard = DashboardDAO.find_by_id(request.dashboard_id)
if not dashboard:
return DeleteDashboardResponse(
deleted=False,
permanent=False,
dashboard=None,
error=(
f"Dashboard with ID {request.dashboard_id} not found. "
"Use list_dashboards to get valid dashboard IDs."
),
)
summary = DeletedDashboardSummary(
id=dashboard.id,
dashboard_title=dashboard.dashboard_title,
slug=dashboard.slug,
uuid=str(dashboard.uuid) if dashboard.uuid else None,
)
with event_logger.log_context(action="mcp.delete_dashboard.delete"):
DeleteDashboardCommand([request.dashboard_id]).run()
logger.info("Deleted dashboard %s", request.dashboard_id)
await ctx.info("Deleted dashboard %s" % (request.dashboard_id,))
return DeleteDashboardResponse(
deleted=True,
permanent=not soft_delete_enabled,
dashboard=summary,
error=None,
)
except DashboardNotFoundError:
return DeleteDashboardResponse(
deleted=False,
permanent=False,
dashboard=None,
error=(
f"Dashboard with ID {request.dashboard_id} not found. "
"Use list_dashboards to get valid dashboard IDs."
),
)
except DashboardForbiddenError:
await ctx.warning(
"Permission denied deleting dashboard %s" % (request.dashboard_id,)
)
return DeleteDashboardResponse(
deleted=False,
permanent=False,
dashboard=summary,
error=(
f"You don't have permission to delete dashboard "
f"with ID {request.dashboard_id}."
),
)
except DashboardDeleteFailedError as exc:
await ctx.error(
"Failed to delete dashboard %s: %s" % (request.dashboard_id, str(exc))
)
return DeleteDashboardResponse(
deleted=False,
permanent=False,
dashboard=summary,
error=f"Failed to delete dashboard {request.dashboard_id}: {exc.message}",
)
except Exception as exc:
await ctx.error(
"Unexpected error deleting dashboard: %s: %s"
% (type(exc).__name__, str(exc))
)
raise

View File

@@ -0,0 +1,165 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
MCP tool: restore_dashboard
This tool restores a soft-deleted dashboard by clearing its ``deleted_at``
timestamp. It is the recovery path for ``delete_dashboard`` when the
``SOFT_DELETE`` feature flag is enabled; hard-deleted dashboards cannot be
restored.
"""
import logging
from fastmcp import Context
from superset_core.mcp.decorators import tool, ToolAnnotations
from superset.extensions import event_logger
from superset.mcp_service.dashboard.schemas import (
DeletedDashboardSummary,
RestoreDashboardRequest,
RestoreDashboardResponse,
)
logger = logging.getLogger(__name__)
@tool(
tags=["mutate"],
class_permission_name="Dashboard",
method_permission_name="write",
annotations=ToolAnnotations(
title="Restore dashboard",
readOnlyHint=False,
destructiveHint=False,
),
)
async def restore_dashboard(
request: RestoreDashboardRequest, ctx: Context
) -> RestoreDashboardResponse:
"""
Restore a soft-deleted dashboard by ID, making it visible and usable
again. Only dashboards deleted while soft delete was enabled can be
restored; hard-deleted dashboards are gone permanently.
"""
from superset.commands.dashboard.exceptions import (
DashboardForbiddenError,
DashboardNotFoundError,
DashboardRestoreFailedError,
DashboardSlugConflictError,
)
from superset.commands.dashboard.restore import RestoreDashboardCommand
from superset.daos.dashboard import DashboardDAO
summary: DeletedDashboardSummary | None = None
try:
with event_logger.log_context(action="mcp.restore_dashboard.validation"):
# Skip the visibility filter so the soft-deleted row is findable,
# and the base filter so an owner's own trash stays reachable;
# RestoreDashboardCommand re-checks ownership before restoring.
dashboard = DashboardDAO.find_by_id(
request.dashboard_id,
skip_base_filter=True,
skip_visibility_filter=True,
)
if not dashboard:
return RestoreDashboardResponse(
restored=False,
dashboard=None,
error=(
f"Dashboard with ID {request.dashboard_id} not found. "
"Hard-deleted dashboards cannot be restored."
),
)
summary = DeletedDashboardSummary(
id=dashboard.id,
dashboard_title=dashboard.dashboard_title,
slug=dashboard.slug,
uuid=str(dashboard.uuid) if dashboard.uuid else None,
)
if dashboard.deleted_at is None:
return RestoreDashboardResponse(
restored=False,
dashboard=summary,
error=(
f"Dashboard with ID {request.dashboard_id} is not "
"deleted; nothing to restore."
),
)
with event_logger.log_context(action="mcp.restore_dashboard.restore"):
RestoreDashboardCommand(str(dashboard.uuid)).run()
logger.info("Restored dashboard %s", request.dashboard_id)
await ctx.info("Restored dashboard %s" % (request.dashboard_id,))
return RestoreDashboardResponse(restored=True, dashboard=summary, error=None)
except DashboardNotFoundError:
return RestoreDashboardResponse(
restored=False,
dashboard=summary,
error=(
f"Dashboard with ID {request.dashboard_id} not found. "
"Hard-deleted dashboards cannot be restored."
),
)
except DashboardForbiddenError:
await ctx.warning(
"Permission denied restoring dashboard %s" % (request.dashboard_id,)
)
return RestoreDashboardResponse(
restored=False,
dashboard=summary,
error=(
f"You don't have permission to restore dashboard "
f"with ID {request.dashboard_id}. Only owners or admins "
"can restore a deleted dashboard."
),
)
except DashboardSlugConflictError:
return RestoreDashboardResponse(
restored=False,
dashboard=summary,
error=(
f"Dashboard with ID {request.dashboard_id} cannot be restored "
"because its slug is now used by another active dashboard. "
"Rename or delete the dashboard currently using the slug, "
"then retry."
),
)
except DashboardRestoreFailedError as exc:
await ctx.error(
"Failed to restore dashboard %s: %s" % (request.dashboard_id, str(exc))
)
return RestoreDashboardResponse(
restored=False,
dashboard=summary,
error=(
f"Failed to restore dashboard {request.dashboard_id}: {exc.message}"
),
)
except Exception as exc:
await ctx.error(
"Unexpected error restoring dashboard: %s: %s"
% (type(exc).__name__, str(exc))
)
raise

View File

@@ -32,9 +32,12 @@ from superset.mcp_service.dashboard.schemas import (
_extract_native_filters, _extract_native_filters,
_safe_user_label, _safe_user_label,
dashboard_serializer, dashboard_serializer,
DeleteDashboardResponse,
DeletedDashboardSummary,
DuplicateDashboardRequest, DuplicateDashboardRequest,
DuplicateDashboardResponse, DuplicateDashboardResponse,
GenerateDashboardRequest, GenerateDashboardRequest,
RestoreDashboardResponse,
serialize_chart_summary, serialize_chart_summary,
serialize_dashboard_object, serialize_dashboard_object,
UpdateDashboardRequest, UpdateDashboardRequest,
@@ -904,3 +907,69 @@ class TestDuplicateDashboardResponse:
"""A null error stays null rather than being wrapped.""" """A null error stays null rather than being wrapped."""
resp = DuplicateDashboardResponse(dashboard_url="http://host/d/1/") resp = DuplicateDashboardResponse(dashboard_url="http://host/d/1/")
assert resp.error is None assert resp.error is None
class TestDeletedDashboardSummary:
"""LLM-context sanitization for the delete/restore dashboard summary."""
def test_title_and_slug_are_wrapped_for_llm_context(self) -> None:
"""Dashboard-controlled title and slug are wrapped before exposure."""
summary = DeletedDashboardSummary(
id=1,
dashboard_title="Injected title",
slug="injected-slug",
uuid="dashboard-uuid-1",
)
assert summary.dashboard_title == _wrapped("Injected title")
assert summary.slug == _wrapped("injected-slug")
assert summary.uuid == "dashboard-uuid-1"
def test_none_fields_are_not_wrapped(self) -> None:
"""Null title/slug stay null rather than being wrapped."""
summary = DeletedDashboardSummary(id=1)
assert summary.dashboard_title is None
assert summary.slug is None
assert summary.uuid is None
class TestDeleteDashboardResponse:
"""Serialization and error sanitization for DeleteDashboardResponse."""
def test_defaults(self) -> None:
"""An empty response has no deletion flags set and null fields."""
resp = DeleteDashboardResponse()
assert resp.deleted is False
assert resp.permanent is False
assert resp.dashboard is None
assert resp.error is None
def test_error_is_wrapped_for_llm_context(self) -> None:
"""Error text is wrapped in LLM-context delimiters before exposure."""
resp = DeleteDashboardResponse(error="Dashboard 'x' not found.")
assert resp.error == _wrapped("Dashboard 'x' not found.")
def test_none_error_is_not_wrapped(self) -> None:
"""A null error stays null rather than being wrapped."""
resp = DeleteDashboardResponse(deleted=True)
assert resp.error is None
class TestRestoreDashboardResponse:
"""Serialization and error sanitization for RestoreDashboardResponse."""
def test_defaults(self) -> None:
"""An empty response has restored=False and null fields."""
resp = RestoreDashboardResponse()
assert resp.restored is False
assert resp.dashboard is None
assert resp.error is None
def test_error_is_wrapped_for_llm_context(self) -> None:
"""Error text is wrapped in LLM-context delimiters before exposure."""
resp = RestoreDashboardResponse(error="Slug 'x' already in use.")
assert resp.error == _wrapped("Slug 'x' already in use.")
def test_none_error_is_not_wrapped(self) -> None:
"""A null error stays null rather than being wrapped."""
resp = RestoreDashboardResponse(restored=True)
assert resp.error is None

View File

@@ -0,0 +1,197 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Unit tests for the delete_dashboard MCP tool.
Covers:
- Successful delete with SOFT_DELETE on (recoverable) and off (permanent)
- confirm=false refusal (safety gate)
- Dashboard not found
- Permission denied (user does not own the dashboard)
- Delete failure (e.g. associated alerts or reports)
"""
from collections.abc import Iterator
from unittest.mock import Mock, patch
import pytest
from fastmcp import Client
from superset.mcp_service.app import mcp
@pytest.fixture
def mcp_server() -> object:
"""Return the FastMCP app instance for use in MCP client tests."""
return mcp
@pytest.fixture(autouse=True)
def mock_auth() -> Iterator[Mock]:
"""Mock authentication for all tests."""
with patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user:
mock_user = Mock()
mock_user.id = 1
mock_user.username = "admin"
mock_get_user.return_value = mock_user
yield mock_get_user
def _mock_dashboard(id: int = 1, title: str = "Sales Dashboard") -> Mock:
"""Create a minimal mock Dashboard object."""
dashboard = Mock()
dashboard.id = id
dashboard.dashboard_title = title
dashboard.slug = f"test-dashboard-{id}"
dashboard.uuid = f"dashboard-uuid-{id}"
return dashboard
@pytest.mark.parametrize("soft_delete_enabled", [True, False])
@patch("superset.commands.dashboard.delete.DeleteDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_successful_delete(
mock_find_by_id: Mock,
mock_delete_cmd_cls: Mock,
mcp_server: object,
soft_delete_enabled: bool,
) -> None:
"""Happy path: dashboard deleted, summary echoed back, and the
``permanent`` field reflects whether SOFT_DELETE is enabled."""
mock_find_by_id.return_value = _mock_dashboard(id=1, title="Sales Dashboard")
mock_delete_cmd = Mock()
mock_delete_cmd.run.return_value = None
mock_delete_cmd_cls.return_value = mock_delete_cmd
with patch(
"superset.is_feature_enabled",
side_effect=lambda flag: flag == "SOFT_DELETE" and soft_delete_enabled,
):
async with Client(mcp_server) as client:
result = await client.call_tool(
"delete_dashboard",
{"request": {"dashboard_id": 1, "confirm": True}},
)
content = result.structured_content
assert content["deleted"] is True
assert content["permanent"] is (not soft_delete_enabled)
assert content["error"] is None
assert content["dashboard"]["id"] == 1
assert "Sales Dashboard" in content["dashboard"]["dashboard_title"]
assert "test-dashboard-1" in content["dashboard"]["slug"]
assert content["dashboard"]["uuid"] == "dashboard-uuid-1"
mock_delete_cmd_cls.assert_called_once_with([1])
mock_delete_cmd.run.assert_called_once()
@patch("superset.commands.dashboard.delete.DeleteDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_not_confirmed_refusal(
mock_find_by_id: Mock, mock_delete_cmd_cls: Mock, mcp_server: object
) -> None:
"""confirm=false: the tool refuses and nothing is deleted."""
async with Client(mcp_server) as client:
result = await client.call_tool(
"delete_dashboard",
{"request": {"dashboard_id": 1, "confirm": False}},
)
content = result.structured_content
assert content["deleted"] is False
assert content["dashboard"] is None
assert "confirm" in (content["error"] or "").lower()
mock_find_by_id.assert_not_called()
mock_delete_cmd_cls.assert_not_called()
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_dashboard_not_found(mock_find_by_id: Mock, mcp_server: object) -> None:
"""Returns a clear error when the target dashboard does not exist."""
mock_find_by_id.return_value = None
async with Client(mcp_server) as client:
result = await client.call_tool(
"delete_dashboard",
{"request": {"dashboard_id": 999, "confirm": True}},
)
content = result.structured_content
assert content["deleted"] is False
assert content["dashboard"] is None
assert "not found" in (content["error"] or "").lower()
@patch("superset.commands.dashboard.delete.DeleteDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_permission_denied(
mock_find_by_id: Mock, mock_delete_cmd_cls: Mock, mcp_server: object
) -> None:
"""Returns a structured error when the user cannot delete the dashboard."""
from superset.commands.dashboard.exceptions import DashboardForbiddenError
mock_find_by_id.return_value = _mock_dashboard(id=1, title="Sales Dashboard")
mock_delete_cmd = Mock()
mock_delete_cmd.run.side_effect = DashboardForbiddenError()
mock_delete_cmd_cls.return_value = mock_delete_cmd
async with Client(mcp_server) as client:
result = await client.call_tool(
"delete_dashboard",
{"request": {"dashboard_id": 1, "confirm": True}},
)
content = result.structured_content
assert content["deleted"] is False
assert "permission" in (content["error"] or "").lower()
assert content["dashboard"]["id"] == 1
@patch("superset.commands.dashboard.delete.DeleteDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_delete_failed_reports_exist(
mock_find_by_id: Mock, mock_delete_cmd_cls: Mock, mcp_server: object
) -> None:
"""Returns a structured error when the delete command fails, e.g. because
the dashboard has associated alerts or reports."""
from superset.commands.dashboard.exceptions import (
DashboardDeleteFailedReportsExistError,
)
mock_find_by_id.return_value = _mock_dashboard(id=1, title="Sales Dashboard")
mock_delete_cmd = Mock()
mock_delete_cmd.run.side_effect = DashboardDeleteFailedReportsExistError()
mock_delete_cmd_cls.return_value = mock_delete_cmd
async with Client(mcp_server) as client:
result = await client.call_tool(
"delete_dashboard",
{"request": {"dashboard_id": 1, "confirm": True}},
)
content = result.structured_content
assert content["deleted"] is False
assert content["permanent"] is False
assert "failed to delete" in (content["error"] or "").lower()
assert "alerts or reports" in (content["error"] or "").lower()
assert content["dashboard"]["id"] == 1

View File

@@ -0,0 +1,220 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Unit tests for the restore_dashboard MCP tool.
Covers:
- Successful restore (happy path)
- Dashboard not found
- Dashboard exists but is not soft-deleted
- Permission denied (user does not own the dashboard)
- Slug conflict with an active dashboard
- Generic restore failure (e.g. database error during commit)
"""
from collections.abc import Iterator
from datetime import datetime
from unittest.mock import Mock, patch
import pytest
from fastmcp import Client
from superset.mcp_service.app import mcp
@pytest.fixture
def mcp_server() -> object:
"""Return the FastMCP app instance for use in MCP client tests."""
return mcp
@pytest.fixture(autouse=True)
def mock_auth() -> Iterator[Mock]:
"""Mock authentication for all tests."""
with patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user:
mock_user = Mock()
mock_user.id = 1
mock_user.username = "admin"
mock_get_user.return_value = mock_user
yield mock_get_user
def _mock_dashboard(
id: int = 1,
title: str = "Sales Dashboard",
deleted: bool = True,
) -> Mock:
"""Create a minimal mock Dashboard object, soft-deleted by default."""
dashboard = Mock()
dashboard.id = id
dashboard.dashboard_title = title
dashboard.slug = f"test-dashboard-{id}"
dashboard.uuid = f"dashboard-uuid-{id}"
dashboard.deleted_at = datetime(2026, 1, 1) if deleted else None
return dashboard
@patch("superset.commands.dashboard.restore.RestoreDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_successful_restore(
mock_find_by_id: Mock, mock_restore_cmd_cls: Mock, mcp_server: object
) -> None:
"""Happy path: soft-deleted dashboard restored, summary echoed back."""
mock_find_by_id.return_value = _mock_dashboard(id=1, title="Sales Dashboard")
mock_restore_cmd = Mock()
mock_restore_cmd.run.return_value = None
mock_restore_cmd_cls.return_value = mock_restore_cmd
async with Client(mcp_server) as client:
result = await client.call_tool(
"restore_dashboard",
{"request": {"dashboard_id": 1}},
)
content = result.structured_content
assert content["restored"] is True
assert content["error"] is None
assert content["dashboard"]["id"] == 1
assert "Sales Dashboard" in content["dashboard"]["dashboard_title"]
assert content["dashboard"]["uuid"] == "dashboard-uuid-1"
mock_find_by_id.assert_called_once_with(
1, skip_base_filter=True, skip_visibility_filter=True
)
mock_restore_cmd_cls.assert_called_once_with("dashboard-uuid-1")
mock_restore_cmd.run.assert_called_once()
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_dashboard_not_found(mock_find_by_id: Mock, mcp_server: object) -> None:
"""Returns a clear error when the dashboard row does not exist at all."""
mock_find_by_id.return_value = None
async with Client(mcp_server) as client:
result = await client.call_tool(
"restore_dashboard",
{"request": {"dashboard_id": 999}},
)
content = result.structured_content
assert content["restored"] is False
assert content["dashboard"] is None
assert "not found" in (content["error"] or "").lower()
@patch("superset.commands.dashboard.restore.RestoreDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_dashboard_not_deleted(
mock_find_by_id: Mock, mock_restore_cmd_cls: Mock, mcp_server: object
) -> None:
"""Returns an error when the dashboard exists but is not soft-deleted."""
mock_find_by_id.return_value = _mock_dashboard(id=1, deleted=False)
async with Client(mcp_server) as client:
result = await client.call_tool(
"restore_dashboard",
{"request": {"dashboard_id": 1}},
)
content = result.structured_content
assert content["restored"] is False
assert content["dashboard"]["id"] == 1
assert "not" in (content["error"] or "").lower()
assert "restore" in (content["error"] or "").lower()
mock_restore_cmd_cls.assert_not_called()
@patch("superset.commands.dashboard.restore.RestoreDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_permission_denied(
mock_find_by_id: Mock, mock_restore_cmd_cls: Mock, mcp_server: object
) -> None:
"""Returns a structured error when the user cannot restore the dashboard."""
from superset.commands.dashboard.exceptions import DashboardForbiddenError
mock_find_by_id.return_value = _mock_dashboard(id=1)
mock_restore_cmd = Mock()
mock_restore_cmd.run.side_effect = DashboardForbiddenError()
mock_restore_cmd_cls.return_value = mock_restore_cmd
async with Client(mcp_server) as client:
result = await client.call_tool(
"restore_dashboard",
{"request": {"dashboard_id": 1}},
)
content = result.structured_content
assert content["restored"] is False
assert "permission" in (content["error"] or "").lower()
assert content["dashboard"]["id"] == 1
@patch("superset.commands.dashboard.restore.RestoreDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_slug_conflict(
mock_find_by_id: Mock, mock_restore_cmd_cls: Mock, mcp_server: object
) -> None:
"""Returns an actionable error when an active dashboard claimed the slug."""
from superset.commands.dashboard.exceptions import DashboardSlugConflictError
mock_find_by_id.return_value = _mock_dashboard(id=1)
mock_restore_cmd = Mock()
mock_restore_cmd.run.side_effect = DashboardSlugConflictError()
mock_restore_cmd_cls.return_value = mock_restore_cmd
async with Client(mcp_server) as client:
result = await client.call_tool(
"restore_dashboard",
{"request": {"dashboard_id": 1}},
)
content = result.structured_content
assert content["restored"] is False
assert "slug" in (content["error"] or "").lower()
assert content["dashboard"]["id"] == 1
@patch("superset.commands.dashboard.restore.RestoreDashboardCommand")
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@pytest.mark.asyncio
async def test_restore_failed(
mock_find_by_id: Mock, mock_restore_cmd_cls: Mock, mcp_server: object
) -> None:
"""Returns a structured error when the restore command fails, e.g. due to
a database error during commit."""
from superset.commands.dashboard.exceptions import DashboardRestoreFailedError
mock_find_by_id.return_value = _mock_dashboard(id=1)
mock_restore_cmd = Mock()
mock_restore_cmd.run.side_effect = DashboardRestoreFailedError()
mock_restore_cmd_cls.return_value = mock_restore_cmd
async with Client(mcp_server) as client:
result = await client.call_tool(
"restore_dashboard",
{"request": {"dashboard_id": 1}},
)
content = result.structured_content
assert content["restored"] is False
assert "failed to restore" in (content["error"] or "").lower()
assert content["dashboard"]["id"] == 1