Compare commits

...
Author SHA1 Message Date
Amin GhadersohiandClaude 8a01ea6fdc fix(mcp): match permalinks by uuid and keep identifier not-found errors
CreateDashboardPermalinkCommand stores dashboardId as str(dashboard.uuid),
so the int() comparison dropped shared state for virtually every real
permalink. The permalink-only path now records that it selected the
dashboard itself and skips re-verification, while the explicit
identifier + permalink_key combination compares the reference against the
dashboard id, uuid, and slug.

A plain identifier that simply does not exist also no longer reports the
permalink wording; its own not-found error is preserved and the permalink
message is reserved for requests that had to resolve through a permalink.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-18 18:50:32 +00:00
Amin Ghadersohi e076608022 Merge branch 'master' into aminghadersohi/resolve-dashboard-permalinks 2026-08-14 10:16:50 -06:00
Amin Ghadersohi e9891ae3c4 Merge branch 'master' into aminghadersohi/resolve-dashboard-permalinks 2026-08-14 00:46:23 -06:00
Amin Ghadersohi bbad000897 Merge branch 'master' into aminghadersohi/resolve-dashboard-permalinks 2026-08-14 00:05:50 -06:00
Amin Ghadersohi 5bd4149736 Merge branch 'master' into aminghadersohi/resolve-dashboard-permalinks 2026-08-13 13:34:27 -06:00
Amin Ghadersohi f8d92e5eee test(mcp): cover dashboard reference schemas 2026-08-13 16:18:59 +00:00
Amin Ghadersohi be9759d73f refactor(mcp): centralize dashboard permalink state 2026-08-13 16:18:59 +00:00
Amin Ghadersohi a7d0644277 fix(mcp): preserve dashboard identifier precedence 2026-08-13 16:18:59 +00:00
Amin Ghadersohi c8441f1789 fix(mcp): resolve dashboard permalinks 2026-08-13 16:18:59 +00:00
9 changed files with 823 additions and 231 deletions
+2 -2
View File
@@ -126,8 +126,8 @@ Available tools:
Dashboard Management:
- list_dashboards: List dashboards with advanced filters (1-based pagination; deleted_state='only'/'include' surfaces trashed dashboards the caller may restore)
- 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)
- get_dashboard_info: Resolve a dashboard by ID/UUID/slug or shared /dashboard/p/<key>/ permalink, including its active-tab and filter state
- get_dashboard_layout: Get parsed tabs and chart positions by dashboard identifier or shared permalink, including the permalink's active-tab and filter context
- get_dashboard_datasets: List the datasets used by a dashboard's charts, with columns and metrics (context for configuring native filters)
- generate_dashboard: Create a dashboard from chart IDs (requires write access)
- update_dashboard: Update an existing dashboard's title/description/slug/published/layout/theme/CSS (requires write access; editorship-checked per-instance)
+192
View File
@@ -0,0 +1,192 @@
# 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.
"""Helpers for resolving dashboard permalink keys and shared URLs."""
import logging
from dataclasses import dataclass
from typing import Callable, Generic, TypeVar
from urllib.parse import urlparse
from flask import g, has_request_context
from superset.commands.dashboard.exceptions import DashboardAccessDeniedError
from superset.commands.dashboard.permalink.get import GetDashboardPermalinkCommand
from superset.dashboards.permalink.exceptions import DashboardPermalinkGetFailedError
from superset.dashboards.permalink.types import DashboardPermalinkValue
from superset.mcp_service.auth import load_user_with_relationships
from superset.mcp_service.dashboard.schemas import (
redact_filter_state_data_model_metadata,
)
from superset.mcp_service.privacy import user_can_view_data_model_metadata
from superset.mcp_service.utils import sanitize_for_llm_context
logger = logging.getLogger(__name__)
LookupResultT = TypeVar("LookupResultT")
@dataclass(frozen=True)
class DashboardLookupResult(Generic[LookupResultT]):
"""Result of resolving either a dashboard identifier or permalink."""
result: LookupResultT | None
permalink_key: str | None = None
permalink_value: DashboardPermalinkValue | None = None
resolved_from_permalink: bool = False
"""True when the dashboard itself was selected from the permalink."""
@dataclass(frozen=True)
class DashboardPermalinkState:
"""Sanitized permalink state belonging to a resolved dashboard."""
key: str
state: dict[str, object]
def extract_dashboard_permalink_key(value: str) -> str:
"""Return a key from a dashboard permalink URL, or the bare input."""
path_parts = [part for part in urlparse(value).path.split("/") if part]
if len(path_parts) >= 3 and path_parts[-3:-1] == ["dashboard", "p"]:
return path_parts[-1]
return value
def refresh_request_user_for_permalink_access() -> None:
"""Reload the request user before permalink access checks."""
if not has_request_context() or not getattr(g, "user", None):
return
current_user = g.user
if getattr(current_user, "is_anonymous", False):
return
username = getattr(current_user, "username", None)
email = getattr(current_user, "email", None)
if not username and not email:
return
refreshed_user = (
load_user_with_relationships(username=username)
if username
else load_user_with_relationships(email=email)
)
if refreshed_user is not None:
g.user = refreshed_user
def get_dashboard_permalink(
key_or_url: str,
) -> tuple[str, DashboardPermalinkValue] | None:
"""Resolve a dashboard permalink key or shared URL, returning its state."""
key = extract_dashboard_permalink_key(key_or_url)
refresh_request_user_for_permalink_access()
try:
value = GetDashboardPermalinkCommand(key).run()
except (DashboardAccessDeniedError, DashboardPermalinkGetFailedError) as ex:
logger.info("Dashboard permalink could not be resolved: %s", ex)
return None
return (key, value) if value else None
def lookup_dashboard_reference(
*,
identifier: int | str | None,
permalink_key: str | None,
lookup: Callable[[int | str], LookupResultT],
is_found: Callable[[LookupResultT], bool],
) -> DashboardLookupResult[LookupResultT]:
"""Look up a dashboard while preserving identifier precedence.
A supplied identifier selects the dashboard and an explicit permalink only
contributes state. Shared permalink URLs and permalink-only requests select
the dashboard embedded in the permalink. Ambiguous bare strings use normal
identifier lookup first, then fall back to permalink resolution.
"""
key = permalink_key
identifier_is_permalink_url = False
if isinstance(identifier, str):
extracted_key = extract_dashboard_permalink_key(identifier)
identifier_is_permalink_url = extracted_key != identifier
if identifier_is_permalink_url:
key = extracted_key
if identifier is not None and not identifier_is_permalink_url:
result = lookup(identifier)
if is_found(result):
resolved = get_dashboard_permalink(key) if key else None
return DashboardLookupResult(
result=result,
permalink_key=resolved[0] if resolved else key,
permalink_value=resolved[1] if resolved else None,
)
if permalink_key is not None or not isinstance(identifier, str):
return DashboardLookupResult(result=result, permalink_key=key)
else:
result = None
reference = key or (identifier if isinstance(identifier, str) else None)
resolved = get_dashboard_permalink(reference) if reference else None
if resolved is None:
return DashboardLookupResult(result=result, permalink_key=reference)
key, value = resolved
return DashboardLookupResult(
result=lookup(value["dashboardId"]),
permalink_key=key,
permalink_value=value,
resolved_from_permalink=True,
)
def get_matching_dashboard_permalink_state(
lookup_result: DashboardLookupResult[LookupResultT],
dashboard_id: int | None,
dashboard_uuid: str | None = None,
dashboard_slug: str | None = None,
) -> DashboardPermalinkState | None:
"""Return sanitized permalink state when it belongs to the dashboard.
``CreateDashboardPermalinkCommand`` stores ``dashboardId`` as the dashboard
UUID string, while older permalinks may hold a numeric ID or a slug, so the
reference is compared against every identifier the dashboard answers to.
"""
value = lookup_result.permalink_value
key = lookup_result.permalink_key
if value is None or key is None:
return None
if not lookup_result.resolved_from_permalink:
# The identifier selected the dashboard, so the permalink only
# contributes state when it points at that same dashboard.
reference = value.get("dashboardId")
known_identifiers = {
str(candidate)
for candidate in (dashboard_id, dashboard_uuid, dashboard_slug)
if candidate is not None
}
if reference is None or str(reference) not in known_identifiers:
return None
raw_state = value.get("state")
state: dict[str, object] = dict(raw_state) if isinstance(raw_state, dict) else {}
if not user_can_view_data_model_metadata():
state = redact_filter_state_data_model_metadata(state)
return DashboardPermalinkState(
key=key,
state=sanitize_for_llm_context(
state,
field_path=("filter_state",),
excluded_field_names=frozenset(),
),
)
+55 -10
View File
@@ -252,7 +252,7 @@ DEFAULT_GET_DASHBOARD_INFO_COLUMNS: List[str] = [
class GetDashboardInfoRequest(MetadataCacheControl):
"""Request schema for get_dashboard_info with support for ID, UUID, or slug.
"""Request schema for dashboard identifiers and shared permalink URLs.
When permalink_key is provided, the tool will retrieve the dashboard's filter
state from the permalink, allowing you to see what filters the user has applied
@@ -263,21 +263,23 @@ class GetDashboardInfoRequest(MetadataCacheControl):
model_config = ConfigDict(populate_by_name=True)
identifier: Annotated[
int | str,
int | str | None,
Field(
description=(
"Dashboard identifier - can be numeric ID, UUID string, or slug"
"Dashboard ID, UUID, slug, bare permalink key, or a shared URL "
"containing /superset/dashboard/p/<key>/. Omit when "
"permalink_key is provided."
),
default=None,
validation_alias=AliasChoices("identifier", "id", "dashboard_id"),
),
]
permalink_key: str | None = Field(
default=None,
description=(
"Optional permalink key for retrieving dashboard filter state. When a "
"user applies filters in a dashboard, the state can be persisted in a "
"permalink. If provided, the tool returns the filter configuration "
"from that permalink."
"Key from a shared dashboard URL such as "
"'/superset/dashboard/p/<key>/'. Resolves the dashboard and returns "
"the shared active-tab and filter context; no identifier is required."
),
)
select_columns: Annotated[
@@ -305,16 +307,45 @@ class GetDashboardInfoRequest(MetadataCacheControl):
parsed = parse_json_or_list(value, "select_columns")
return parsed if parsed else list(DEFAULT_GET_DASHBOARD_INFO_COLUMNS)
@model_validator(mode="after")
def _require_identifier_or_permalink(self) -> "GetDashboardInfoRequest":
if self.identifier is None and self.permalink_key is None:
raise ValueError("Provide identifier or permalink_key")
return self
class GetDashboardLayoutRequest(BaseModel):
"""Request schema for get_dashboard_layout."""
"""Request a dashboard layout by its identifier or shared permalink.
Permalink requests resolve the dashboard while preserving shared active-tab
and filter state in the response.
"""
identifier: Annotated[
int | str,
int | str | None,
Field(
description="Dashboard identifier - can be numeric ID, UUID string, or slug"
default=None,
description=(
"Dashboard ID, UUID, slug, bare permalink key, or a shared URL "
"containing /superset/dashboard/p/<key>/. Omit when "
"permalink_key is provided."
),
),
]
permalink_key: str | None = Field(
default=None,
description=(
"Key from a shared dashboard URL such as "
"'/superset/dashboard/p/<key>/'. Resolves the dashboard and includes "
"the shared active-tab and filter context in the layout response."
),
)
@model_validator(mode="after")
def _require_identifier_or_permalink(self) -> "GetDashboardLayoutRequest":
if self.identifier is None and self.permalink_key is None:
raise ValueError("Provide identifier or permalink_key")
return self
class GetDashboardDatasetsRequest(BaseModel):
@@ -1567,6 +1598,20 @@ class DashboardLayout(BaseModel):
default=False,
description="False when position_json is missing or empty",
)
permalink_key: str | None = Field(
None, description="Resolved key when the input was a dashboard permalink"
)
filter_state: Dict[str, Any] | None = Field(
None,
description=(
"Shared dashboard state, including activeTabs, anchor, dataMask, "
"chartStates, and urlParams when present."
),
)
is_permalink_state: bool = Field(
False,
description="True when filter_state was resolved from a dashboard permalink",
)
def _parse_json_metadata(json_metadata_str: str | None) -> Dict[str, Any] | None:
@@ -27,82 +27,64 @@ from datetime import datetime, timezone
from typing import Any
from fastmcp import Context
from flask import g, has_request_context
from sqlalchemy.orm import subqueryload
from superset_core.mcp.decorators import tool, ToolAnnotations
from superset.dashboards.permalink.exceptions import DashboardPermalinkGetFailedError
from superset.dashboards.permalink.types import DashboardPermalinkValue
from superset.extensions import event_logger
from superset.mcp_service.auth import load_user_with_relationships
from superset.mcp_service.dashboard.permalink import (
DashboardLookupResult,
get_matching_dashboard_permalink_state,
lookup_dashboard_reference,
)
from superset.mcp_service.dashboard.schemas import (
dashboard_serializer,
DashboardError,
DashboardInfo,
DEFAULT_GET_DASHBOARD_INFO_COLUMNS,
GetDashboardInfoRequest,
redact_filter_state_data_model_metadata,
)
from superset.mcp_service.mcp_core import ModelGetInfoCore
from superset.mcp_service.privacy import user_can_view_data_model_metadata
from superset.mcp_service.utils import sanitize_for_llm_context
logger = logging.getLogger(__name__)
def _refresh_request_user_for_permalink_access() -> None:
"""Reload the request user before permalink access checks."""
if not has_request_context() or not getattr(g, "user", None):
return
current_user = g.user
if getattr(current_user, "is_anonymous", False):
return
username = getattr(current_user, "username", None)
email = getattr(current_user, "email", None)
if not username and not email:
return
refreshed_user = (
load_user_with_relationships(username=username)
if username
else load_user_with_relationships(email=email)
)
if refreshed_user is not None:
g.user = refreshed_user
def _apply_permalink_state(
result: DashboardInfo,
permalink_key: str,
permalink_state: dict[str, object],
) -> DashboardInfo:
"""Sanitize only the raw permalink fields added after serialization."""
"""Add sanitized permalink fields after dashboard serialization."""
payload = result.model_dump(mode="python")
payload["permalink_key"] = permalink_key
payload["filter_state"] = sanitize_for_llm_context(
permalink_state,
field_path=("filter_state",),
excluded_field_names=frozenset(),
)
payload["filter_state"] = permalink_state
payload["is_permalink_state"] = True
return DashboardInfo.model_validate(payload)
def _get_permalink_state(permalink_key: str) -> DashboardPermalinkValue | None:
"""Retrieve dashboard filter state from permalink.
Returns the permalink value containing dashboardId and state if found,
None otherwise.
"""
from superset.commands.dashboard.permalink.get import GetDashboardPermalinkCommand
try:
return GetDashboardPermalinkCommand(permalink_key).run()
except DashboardPermalinkGetFailedError as e:
logger.warning("Failed to retrieve permalink state: %s", e)
return None
def _lookup_dashboard(
tool: ModelGetInfoCore,
request: GetDashboardInfoRequest,
) -> tuple[
DashboardInfo | DashboardError,
DashboardLookupResult[DashboardInfo | DashboardError],
]:
"""Resolve an ordinary identifier or dashboard permalink, then run lookup."""
lookup_result = lookup_dashboard_reference(
identifier=request.identifier,
permalink_key=request.permalink_key,
lookup=tool.run_tool,
is_found=lambda result: isinstance(result, DashboardInfo),
)
result = lookup_result.result
if result is None:
# Only reachable when the dashboard had to come from a permalink, so the
# identifier's own "not found" error (when there is one) is preserved.
result = DashboardError.create(
"Dashboard permalink could not be resolved. It may be invalid or "
"expired; ask for a fresh shared dashboard link.",
"permalink_not_found",
)
return result, lookup_result
@tool(
@@ -118,7 +100,7 @@ async def get_dashboard_info(
request: GetDashboardInfoRequest, ctx: Context
) -> dict[str, Any] | DashboardError:
"""
Get dashboard metadata by ID, UUID, or slug.
Get dashboard metadata by ID, UUID, slug, or dashboard permalink.
Returns title, charts, and layout details.
@@ -130,9 +112,9 @@ async def get_dashboard_info(
with ``filters=[{"col": "dashboards", "opr": "eq", "value": <dashboard
id>}]`` and page through the results using ``page``/``page_size``.
When permalink_key is provided, also returns the filter state from that
permalink, allowing you to see what filters the user has applied to the
dashboard (not just the default filter state).
If the user gives you a shared URL containing ``/dashboard/p/<key>/``, pass
the URL or bare key as ``identifier`` (or use ``permalink_key`` alone). The
response includes the dashboard ID plus active tab and filter state.
Example usage:
```json
@@ -144,7 +126,6 @@ async def get_dashboard_info(
With permalink (filter state from URL):
```json
{
"identifier": 123,
"permalink_key": "abc123def456"
}
```
@@ -185,64 +166,44 @@ async def get_dashboard_info(
query_options=eager_options,
)
result = tool.run_tool(request.identifier)
result, lookup_result = _lookup_dashboard(tool, request)
permalink_key = lookup_result.permalink_key
permalink_value = lookup_result.permalink_value
if isinstance(result, DashboardInfo):
# If permalink_key is provided, retrieve filter state
if request.permalink_key:
if permalink_key:
await ctx.info(
"Retrieving filter state from permalink: permalink_key=%s"
% (request.permalink_key,)
% (permalink_key,)
)
_refresh_request_user_for_permalink_access()
permalink_value = _get_permalink_state(request.permalink_key)
if permalink_value:
# Verify the permalink belongs to the requested dashboard
# dashboardId in permalink is stored as str, result.id is int
permalink_dashboard_id = permalink_value.get("dashboardId")
try:
permalink_dashboard_id_int = (
int(permalink_dashboard_id)
if permalink_dashboard_id
else None
)
except (ValueError, TypeError):
permalink_dashboard_id_int = None
if (
permalink_dashboard_id_int is not None
and permalink_dashboard_id_int != result.id
):
permalink_state = get_matching_dashboard_permalink_state(
lookup_result,
result.id,
result.uuid,
result.slug,
)
if permalink_state is None:
await ctx.warning(
"permalink_key dashboardId (%s) does not match "
"requested dashboard id (%s); ignoring permalink "
"filter state." % (permalink_dashboard_id, result.id)
"permalink_key belongs to a different dashboard; "
"ignoring permalink filter state."
)
else:
# Extract the state from permalink value
# Handle None or non-dict state gracefully
raw_state = permalink_value.get("state")
permalink_state = (
dict(raw_state) if isinstance(raw_state, dict) else {}
)
if not user_can_view_data_model_metadata():
permalink_state = redact_filter_state_data_model_metadata(
permalink_state
)
result = _apply_permalink_state(
result,
request.permalink_key,
permalink_state,
permalink_state.key,
permalink_state.state,
)
await ctx.info(
"Filter state retrieved from permalink: "
"has_dataMask=%s, has_chartStates=%s, has_activeTabs=%s"
% (
"dataMask" in permalink_state,
"chartStates" in permalink_state,
"activeTabs" in permalink_state,
"dataMask" in permalink_state.state,
"chartStates" in permalink_state.state,
"activeTabs" in permalink_state.state,
)
)
else:
@@ -266,7 +227,7 @@ async def get_dashboard_info(
# override select_columns, ensure filter_state is present so the
# caller gets the data they came for.
effective_select_columns = list(request.select_columns)
if request.permalink_key and effective_select_columns == list(
if result.is_permalink_state and effective_select_columns == list(
DEFAULT_GET_DASHBOARD_INFO_COLUMNS
):
effective_select_columns.append("filter_state")
@@ -31,6 +31,10 @@ from fastmcp import Context
from superset_core.mcp.decorators import tool, ToolAnnotations
from superset.extensions import event_logger
from superset.mcp_service.dashboard.permalink import (
get_matching_dashboard_permalink_state,
lookup_dashboard_reference,
)
from superset.mcp_service.dashboard.schemas import (
dashboard_layout_serializer,
DashboardError,
@@ -55,7 +59,7 @@ async def get_dashboard_layout(
request: GetDashboardLayoutRequest, ctx: Context
) -> DashboardLayout | DashboardError:
"""
Get parsed dashboard layout by ID, UUID, or slug.
Get parsed dashboard layout by ID, UUID, slug, or dashboard permalink.
Returns the tabs and chart positions extracted from the dashboard's
position_json. get_dashboard_info omits position_json to keep responses
@@ -63,6 +67,10 @@ async def get_dashboard_layout(
explain which charts live under which tab, or to locate a chart by
its parent tab).
If the user gives you a shared URL containing ``/dashboard/p/<key>/``, pass
the URL or bare key as ``identifier`` (or use ``permalink_key`` alone). The
response identifies the active tab and includes the shared filter state.
Example usage:
```json
{
@@ -86,9 +94,40 @@ async def get_dashboard_layout(
supports_slug=True,
logger=logger,
)
result = core.run_tool(request.identifier)
lookup_result = lookup_dashboard_reference(
identifier=request.identifier,
permalink_key=request.permalink_key,
lookup=core.run_tool,
is_found=lambda value: isinstance(value, DashboardLayout),
)
result = lookup_result.result
if result is None:
# Only reachable when the dashboard had to come from a permalink,
# so an identifier's own "not found" error is preserved below.
return DashboardError.create(
"Dashboard permalink could not be resolved. It may be invalid "
"or expired; ask for a fresh shared dashboard link.",
"permalink_not_found",
)
if isinstance(result, DashboardLayout):
if lookup_result.permalink_value:
permalink_state = get_matching_dashboard_permalink_state(
lookup_result, result.id, result.uuid
)
if permalink_state:
payload = result.model_dump(mode="python")
payload.update(
permalink_key=permalink_state.key,
filter_state=permalink_state.state,
is_permalink_state=True,
)
result = DashboardLayout.model_validate(payload)
else:
await ctx.warning(
"permalink_key belongs to a different dashboard; ignoring "
"its active-tab and filter state."
)
await ctx.info(
"Dashboard layout retrieved: id=%s, tab_count=%s, chart_count=%s, "
"has_layout=%s"
@@ -38,6 +38,7 @@ from superset.mcp_service.dashboard.schemas import (
DuplicateDashboardResponse,
GenerateDashboardRequest,
GetDashboardInfoRequest,
GetDashboardLayoutRequest,
ListDashboardsRequest,
serialize_chart_summary,
serialize_dashboard_object,
@@ -960,6 +961,29 @@ class TestRequestSchemaAliasChoices:
)
assert req.select_columns == ["id", "dashboard_title"]
def test_get_dashboard_info_requires_reference(self) -> None:
with pytest.raises(ValidationError, match="identifier or permalink_key"):
GetDashboardInfoRequest.model_validate({})
@pytest.mark.parametrize(
"payload",
[
{"identifier": 42},
{"permalink_key": "shared-key"},
{"identifier": 42, "permalink_key": "shared-key"},
],
)
def test_get_dashboard_layout_accepts_reference(
self, payload: dict[str, Any]
) -> None:
request = GetDashboardLayoutRequest.model_validate(payload)
assert request.identifier == payload.get("identifier")
assert request.permalink_key == payload.get("permalink_key")
def test_get_dashboard_layout_requires_reference(self) -> None:
with pytest.raises(ValidationError, match="identifier or permalink_key"):
GetDashboardLayoutRequest.model_validate({})
def test_list_dashboards_select_columns_columns_alias(self) -> None:
req = ListDashboardsRequest.model_validate(
{"columns": ["id", "dashboard_title"]}
@@ -0,0 +1,169 @@
# 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.
"""Tests for MCP dashboard permalink helpers."""
from unittest.mock import patch
import pytest
from flask import g
from superset.commands.dashboard.exceptions import DashboardAccessDeniedError
from superset.mcp_service.dashboard.permalink import (
DashboardLookupResult,
extract_dashboard_permalink_key,
get_dashboard_permalink,
get_matching_dashboard_permalink_state,
refresh_request_user_for_permalink_access,
)
@pytest.mark.parametrize(
("value", "expected"),
[
("bare-key", "bare-key"),
("/superset/dashboard/p/shared-key/", "shared-key"),
(
"https://example.test/prefix/dashboard/p/shared-key/?foo=bar#tab",
"shared-key",
),
(
"https://example.test/dashboard/not-p/shared-key/",
"https://example.test/dashboard/not-p/shared-key/",
),
],
)
def test_extract_dashboard_permalink_key(value: str, expected: str) -> None:
assert extract_dashboard_permalink_key(value) == expected
@patch(
"superset.commands.dashboard.permalink.get.GetDashboardPermalinkCommand.run",
side_effect=DashboardAccessDeniedError(),
)
@patch(
"superset.mcp_service.dashboard.permalink.refresh_request_user_for_permalink_access"
)
def test_get_dashboard_permalink_hides_access_denial(mock_refresh, mock_run) -> None:
assert get_dashboard_permalink("inaccessible-key") is None
mock_refresh.assert_called_once_with()
mock_run.assert_called_once_with()
@pytest.mark.parametrize(
("username", "email", "expected_kwargs"),
[
("admin", None, {"username": "admin"}),
(None, "admin@example.com", {"email": "admin@example.com"}),
],
)
def test_refresh_request_user_for_permalink_access(
app, username: str | None, email: str | None, expected_kwargs: dict[str, str]
) -> None:
current_user = type(
"CurrentUser",
(),
{"username": username, "email": email, "is_anonymous": False},
)()
refreshed_user = object()
with (
patch(
"superset.mcp_service.dashboard.permalink.load_user_with_relationships",
return_value=refreshed_user,
) as mock_load,
app.test_request_context("/mcp"),
):
g.user = current_user
refresh_request_user_for_permalink_access()
assert g.user is refreshed_user
mock_load.assert_called_once_with(**expected_kwargs)
@pytest.mark.parametrize(
("username", "email", "is_anonymous"),
[("anonymous", "anonymous@example.com", True), (None, None, False)],
)
def test_refresh_request_user_for_permalink_access_skips_unresolvable_user(
app, username: str | None, email: str | None, is_anonymous: bool
) -> None:
current_user = type(
"CurrentUser",
(),
{"username": username, "email": email, "is_anonymous": is_anonymous},
)()
with (
patch(
"superset.mcp_service.dashboard.permalink.load_user_with_relationships"
) as mock_load,
app.test_request_context("/mcp"),
):
g.user = current_user
refresh_request_user_for_permalink_access()
assert g.user is current_user
mock_load.assert_not_called()
@pytest.mark.parametrize(
("reference", "expected_match"),
[
# CreateDashboardPermalinkCommand stores str(dashboard.uuid).
("3f1a2b6c-9d4e-4f80-9c2a-7b1d5e6f8a90", True),
# Legacy permalinks may hold the numeric id or the slug.
("42", True),
("sales-dashboard", True),
("99", False),
("00000000-0000-0000-0000-000000000000", False),
],
)
def test_get_matching_dashboard_permalink_state_accepts_every_identifier(
app, reference: str, expected_match: bool
) -> None:
lookup_result = DashboardLookupResult(
result=object(),
permalink_key="key-1",
permalink_value={"dashboardId": reference, "state": {"activeTabs": ["TAB-A"]}},
)
with app.test_request_context("/mcp"):
state = get_matching_dashboard_permalink_state(
lookup_result,
42,
"3f1a2b6c-9d4e-4f80-9c2a-7b1d5e6f8a90",
"sales-dashboard",
)
assert (state is not None) is expected_match
def test_get_matching_dashboard_permalink_state_skips_check_when_permalink_resolved(
app,
) -> None:
"""The permalink-only path already selected the dashboard from the permalink,
so its state is never re-verified against the resolved identifiers.
"""
lookup_result = DashboardLookupResult(
result=object(),
permalink_key="key-1",
permalink_value={"dashboardId": "whatever", "state": {"activeTabs": ["TAB-A"]}},
resolved_from_permalink=True,
)
with app.test_request_context("/mcp"):
state = get_matching_dashboard_permalink_state(lookup_result, 42)
assert state is not None
assert state.key == "key-1"
@@ -26,15 +26,13 @@ from unittest.mock import Mock, patch
import pytest
from fastmcp import Client
from fastmcp.exceptions import ToolError
from flask import g
from superset.mcp_service.app import mcp
from superset.mcp_service.dashboard.schemas import (
DashboardError,
DashboardInfo,
ListDashboardsRequest,
)
from superset.mcp_service.dashboard.tool.get_dashboard_info import (
_refresh_request_user_for_permalink_access,
)
from superset.mcp_service.utils.sanitization import (
LLM_CONTEXT_CLOSE_DELIMITER,
LLM_CONTEXT_OPEN_DELIMITER,
@@ -471,15 +469,14 @@ async def test_get_dashboard_info_permalink_does_not_double_sanitize(
"superset.mcp_service.dashboard.schemas.user_can_view_data_model_metadata",
return_value=True,
),
patch.object(
get_dashboard_info_module,
patch(
"superset.mcp_service.dashboard.permalink."
"user_can_view_data_model_metadata",
return_value=True,
),
patch.object(
get_dashboard_info_module,
"_get_permalink_state",
return_value=permalink_value,
patch(
"superset.mcp_service.dashboard.permalink.get_dashboard_permalink",
return_value=("permalink-1", permalink_value),
),
):
async with Client(mcp_server) as client:
@@ -555,15 +552,14 @@ async def test_get_dashboard_info_permalink_key_includes_filter_state(
"superset.mcp_service.dashboard.schemas.user_can_view_data_model_metadata",
return_value=True,
),
patch.object(
get_dashboard_info_module,
patch(
"superset.mcp_service.dashboard.permalink."
"user_can_view_data_model_metadata",
return_value=True,
),
patch.object(
get_dashboard_info_module,
"_get_permalink_state",
return_value=permalink_value,
patch(
"superset.mcp_service.dashboard.permalink.get_dashboard_permalink",
return_value=("some-key", permalink_value),
),
):
async with Client(mcp_server) as client:
@@ -579,121 +575,151 @@ async def test_get_dashboard_info_permalink_key_includes_filter_state(
assert result.data["permalink_key"] == "some-key"
def test_refresh_request_user_for_permalink_access(
app,
@patch("superset.mcp_service.mcp_core.ModelGetInfoCore.run_tool")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_info_resolves_permalink_without_identifier(
mock_permalink, mock_run_tool, mcp_server
):
refreshed_user = Mock()
refreshed_user.username = "admin"
refreshed_user.roles = []
refreshed_user.groups = []
mock_permalink.return_value = (
"shared-key",
{"dashboardId": "42", "state": {"activeTabs": ["TAB-A"], "dataMask": {}}},
)
mock_run_tool.return_value = DashboardInfo(id=42, dashboard_title="Sales Dashboard")
current_user = Mock()
current_user.username = "admin"
current_user.email = None
current_user.is_anonymous = False
with (
patch.object(
get_dashboard_info_module,
"load_user_with_relationships",
return_value=refreshed_user,
) as mock_load_user_with_relationships,
app.test_request_context("/mcp"),
):
g.user = current_user
_refresh_request_user_for_permalink_access()
mock_load_user_with_relationships.assert_called_once_with(username="admin")
assert g.user is refreshed_user
def test_refresh_request_user_for_permalink_access_uses_email_when_username_missing(
app,
):
refreshed_user = Mock()
refreshed_user.email = "admin@example.com"
current_user = Mock()
current_user.username = None
current_user.email = "admin@example.com"
current_user.is_anonymous = False
with (
patch.object(
get_dashboard_info_module,
"load_user_with_relationships",
return_value=refreshed_user,
) as mock_load_user_with_relationships,
app.test_request_context("/mcp"),
):
g.user = current_user
_refresh_request_user_for_permalink_access()
mock_load_user_with_relationships.assert_called_once_with(
email="admin@example.com"
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_info", {"request": {"permalink_key": "shared-key"}}
)
assert g.user is refreshed_user
assert result.data["id"] == 42
assert result.data["permalink_key"] == "shared-key"
assert result.data["filter_state"]["activeTabs"] == [_wrapped("TAB-A")]
mock_run_tool.assert_called_once_with("42")
def test_refresh_request_user_for_permalink_access_skips_anonymous_user(app):
current_user = Mock()
current_user.username = "anonymous"
current_user.email = "anonymous@example.com"
current_user.is_anonymous = True
@patch(
"superset.mcp_service.dashboard.permalink.get_dashboard_permalink",
return_value=None,
)
@pytest.mark.asyncio
async def test_get_dashboard_info_invalid_permalink_is_actionable(
mock_permalink, mcp_server
):
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_info", {"request": {"permalink_key": "expired-key"}}
)
with (
patch.object(
get_dashboard_info_module,
"load_user_with_relationships",
) as mock_load_user_with_relationships,
app.test_request_context("/mcp"),
):
g.user = current_user
_refresh_request_user_for_permalink_access()
mock_load_user_with_relationships.assert_not_called()
assert g.user is current_user
assert result.data["error_type"] == "permalink_not_found"
assert "fresh shared dashboard link" in result.data["error"]
def test_refresh_request_user_for_permalink_access_skips_missing_identifier(app):
current_user = Mock()
current_user.username = None
current_user.email = None
current_user.is_anonymous = False
@patch("superset.mcp_service.mcp_core.ModelGetInfoCore.run_tool")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_info_identifier_takes_precedence_over_permalink(
mock_permalink, mock_run_tool, mcp_server
):
mock_permalink.return_value = (
"dashboard-20-key",
{"dashboardId": "20", "state": {"activeTabs": ["TAB-20"]}},
)
mock_run_tool.return_value = DashboardInfo(
id=10, dashboard_title="Requested Dashboard"
)
with (
patch.object(
get_dashboard_info_module,
"load_user_with_relationships",
) as mock_load_user_with_relationships,
app.test_request_context("/mcp"),
):
g.user = current_user
_refresh_request_user_for_permalink_access()
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_info",
{"request": {"identifier": 10, "permalink_key": "dashboard-20-key"}},
)
mock_load_user_with_relationships.assert_not_called()
assert g.user is current_user
assert result.data["id"] == 10
assert result.data["is_permalink_state"] is False
assert "filter_state" not in result.data
mock_run_tool.assert_called_once_with(10)
def test_refresh_request_user_for_permalink_access_keeps_user_when_reload_fails(app):
current_user = Mock()
current_user.username = "admin"
current_user.email = None
current_user.is_anonymous = False
@patch("superset.mcp_service.mcp_core.ModelGetInfoCore.run_tool")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_info_permalink_with_uuid_dashboard_id(
mock_permalink, mock_run_tool, mcp_server
):
"""CreateDashboardPermalinkCommand stores dashboardId as the dashboard UUID,
so an explicit identifier plus that permalink must still yield filter state.
"""
dashboard_uuid = "3f1a2b6c-9d4e-4f80-9c2a-7b1d5e6f8a90"
mock_permalink.return_value = (
"uuid-key",
{"dashboardId": dashboard_uuid, "state": {"activeTabs": ["TAB-A"]}},
)
mock_run_tool.return_value = DashboardInfo(
id=42, dashboard_title="Sales Dashboard", uuid=dashboard_uuid
)
with (
patch.object(
get_dashboard_info_module,
"load_user_with_relationships",
return_value=None,
) as mock_load_user_with_relationships,
app.test_request_context("/mcp"),
):
g.user = current_user
_refresh_request_user_for_permalink_access()
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_info",
{"request": {"identifier": 42, "permalink_key": "uuid-key"}},
)
mock_load_user_with_relationships.assert_called_once_with(username="admin")
assert g.user is current_user
assert result.data["id"] == 42
assert result.data["is_permalink_state"] is True
assert result.data["permalink_key"] == "uuid-key"
assert result.data["filter_state"]["activeTabs"] == [_wrapped("TAB-A")]
@patch("superset.mcp_service.mcp_core.ModelGetInfoCore.run_tool")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_info_permalink_with_slug_dashboard_id(
mock_permalink, mock_run_tool, mcp_server
):
"""Pre-3.1 permalinks can carry a slug in dashboardId."""
mock_permalink.return_value = (
"slug-key",
{"dashboardId": "sales-dashboard", "state": {"activeTabs": ["TAB-A"]}},
)
mock_run_tool.return_value = DashboardInfo(
id=42, dashboard_title="Sales Dashboard", slug="sales-dashboard"
)
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_info",
{"request": {"identifier": 42, "permalink_key": "slug-key"}},
)
assert result.data["is_permalink_state"] is True
assert result.data["filter_state"]["activeTabs"] == [_wrapped("TAB-A")]
@patch("superset.mcp_service.mcp_core.ModelGetInfoCore.run_tool")
@patch(
"superset.mcp_service.dashboard.permalink.get_dashboard_permalink",
return_value=None,
)
@pytest.mark.asyncio
async def test_get_dashboard_info_unknown_slug_keeps_not_found_error(
mock_permalink, mock_run_tool, mcp_server
):
"""A plain slug typo keeps its own not-found error instead of asking the
user for a shared link they never mentioned.
"""
mock_run_tool.return_value = DashboardError.create(
"DashboardInfo with identifier 'sales-dashbord' not found", "not_found"
)
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_info", {"request": {"identifier": "sales-dashbord"}}
)
assert result.data["error_type"] == "not_found"
assert "sales-dashbord" in result.data["error"]
assert "fresh shared dashboard link" not in result.data["error"]
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@@ -930,15 +956,14 @@ async def test_get_dashboard_info_restricted_user_redacts_permalink_filter_state
"superset.mcp_service.dashboard.schemas.user_can_view_data_model_metadata",
return_value=False,
),
patch.object(
get_dashboard_info_module,
patch(
"superset.mcp_service.dashboard.permalink."
"user_can_view_data_model_metadata",
return_value=False,
),
patch.object(
get_dashboard_info_module,
"_get_permalink_state",
return_value=permalink_value,
patch(
"superset.mcp_service.dashboard.permalink.get_dashboard_permalink",
return_value=("abc123", permalink_value),
),
):
async with Client(mcp_server) as client:
@@ -263,6 +263,143 @@ async def test_get_dashboard_layout_not_found(mock_find, mcp_server):
assert data["error_type"] == "not_found"
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_layout_resolves_shared_permalink(
mock_permalink, mock_find, mcp_server
):
mock_permalink.return_value = (
"shared-key",
{
"dashboardId": "42",
"state": {
"activeTabs": ["TAB-2"],
"dataMask": {"FILTER-1": {"filterState": {"value": "EMEA"}}},
},
},
)
mock_find.return_value = _build_dashboard_mock(
dashboard_id=42, position_json=_tabbed_layout()
)
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_layout",
{
"request": {
"identifier": "https://example.test/superset/dashboard/p/shared-key/"
}
},
)
data = json.loads(result.content[0].text)
assert data["id"] == 42
assert data["permalink_key"] == "shared-key"
assert data["is_permalink_state"] is True
assert data["filter_state"]["activeTabs"] == [_wrapped("TAB-2")]
assert mock_find.call_args_list[-1].args == (42,)
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_layout_invalid_permalink_is_actionable(
mock_permalink, mcp_server
):
mock_permalink.return_value = None
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_layout", {"request": {"permalink_key": "expired-key"}}
)
data = json.loads(result.content[0].text)
assert data["error_type"] == "permalink_not_found"
assert "fresh shared dashboard link" in data["error"]
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_layout_identifier_takes_precedence_over_permalink(
mock_permalink, mock_find, mcp_server
):
mock_permalink.return_value = (
"dashboard-20-key",
{"dashboardId": "20", "state": {"activeTabs": ["TAB-20"]}},
)
mock_find.return_value = _build_dashboard_mock(
dashboard_id=10, position_json=_tabbed_layout()
)
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_layout",
{"request": {"identifier": 10, "permalink_key": "dashboard-20-key"}},
)
data = json.loads(result.content[0].text)
assert data["id"] == 10
assert data["is_permalink_state"] is False
mock_find.assert_called_once_with(10, query_options=None)
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@pytest.mark.asyncio
async def test_get_dashboard_layout_permalink_with_uuid_dashboard_id(
mock_permalink, mock_find, mcp_server
):
"""CreateDashboardPermalinkCommand stores dashboardId as the dashboard UUID,
so an explicit identifier plus that permalink must still yield filter state.
"""
dashboard_uuid = "3f1a2b6c-9d4e-4f80-9c2a-7b1d5e6f8a90"
mock_permalink.return_value = (
"uuid-key",
{"dashboardId": dashboard_uuid, "state": {"activeTabs": ["TAB-2"]}},
)
mock_find.return_value = _build_dashboard_mock(
dashboard_id=42, uuid=dashboard_uuid, position_json=_tabbed_layout()
)
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_layout",
{"request": {"identifier": 42, "permalink_key": "uuid-key"}},
)
data = json.loads(result.content[0].text)
assert data["id"] == 42
assert data["is_permalink_state"] is True
assert data["permalink_key"] == "uuid-key"
assert data["filter_state"]["activeTabs"] == [_wrapped("TAB-2")]
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@patch(
"superset.mcp_service.dashboard.permalink.get_dashboard_permalink",
return_value=None,
)
@pytest.mark.asyncio
async def test_get_dashboard_layout_unknown_slug_keeps_not_found_error(
mock_permalink, mock_find, mcp_server
):
"""A plain slug typo keeps its own not-found error instead of asking the
user for a shared link they never mentioned.
"""
mock_find.return_value = None
async with Client(mcp_server) as client:
result = await client.call_tool(
"get_dashboard_layout", {"request": {"identifier": "sales-dashbord"}}
)
data = json.loads(result.content[0].text)
assert data["error_type"] == "not_found"
assert "sales-dashbord" in data["error"]
assert "fresh shared dashboard link" not in data["error"]
def test_extract_layout_handles_invalid_json():
tabs, charts = _extract_layout_from_position("{ not json")
assert tabs == []