You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by mi...@apache.org on 2026/06/11 00:06:28 UTC
(superset) 01/01: feat(mcp): add duplicate_dashboard tool
This is an automated email from the ASF dual-hosted git repository.
aminghadersohi pushed a commit to branch mcp-duplicate-dashboard
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 5495f989f38430141863ce7f403d7c6327b4d15d
Author: Amin Ghadersohi <am...@gmail.com>
AuthorDate: Thu Jun 11 00:06:12 2026 +0000
feat(mcp): add duplicate_dashboard tool
Adds a duplicate_dashboard MCP tool that clones an existing dashboard
via CopyDashboardCommand. The source dashboard can be identified by
numeric ID, UUID, or slug. By default the copy references the same
charts; duplicate_slices=true deep-copies every chart into new objects
owned by the caller.
The tool builds the required json_metadata payload (source metadata
plus a positions key from position_json), mirroring what the frontend
"Save as" flow sends to the /copy/ endpoint. The new title is
sanitized for XSS, and the tool is excluded from MCP response caching.
---
superset/mcp_service/app.py | 7 +-
superset/mcp_service/dashboard/schemas.py | 132 ++++++++
superset/mcp_service/dashboard/tool/__init__.py | 2 +
.../dashboard/tool/duplicate_dashboard.py | 269 +++++++++++++++
superset/mcp_service/mcp_config.py | 1 +
.../dashboard/tool/test_duplicate_dashboard.py | 361 +++++++++++++++++++++
6 files changed, 770 insertions(+), 2 deletions(-)
diff --git a/superset/mcp_service/app.py b/superset/mcp_service/app.py
index 04c0eb0aebf..bd1e6830a5c 100644
--- a/superset/mcp_service/app.py
+++ b/superset/mcp_service/app.py
@@ -129,6 +129,7 @@ Dashboard Management:
- get_dashboard_info: Get detailed dashboard information by ID
- get_dashboard_layout: Get parsed tabs and chart positions for a dashboard (companion to get_dashboard_info when its omitted_fields hint flags position_json)
- generate_dashboard: Create a dashboard from chart IDs (requires write access)
+- duplicate_dashboard: Duplicate an existing dashboard, optionally deep-copying its charts (requires write access)
- add_chart_to_existing_dashboard: Add a chart to an existing dashboard (requires write access)
Annotation Layers:
@@ -413,8 +414,9 @@ Input format:
{_feature_availability}Permission Awareness:
{_instance_info_role_bullet}- ALWAYS check the user's roles BEFORE suggesting write operations (creating datasets,
charts, or dashboards). SQL execution is a separate permission — see execute_sql below.
-- Write tools (generate_chart, generate_dashboard, update_chart, create_virtual_dataset,
- save_sql_query, add_chart_to_existing_dashboard, update_chart_preview) require write
+- Write tools (generate_chart, generate_dashboard, duplicate_dashboard, update_chart,
+ create_virtual_dataset, save_sql_query, add_chart_to_existing_dashboard,
+ update_chart_preview) require write
permissions. These tools are only listed for users who have the necessary access.
If a write tool does not appear in the tool list, the current user lacks write access.
- execute_sql requires SQL Lab access (execute_sql_query permission), which is separate
@@ -679,6 +681,7 @@ from superset.mcp_service.chart.tool import ( # noqa: F401, E402
)
from superset.mcp_service.dashboard.tool import ( # noqa: F401, E402
add_chart_to_existing_dashboard,
+ duplicate_dashboard,
generate_dashboard,
get_dashboard_info,
get_dashboard_layout,
diff --git a/superset/mcp_service/dashboard/schemas.py b/superset/mcp_service/dashboard/schemas.py
index 73a57a6e292..8a675d367b3 100644
--- a/superset/mcp_service/dashboard/schemas.py
+++ b/superset/mcp_service/dashboard/schemas.py
@@ -708,6 +708,138 @@ class GenerateDashboardResponse(BaseModel):
)
+class DuplicateDashboardRequest(BaseModel):
+ """Request schema for duplicating an existing dashboard."""
+
+ model_config = ConfigDict(populate_by_name=True)
+
+ dashboard_id: Annotated[
+ int | str,
+ Field(
+ description=(
+ "Source dashboard identifier - can be numeric ID, UUID string, or slug"
+ )
+ ),
+ ]
+ dashboard_title: str = Field(
+ ...,
+ description="Title for the new (duplicated) dashboard",
+ validation_alias=AliasChoices("dashboard_title", "title", "name"),
+ )
+ duplicate_slices: bool = Field(
+ default=False,
+ description=(
+ "When true, every chart on the source dashboard is deep-copied "
+ "into a new chart object owned by the caller. When false "
+ "(default), the new dashboard references the same charts as the "
+ "source."
+ ),
+ )
+ sanitization_warnings: List[str] = Field(
+ default_factory=list,
+ description=(
+ "Internal: warnings emitted when user input was altered by "
+ "sanitization. Populated by the ``mode='before'`` validator "
+ "before dashboard_title is rewritten, so the tool can surface "
+ "a notice to the caller instead of silently dropping content."
+ ),
+ )
+
+ @model_validator(mode="before")
+ @classmethod
+ def _detect_dashboard_title_sanitization(cls, data: Any) -> Any:
+ """Reject empty-after-sanitization titles and warn on partial strip.
+
+ Runs before the ``dashboard_title`` field validator rewrites the
+ value. If the caller supplied a title that sanitization would strip
+ entirely (XSS-only content), we raise so the caller gets a clear
+ error instead of a blank-titled dashboard. When the sanitizer only
+ trims part of the title, we record a warning the tool can return
+ alongside the successful result.
+
+ ``sanitization_warnings`` is a server-only field — any value the
+ caller supplied is discarded here so the tool cannot be tricked
+ into echoing attacker-controlled text back through the response.
+ """
+ if not isinstance(data, dict):
+ return data
+ data["sanitization_warnings"] = []
+ for key in ("dashboard_title", "title", "name"):
+ if key in data:
+ raw = data[key]
+ break
+ else:
+ raw = None
+ if not isinstance(raw, str) or not raw.strip():
+ return data
+ sanitized, was_modified = sanitize_user_input_with_changes(
+ raw, "Dashboard title", max_length=500, allow_empty=True
+ )
+ if was_modified and not sanitized:
+ raise ValueError(
+ "dashboard_title contained only disallowed content "
+ "(HTML/script/URL schemes) and was removed entirely by "
+ "sanitization. Provide a dashboard_title with plain text."
+ )
+ if was_modified:
+ data["sanitization_warnings"].append(
+ "dashboard_title was modified during sanitization to "
+ "remove potentially unsafe content; the stored title "
+ "differs from the input."
+ )
+ return data
+
+ @field_validator("dashboard_title")
+ @classmethod
+ def sanitize_dashboard_title(cls, v: str) -> str:
+ """Sanitize dashboard title to prevent XSS."""
+ sanitized = sanitize_user_input(
+ v, "Dashboard title", max_length=500, allow_empty=True
+ )
+ if not sanitized:
+ raise ValueError("dashboard_title cannot be empty")
+ return sanitized
+
+
+class DuplicateDashboardResponse(BaseModel):
+ """Response schema for dashboard duplication."""
+
+ dashboard: DashboardInfo | None = Field(
+ None, description="The newly created dashboard info, if successful"
+ )
+ dashboard_url: str | None = Field(None, description="URL to view the new dashboard")
+ duplicated_slices: bool = Field(
+ default=False,
+ description=(
+ "True when the source dashboard's charts were deep-copied into "
+ "new chart objects; False when the new dashboard references the "
+ "original charts."
+ ),
+ )
+ error: str | None = Field(None, description="Error message, if duplication failed")
+ warnings: List[str] = Field(
+ default_factory=list,
+ description=(
+ "Non-fatal advisory messages about the duplicated dashboard — "
+ "for example, that the supplied title was altered by "
+ "sanitization."
+ ),
+ )
+
+ @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 dashboard-controlled content such as the source
+ dashboard title — wrap it 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 ChartPosition(BaseModel):
"""Position and identity of a chart within a dashboard layout."""
diff --git a/superset/mcp_service/dashboard/tool/__init__.py b/superset/mcp_service/dashboard/tool/__init__.py
index 389acfb192a..e17b7136c58 100644
--- a/superset/mcp_service/dashboard/tool/__init__.py
+++ b/superset/mcp_service/dashboard/tool/__init__.py
@@ -16,6 +16,7 @@
# under the License.
from .add_chart_to_existing_dashboard import add_chart_to_existing_dashboard
+from .duplicate_dashboard import duplicate_dashboard
from .generate_dashboard import generate_dashboard
from .get_dashboard_info import get_dashboard_info
from .get_dashboard_layout import get_dashboard_layout
@@ -26,5 +27,6 @@ __all__ = [
"get_dashboard_info",
"get_dashboard_layout",
"generate_dashboard",
+ "duplicate_dashboard",
"add_chart_to_existing_dashboard",
]
diff --git a/superset/mcp_service/dashboard/tool/duplicate_dashboard.py b/superset/mcp_service/dashboard/tool/duplicate_dashboard.py
new file mode 100644
index 00000000000..8a0e803f414
--- /dev/null
+++ b/superset/mcp_service/dashboard/tool/duplicate_dashboard.py
@@ -0,0 +1,269 @@
+# 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: duplicate_dashboard
+
+Duplicates an existing dashboard, optionally deep-copying its charts.
+Canonical workflow: clone a template dashboard, then edit the copy
+(e.g. to create a regional or staging variant).
+"""
+
+import logging
+from typing import Any
+
+from fastmcp import Context
+from sqlalchemy.exc import SQLAlchemyError
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.dashboard.schemas import (
+ DashboardInfo,
+ DuplicateDashboardRequest,
+ DuplicateDashboardResponse,
+ serialize_chart_summary,
+)
+from superset.mcp_service.privacy import user_can_view_data_model_metadata
+from superset.mcp_service.utils.url_utils import get_superset_base_url
+from superset.utils import json
+
+logger = logging.getLogger(__name__)
+
+
+def _build_copy_payload(
+ source: Any, dashboard_title: str, duplicate_slices: bool
+) -> dict[str, Any]:
+ """Build the data payload expected by ``CopyDashboardCommand``.
+
+ Mirrors what the frontend "Save as" flow sends to the
+ ``/api/v1/dashboard/<id>/copy/`` endpoint: the source dashboard's
+ current ``json_metadata`` with a ``positions`` key holding the current
+ layout (``position_json``). ``DashboardCopySchema`` requires
+ ``json_metadata``, and ``DashboardDAO.copy_dashboard`` reads
+ ``positions`` from it to remap chart IDs when ``duplicate_slices``
+ is enabled.
+ """
+ try:
+ metadata = json.loads(source.json_metadata or "{}")
+ except (json.JSONDecodeError, TypeError):
+ metadata = {}
+ if not isinstance(metadata, dict):
+ metadata = {}
+
+ try:
+ positions = json.loads(source.position_json or "{}")
+ except (json.JSONDecodeError, TypeError):
+ positions = {}
+ if not isinstance(positions, dict):
+ positions = {}
+
+ metadata["positions"] = positions
+
+ return {
+ "dashboard_title": dashboard_title,
+ "css": source.css,
+ "duplicate_slices": duplicate_slices,
+ "json_metadata": json.dumps(metadata),
+ }
+
+
+def _serialize_new_dashboard(dashboard: Any) -> tuple[DashboardInfo, str]:
+ """Build the response ``DashboardInfo`` and URL for the new dashboard."""
+ from superset.mcp_service.dashboard.schemas import serialize_tag_object
+
+ dashboard_url = f"{get_superset_base_url()}/superset/dashboard/{dashboard.id}/"
+ include_data_model_metadata = user_can_view_data_model_metadata()
+ info = DashboardInfo(
+ id=dashboard.id,
+ dashboard_title=dashboard.dashboard_title,
+ slug=dashboard.slug,
+ description=dashboard.description,
+ published=dashboard.published,
+ created_on=dashboard.created_on,
+ changed_on=dashboard.changed_on,
+ uuid=str(dashboard.uuid) if dashboard.uuid else None,
+ url=dashboard_url,
+ chart_count=len(dashboard.slices),
+ tags=[
+ obj
+ for tag in getattr(dashboard, "tags", [])
+ if (obj := serialize_tag_object(tag)) is not None
+ ],
+ charts=[
+ obj
+ for chart in getattr(dashboard, "slices", [])
+ if (
+ obj := serialize_chart_summary(
+ chart,
+ include_data_model_metadata=include_data_model_metadata,
+ )
+ )
+ is not None
+ ],
+ )
+ return info, dashboard_url
+
+
+@tool(
+ tags=["mutate"],
+ class_permission_name="Dashboard",
+ method_permission_name="write",
+ annotations=ToolAnnotations(
+ title="Duplicate dashboard",
+ readOnlyHint=False,
+ destructiveHint=False,
+ ),
+)
+async def duplicate_dashboard(
+ request: DuplicateDashboardRequest, ctx: Context
+) -> DuplicateDashboardResponse:
+ """
+ Duplicate an existing dashboard under a new title.
+
+ By default the copy references the same charts as the source.
+ Set duplicate_slices=true to also deep-copy every chart into new
+ chart objects owned by you, so edits to the copies never affect
+ the originals.
+
+ The source dashboard can be identified by numeric ID, UUID, or slug.
+ Returns the new dashboard's ID, title, and URL.
+ """
+ await ctx.info(
+ "Duplicating dashboard: dashboard_id=%s, duplicate_slices=%s"
+ % (request.dashboard_id, request.duplicate_slices)
+ )
+
+ from superset.commands.dashboard.copy import CopyDashboardCommand
+ from superset.commands.dashboard.exceptions import (
+ DashboardAccessDeniedError,
+ DashboardCopyError,
+ DashboardForbiddenError,
+ DashboardInvalidError,
+ DashboardNotFoundError,
+ )
+ from superset.daos.dashboard import DashboardDAO
+
+ try:
+ with event_logger.log_context(action="mcp.duplicate_dashboard.lookup"):
+ try:
+ source = DashboardDAO.get_by_id_or_slug(str(request.dashboard_id))
+ except DashboardNotFoundError:
+ return DuplicateDashboardResponse(
+ error=(
+ f"Dashboard '{request.dashboard_id}' not found. "
+ "Use list_dashboards to get valid dashboard IDs."
+ ),
+ )
+ except DashboardAccessDeniedError:
+ return DuplicateDashboardResponse(
+ error=(
+ f"You don't have access to dashboard "
+ f"'{request.dashboard_id}', so it cannot be duplicated."
+ ),
+ )
+
+ data = _build_copy_payload(
+ source, request.dashboard_title, request.duplicate_slices
+ )
+
+ with event_logger.log_context(action="mcp.duplicate_dashboard.copy"):
+ new_dashboard = CopyDashboardCommand(source, data).run()
+
+ # Re-fetch with eager-loaded relationships to avoid lazy-loading on
+ # a session that the command's commit may have invalidated.
+ from sqlalchemy.orm import subqueryload
+
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ try:
+ new_dashboard = (
+ DashboardDAO.find_by_id(
+ new_dashboard.id,
+ query_options=[
+ subqueryload(Dashboard.slices).subqueryload(Slice.tags),
+ subqueryload(Dashboard.tags),
+ ],
+ )
+ or new_dashboard
+ )
+ info, dashboard_url = _serialize_new_dashboard(new_dashboard)
+ except SQLAlchemyError:
+ logger.warning(
+ "Re-fetch of dashboard %s failed; returning minimal response",
+ new_dashboard.id,
+ exc_info=True,
+ )
+ dashboard_url = (
+ f"{get_superset_base_url()}/superset/dashboard/{new_dashboard.id}/"
+ )
+ info = DashboardInfo(
+ id=new_dashboard.id,
+ dashboard_title=request.dashboard_title,
+ url=dashboard_url,
+ )
+
+ logger.info(
+ "Duplicated dashboard %s into dashboard %s (duplicate_slices=%s)",
+ request.dashboard_id,
+ new_dashboard.id,
+ request.duplicate_slices,
+ )
+
+ return DuplicateDashboardResponse(
+ dashboard=info,
+ dashboard_url=dashboard_url,
+ duplicated_slices=request.duplicate_slices,
+ warnings=list(request.sanitization_warnings),
+ )
+
+ except DashboardForbiddenError:
+ await ctx.error(
+ "Dashboard duplication forbidden: dashboard_id=%s" % (request.dashboard_id,)
+ )
+ return DuplicateDashboardResponse(
+ error=(
+ f"You don't have permission to duplicate dashboard "
+ f"'{request.dashboard_id}'."
+ ),
+ )
+ except DashboardInvalidError:
+ return DuplicateDashboardResponse(
+ error=(
+ "Dashboard duplication parameters were invalid. "
+ "Provide a non-empty dashboard_title."
+ ),
+ )
+ except DashboardCopyError as exc:
+ from superset import db
+
+ try:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ except SQLAlchemyError:
+ logger.warning(
+ "Database rollback failed during error handling", exc_info=True
+ )
+ await ctx.error("Dashboard duplication failed: %s" % (str(exc),))
+ return DuplicateDashboardResponse(
+ error=f"Failed to duplicate dashboard: {exc}",
+ )
+ except Exception as exc:
+ await ctx.error(
+ "Unexpected error duplicating dashboard: %s: %s"
+ % (type(exc).__name__, str(exc))
+ )
+ raise
diff --git a/superset/mcp_service/mcp_config.py b/superset/mcp_service/mcp_config.py
index 33a4e45c142..0e5d2d79d62 100644
--- a/superset/mcp_service/mcp_config.py
+++ b/superset/mcp_service/mcp_config.py
@@ -212,6 +212,7 @@ MCP_CACHE_CONFIG: Dict[str, Any] = {
"excluded_tools": [ # Tools that should never be cached (side effects, dynamic)
"execute_sql",
"generate_dashboard",
+ "duplicate_dashboard",
"generate_chart",
"update_chart",
],
diff --git a/tests/unit_tests/mcp_service/dashboard/tool/test_duplicate_dashboard.py b/tests/unit_tests/mcp_service/dashboard/tool/test_duplicate_dashboard.py
new file mode 100644
index 00000000000..5cb02736d22
--- /dev/null
+++ b/tests/unit_tests/mcp_service/dashboard/tool/test_duplicate_dashboard.py
@@ -0,0 +1,361 @@
+# 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 duplicate_dashboard MCP tool.
+
+Follows the same pattern used in test_add_chart_to_existing_dashboard.py:
+- Tests run through the async MCP Client (not direct function calls)
+- Patches applied at source locations (superset.daos.dashboard.*,
+ superset.commands.dashboard.copy.*)
+- auth is mocked via the autouse mock_auth fixture
+
+Covers:
+- Duplicate referencing the same charts (duplicate_slices=False, default)
+- Duplicate with deep-copied charts (duplicate_slices=True)
+- Source dashboard not found
+- Source dashboard access denied / copy forbidden
+- Title sanitization (XSS stripped, XSS-only title rejected)
+"""
+
+import logging
+from unittest.mock import Mock, patch
+
+import pytest
+from fastmcp import Client
+
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@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():
+ """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
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+SOURCE_POSITIONS = {
+ "DASHBOARD_VERSION_KEY": "v2",
+ "ROOT_ID": {"children": ["GRID_ID"], "id": "ROOT_ID", "type": "ROOT"},
+ "GRID_ID": {
+ "children": ["CHART-10"],
+ "id": "GRID_ID",
+ "parents": ["ROOT_ID"],
+ "type": "GRID",
+ },
+ "CHART-10": {
+ "children": [],
+ "id": "CHART-10",
+ "meta": {"chartId": 10, "height": 50, "width": 4},
+ "parents": ["ROOT_ID", "GRID_ID"],
+ "type": "CHART",
+ },
+}
+
+
+def _mock_chart(id: int = 10, slice_name: str = "Test Chart") -> Mock:
+ """Create a minimal mock Slice object with the given ID and name."""
+ chart = Mock()
+ chart.id = id
+ chart.slice_name = slice_name
+ chart.uuid = f"chart-uuid-{id}"
+ chart.tags = []
+ chart.owners = []
+ chart.viz_type = "table"
+ chart.datasource_name = None
+ chart.description = None
+ return chart
+
+
+def _mock_dashboard(
+ id: int = 1,
+ title: str = "Sales Dashboard",
+ slices: list[Mock] | None = None,
+ json_metadata: str | None = None,
+ position_json: str | None = None,
+) -> Mock:
+ """Create a minimal mock Dashboard object."""
+ dashboard = Mock()
+ dashboard.id = id
+ dashboard.dashboard_title = title
+ dashboard.slug = f"test-dashboard-{id}"
+ dashboard.description = None
+ dashboard.published = True
+ dashboard.created_on = None
+ dashboard.changed_on = None
+ dashboard.uuid = f"dashboard-uuid-{id}"
+ dashboard.slices = slices or []
+ dashboard.owners = []
+ dashboard.tags = []
+ dashboard.roles = []
+ dashboard.position_json = position_json or json.dumps(SOURCE_POSITIONS)
+ dashboard.json_metadata = json_metadata
+ dashboard.css = None
+ dashboard.certified_by = None
+ dashboard.certification_details = None
+ dashboard.is_managed_externally = False
+ dashboard.external_url = None
+ return dashboard
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_duplicate_referencing_same_charts(
+ mock_get_by_id_or_slug: Mock,
+ mock_copy_cmd_cls: Mock,
+ mock_find_by_id: Mock,
+ mcp_server: object,
+) -> None:
+ """Happy path: the copy references the same charts (default)."""
+ chart = _mock_chart(id=10)
+ source = _mock_dashboard(
+ id=1,
+ slices=[chart],
+ json_metadata=json.dumps({"color_scheme": "supersetColors"}),
+ )
+ new_dashboard = _mock_dashboard(id=2, title="Staging Copy", slices=[chart])
+
+ mock_get_by_id_or_slug.return_value = source
+ mock_copy_cmd_cls.return_value.run.return_value = new_dashboard
+ mock_find_by_id.return_value = new_dashboard
+
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "duplicate_dashboard",
+ {"request": {"dashboard_id": 1, "dashboard_title": "Staging Copy"}},
+ )
+
+ content = result.structured_content
+ assert content["error"] is None
+ assert content["duplicated_slices"] is False
+ assert content["dashboard"]["id"] == 2
+ assert content["dashboard"]["dashboard_title"] == "Staging Copy"
+ assert "/superset/dashboard/2/" in content["dashboard_url"]
+
+ # The copy data contract must mirror what the frontend "Save as" sends:
+ # required json_metadata containing the source's metadata + positions.
+ mock_copy_cmd_cls.assert_called_once()
+ cmd_source, cmd_data = mock_copy_cmd_cls.call_args.args
+ assert cmd_source is source
+ assert cmd_data["dashboard_title"] == "Staging Copy"
+ assert cmd_data["duplicate_slices"] is False
+ assert cmd_data["css"] is None
+ sent_metadata = json.loads(cmd_data["json_metadata"])
+ assert sent_metadata["color_scheme"] == "supersetColors"
+ assert sent_metadata["positions"] == SOURCE_POSITIONS
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_duplicate_with_duplicate_slices(
+ mock_get_by_id_or_slug: Mock,
+ mock_copy_cmd_cls: Mock,
+ mock_find_by_id: Mock,
+ mcp_server: object,
+) -> None:
+ """duplicate_slices=True is forwarded to the command and reported back."""
+ source = _mock_dashboard(id=1, slices=[_mock_chart(id=10)])
+ new_chart = _mock_chart(id=20)
+ new_dashboard = _mock_dashboard(id=3, title="Regional Variant", slices=[new_chart])
+
+ mock_get_by_id_or_slug.return_value = source
+ mock_copy_cmd_cls.return_value.run.return_value = new_dashboard
+ mock_find_by_id.return_value = new_dashboard
+
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "duplicate_dashboard",
+ {
+ "request": {
+ "dashboard_id": 1,
+ "dashboard_title": "Regional Variant",
+ "duplicate_slices": True,
+ }
+ },
+ )
+
+ content = result.structured_content
+ assert content["error"] is None
+ assert content["duplicated_slices"] is True
+ assert content["dashboard"]["id"] == 3
+ assert "/superset/dashboard/3/" in content["dashboard_url"]
+
+ _, cmd_data = mock_copy_cmd_cls.call_args.args
+ assert cmd_data["duplicate_slices"] is True
+ # positions must always be present in json_metadata: the DAO reads it to
+ # remap chart IDs when duplicating slices.
+ assert "positions" in json.loads(cmd_data["json_metadata"])
+
+
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_source_not_found(
+ mock_get_by_id_or_slug: Mock, mcp_server: object
+) -> None:
+ """Returns a clear error when the source dashboard does not exist."""
+ from superset.commands.dashboard.exceptions import DashboardNotFoundError
+
+ mock_get_by_id_or_slug.side_effect = DashboardNotFoundError()
+
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "duplicate_dashboard",
+ {"request": {"dashboard_id": 999, "dashboard_title": "Copy"}},
+ )
+
+ content = result.structured_content
+ assert content["dashboard"] is None
+ assert content["dashboard_url"] is None
+ assert "not found" in (content["error"] or "").lower()
+
+
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_source_access_denied(
+ mock_get_by_id_or_slug: Mock, mcp_server: object
+) -> None:
+ """Returns an error when the user cannot access the source dashboard."""
+ from superset.commands.dashboard.exceptions import DashboardAccessDeniedError
+
+ mock_get_by_id_or_slug.side_effect = DashboardAccessDeniedError()
+
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "duplicate_dashboard",
+ {"request": {"dashboard_id": 1, "dashboard_title": "Copy"}},
+ )
+
+ content = result.structured_content
+ assert content["dashboard"] is None
+ assert "access" in (content["error"] or "").lower()
+
+
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_copy_forbidden(
+ mock_get_by_id_or_slug: Mock,
+ mock_copy_cmd_cls: Mock,
+ mcp_server: object,
+) -> None:
+ """Returns an error when the copy command raises DashboardForbiddenError
+ (e.g. DASHBOARD_RBAC requires ownership of the source)."""
+ from superset.commands.dashboard.exceptions import DashboardForbiddenError
+
+ mock_get_by_id_or_slug.return_value = _mock_dashboard(id=1)
+ mock_copy_cmd_cls.return_value.run.side_effect = DashboardForbiddenError()
+
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "duplicate_dashboard",
+ {"request": {"dashboard_id": 1, "dashboard_title": "Copy"}},
+ )
+
+ content = result.structured_content
+ assert content["dashboard"] is None
+ assert "permission" in (content["error"] or "").lower()
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_title_xss_is_sanitized(
+ mock_get_by_id_or_slug: Mock,
+ mock_copy_cmd_cls: Mock,
+ mock_find_by_id: Mock,
+ mcp_server: object,
+) -> None:
+ """HTML/script content is stripped from the title and a warning surfaced."""
+ source = _mock_dashboard(id=1)
+ new_dashboard = _mock_dashboard(id=4, title="Regional Copy")
+
+ mock_get_by_id_or_slug.return_value = source
+ mock_copy_cmd_cls.return_value.run.return_value = new_dashboard
+ mock_find_by_id.return_value = new_dashboard
+
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "duplicate_dashboard",
+ {
+ "request": {
+ "dashboard_id": 1,
+ "dashboard_title": "<script>alert('x')</script>Regional Copy",
+ }
+ },
+ )
+
+ content = result.structured_content
+ assert content["error"] is None
+ # The sanitized title — not the raw payload — is sent to the command.
+ _, cmd_data = mock_copy_cmd_cls.call_args.args
+ assert cmd_data["dashboard_title"] == "Regional Copy"
+ assert content["warnings"], "expected a sanitization warning"
+
+
+def test_title_xss_only_rejected_by_schema() -> None:
+ """A title that sanitizes to nothing is rejected with a clear error."""
+ from pydantic import ValidationError
+
+ from superset.mcp_service.dashboard.schemas import DuplicateDashboardRequest
+
+ with pytest.raises(ValidationError):
+ DuplicateDashboardRequest(
+ dashboard_id=1, dashboard_title="<script>alert(1)</script>"
+ )
+
+
+def test_empty_title_rejected_by_schema() -> None:
+ """An empty title is rejected at the schema layer."""
+ from pydantic import ValidationError
+
+ from superset.mcp_service.dashboard.schemas import DuplicateDashboardRequest
+
+ with pytest.raises(ValidationError):
+ DuplicateDashboardRequest(dashboard_id=1, dashboard_title="")