Compare commits

...

1 Commits

Author SHA1 Message Date
Amin Ghadersohi
9b6ee2522c feat(mcp): add create_css_template and update_css_template tools
- Add CreateCssTemplateCommand, UpdateCssTemplateCommand with exceptions
- Add create_css_template and update_css_template MCP tools with Pydantic
  schemas, event logging, and structured error responses
- Add sanitize_error_for_llm_context validator on error fields (CWE-79)
- Add unit tests for both tools including schema validation and exception
  re-raise coverage
2026-05-30 04:40:47 +00:00
10 changed files with 847 additions and 1 deletions

View File

@@ -0,0 +1,46 @@
# 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.
import logging
from functools import partial
from typing import Any
from superset.commands.base import BaseCommand
from superset.commands.css.exceptions import (
CssTemplateCreateFailedError,
CssTemplateInvalidError,
)
from superset.daos.css import CssTemplateDAO
from superset.models.core import CssTemplate
from superset.utils.decorators import on_error, transaction
logger = logging.getLogger(__name__)
class CreateCssTemplateCommand(BaseCommand):
def __init__(self, properties: dict[str, Any]):
self._properties = properties
@transaction(on_error=partial(on_error, reraise=CssTemplateCreateFailedError))
def run(self) -> CssTemplate:
self.validate()
return CssTemplateDAO.create(attributes=self._properties)
def validate(self) -> None:
if not self._properties.get("template_name", "").strip():
raise CssTemplateInvalidError()
if "css" not in self._properties:
raise CssTemplateInvalidError()

View File

@@ -16,7 +16,13 @@
# under the License.
from flask_babel import lazy_gettext as _
from superset.commands.exceptions import CommandException, DeleteFailedError
from superset.commands.exceptions import (
CommandException,
CommandInvalidError,
CreateFailedError,
DeleteFailedError,
UpdateFailedError,
)
class CssTemplateDeleteFailedError(DeleteFailedError):
@@ -25,3 +31,15 @@ class CssTemplateDeleteFailedError(DeleteFailedError):
class CssTemplateNotFoundError(CommandException):
message = _("CSS template not found.")
class CssTemplateCreateFailedError(CreateFailedError):
message = _("CSS template could not be created.")
class CssTemplateInvalidError(CommandInvalidError):
message = _("CSS template parameters are invalid.")
class CssTemplateUpdateFailedError(UpdateFailedError):
message = _("CSS template could not be updated.")

View File

@@ -0,0 +1,53 @@
# 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.
import logging
from functools import partial
from typing import Any
from superset.commands.base import BaseCommand
from superset.commands.css.exceptions import (
CssTemplateInvalidError,
CssTemplateNotFoundError,
CssTemplateUpdateFailedError,
)
from superset.daos.css import CssTemplateDAO
from superset.models.core import CssTemplate
from superset.utils.decorators import on_error, transaction
logger = logging.getLogger(__name__)
class UpdateCssTemplateCommand(BaseCommand):
def __init__(self, model_id: int, properties: dict[str, Any]):
self._model_id = model_id
self._properties = properties
self._model: CssTemplate | None = None
@transaction(on_error=partial(on_error, reraise=CssTemplateUpdateFailedError))
def run(self) -> CssTemplate:
self.validate()
assert self._model
return CssTemplateDAO.update(self._model, attributes=self._properties)
def validate(self) -> None:
self._model = CssTemplateDAO.find_by_id(self._model_id)
if not self._model:
raise CssTemplateNotFoundError()
template_name = self._properties.get("template_name")
if template_name is not None and not template_name.strip():
raise CssTemplateInvalidError()

View File

@@ -148,6 +148,8 @@ Database Connections:
CSS Templates:
- list_css_templates: List CSS templates with advanced filters (1-based pagination)
- get_css_template_info: Get CSS template details by ID (includes full css content)
- create_css_template: Create a new named CSS template for dashboard styling
- update_css_template: Update an existing CSS template's name or CSS content
Themes:
- list_themes: List themes with advanced filters (1-based pagination)
@@ -685,8 +687,10 @@ from superset.mcp_service.chart.tool import ( # noqa: F401, E402
update_chart_preview,
)
from superset.mcp_service.css_template.tool import ( # noqa: F401, E402
create_css_template,
get_css_template_info,
list_css_templates,
update_css_template,
)
from superset.mcp_service.dashboard.tool import ( # noqa: F401, E402
add_chart_to_existing_dashboard,

View File

@@ -278,3 +278,107 @@ def serialize_css_template_object(obj: Any) -> CssTemplateInfo | None:
changed_by_name=getattr(obj, "changed_by_name", None) or None,
)
)
class CreateCssTemplateRequest(BaseModel):
model_config = ConfigDict(populate_by_name=True)
template_name: str = Field(
...,
min_length=1,
max_length=250,
description="Name for the CSS template.",
)
css: str = Field(
...,
description="CSS content for the template.",
)
@field_validator("template_name")
@classmethod
def template_name_must_not_be_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("template_name must not be empty")
return v.strip()
class CreateCssTemplateResponse(BaseModel):
"""Response schema for create_css_template."""
id: int | None = Field(
None,
description="ID of the created CSS template. None if creation failed.",
)
template_name: str | None = Field(
None,
description="Name of the created CSS template.",
)
css: str | None = Field(
None,
description="CSS content of the created template.",
)
error: str | None = Field(
None,
description="Error message if creation failed, otherwise null.",
)
@field_validator("error")
@classmethod
def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
"""Sanitize error text before it is exposed to LLM context."""
if value is None:
return value
return sanitize_for_llm_context(value, field_path=("error",))
class UpdateCssTemplateRequest(BaseModel):
model_config = ConfigDict(populate_by_name=True)
id: int = Field(..., description="ID of the CSS template to update.")
template_name: str | None = Field(
None,
max_length=250,
description="New name for the CSS template.",
)
css: str | None = Field(
None,
description="New CSS content for the template.",
)
@field_validator("template_name")
@classmethod
def template_name_must_not_be_empty(cls, v: str | None) -> str | None:
if v is not None:
if not v.strip():
raise ValueError("template_name must not be empty")
return v.strip()
return v
class UpdateCssTemplateResponse(BaseModel):
"""Response schema for update_css_template."""
id: int | None = Field(
None,
description="ID of the updated CSS template. None if update failed.",
)
template_name: str | None = Field(
None,
description="Name of the updated CSS template.",
)
css: str | None = Field(
None,
description="CSS content of the updated template.",
)
error: str | None = Field(
None,
description="Error message if update failed, otherwise null.",
)
@field_validator("error")
@classmethod
def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
"""Sanitize error text before it is exposed to LLM context."""
if value is None:
return value
return sanitize_for_llm_context(value, field_path=("error",))

View File

@@ -15,10 +15,14 @@
# specific language governing permissions and limitations
# under the License.
from .create_css_template import create_css_template
from .get_css_template_info import get_css_template_info
from .list_css_templates import list_css_templates
from .update_css_template import update_css_template
__all__ = [
"list_css_templates",
"get_css_template_info",
"create_css_template",
"update_css_template",
]

View File

@@ -0,0 +1,95 @@
# 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.
from fastmcp import Context
from superset_core.mcp.decorators import tool, ToolAnnotations
from superset.extensions import event_logger
from superset.mcp_service.css_template.schemas import (
CreateCssTemplateRequest,
CreateCssTemplateResponse,
)
@tool(
tags=["mutate"],
class_permission_name="CssTemplate",
method_permission_name="write",
annotations=ToolAnnotations(
title="Create CSS template",
readOnlyHint=False,
destructiveHint=False,
),
)
async def create_css_template(
request: CreateCssTemplateRequest, ctx: Context
) -> CreateCssTemplateResponse:
"""Create a new CSS template that can be applied to dashboards.
Use this tool when a user wants to save a CSS stylesheet as a named
template for reuse across multiple dashboards.
The returned ``id`` can be used when configuring dashboard appearance.
"""
await ctx.info("Creating CSS template: template_name=%r" % (request.template_name,))
try:
from superset.commands.css.create import CreateCssTemplateCommand
from superset.commands.css.exceptions import (
CssTemplateCreateFailedError,
CssTemplateInvalidError,
)
with event_logger.log_context(action="mcp.create_css_template.create"):
template = CreateCssTemplateCommand(
{
"template_name": request.template_name,
"css": request.css,
}
).run()
await ctx.info(
"CSS template created: id=%s, template_name=%r"
% (template.id, template.template_name)
)
return CreateCssTemplateResponse(
id=template.id,
template_name=template.template_name,
css=template.css,
)
except CssTemplateInvalidError as exc:
await ctx.warning("CSS template validation failed: %s" % (str(exc),))
return CreateCssTemplateResponse(
template_name=request.template_name,
css=request.css,
error=str(exc),
)
except CssTemplateCreateFailedError as exc:
await ctx.error("CSS template creation failed: %s" % (str(exc),))
return CreateCssTemplateResponse(
template_name=request.template_name,
css=request.css,
error=f"Failed to create CSS template: {exc}",
)
except Exception as exc:
await ctx.error(
"Unexpected error creating CSS template: %s: %s"
% (type(exc).__name__, str(exc))
)
raise

View File

@@ -0,0 +1,111 @@
# 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.
from typing import Any
from fastmcp import Context
from superset_core.mcp.decorators import tool, ToolAnnotations
from superset.extensions import event_logger
from superset.mcp_service.css_template.schemas import (
UpdateCssTemplateRequest,
UpdateCssTemplateResponse,
)
@tool(
tags=["mutate"],
class_permission_name="CssTemplate",
method_permission_name="write",
annotations=ToolAnnotations(
title="Update CSS template",
readOnlyHint=False,
destructiveHint=False,
),
)
async def update_css_template(
request: UpdateCssTemplateRequest, ctx: Context
) -> UpdateCssTemplateResponse:
"""Update an existing CSS template's name or CSS content.
Use this tool when a user wants to rename a CSS template or replace its
CSS content. At least one of ``template_name`` or ``css`` must be provided.
The template is identified by its ``id``.
"""
await ctx.info(
"Updating CSS template: id=%s, fields=%r"
% (
request.id,
[f for f in ("template_name", "css") if getattr(request, f) is not None],
)
)
try:
from superset.commands.css.exceptions import (
CssTemplateInvalidError,
CssTemplateNotFoundError,
CssTemplateUpdateFailedError,
)
from superset.commands.css.update import UpdateCssTemplateCommand
properties: dict[str, Any] = {}
if request.template_name is not None:
properties["template_name"] = request.template_name
if request.css is not None:
properties["css"] = request.css
if not properties:
return UpdateCssTemplateResponse(
error="At least one of template_name or css must be provided.",
)
with event_logger.log_context(action="mcp.update_css_template.update"):
template = UpdateCssTemplateCommand(request.id, properties).run()
await ctx.info(
"CSS template updated: id=%s, template_name=%r"
% (template.id, template.template_name)
)
return UpdateCssTemplateResponse(
id=template.id,
template_name=template.template_name,
css=template.css,
)
except CssTemplateNotFoundError:
await ctx.warning("CSS template not found: id=%s" % (request.id,))
return UpdateCssTemplateResponse(
error="CSS template not found: %s" % (request.id,),
)
except CssTemplateInvalidError as exc:
await ctx.warning("CSS template validation failed: %s" % (str(exc),))
return UpdateCssTemplateResponse(
error=str(exc),
)
except CssTemplateUpdateFailedError as exc:
await ctx.error("CSS template update failed: %s" % (str(exc),))
return UpdateCssTemplateResponse(
error="Failed to update CSS template: %s" % (exc,),
)
except Exception as exc:
await ctx.error(
"Unexpected error updating CSS template: %s: %s"
% (type(exc).__name__, str(exc))
)
raise

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.
from unittest.mock import MagicMock, Mock, patch
import pytest
from fastmcp import Client
from superset.mcp_service.app import mcp
from superset.mcp_service.css_template.schemas import CreateCssTemplateRequest
from superset.utils import json
@pytest.fixture
def mcp_server():
return mcp
@pytest.fixture(autouse=True)
def mock_auth():
"""Mock authentication for all tests."""
with patch(
"superset.mcp_service.auth.get_user_from_request",
return_value=Mock(is_authenticated=True),
):
yield
# ---------------------------------------------------------------------------
# Schema tests
# ---------------------------------------------------------------------------
def test_create_css_template_request_valid() -> None:
req = CreateCssTemplateRequest(
template_name="My Theme",
css=".header { color: red; }",
)
assert req.template_name == "My Theme"
assert req.css == ".header { color: red; }"
def test_create_css_template_request_strips_name_whitespace() -> None:
req = CreateCssTemplateRequest(
template_name=" My Theme ",
css=".header { color: red; }",
)
assert req.template_name == "My Theme"
def test_create_css_template_request_empty_name_fails() -> None:
from pydantic import ValidationError
with pytest.raises(ValidationError, match="template_name must not be empty"):
CreateCssTemplateRequest(template_name=" ", css=".header { color: red; }")
def test_create_css_template_request_name_too_long() -> None:
from pydantic import ValidationError
with pytest.raises(ValidationError):
CreateCssTemplateRequest(template_name="a" * 251, css="")
# ---------------------------------------------------------------------------
# Tool logic tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_css_template_success(mcp_server: object) -> None:
"""Happy path: template created, id and fields returned."""
mock_template = MagicMock()
mock_template.id = 7
mock_template.template_name = "Dark Theme"
mock_template.css = "body { background: #000; }"
mock_command = MagicMock()
mock_command.run.return_value = mock_template
with patch(
"superset.commands.css.create.CreateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = CreateCssTemplateRequest(
template_name="Dark Theme",
css="body { background: #000; }",
)
result = await client.call_tool(
"create_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] == 7
assert data["template_name"] == "Dark Theme"
assert data["css"] == "body { background: #000; }"
assert data["error"] is None
@pytest.mark.asyncio
async def test_create_css_template_invalid_error(mcp_server: object) -> None:
"""CssTemplateInvalidError is caught and returned as an error response."""
from superset.commands.css.exceptions import CssTemplateInvalidError
mock_command = MagicMock()
mock_command.run.side_effect = CssTemplateInvalidError()
with patch(
"superset.commands.css.create.CreateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = CreateCssTemplateRequest(
template_name="Bad Template",
css="",
)
result = await client.call_tool(
"create_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] is None
assert data["error"] is not None
@pytest.mark.asyncio
async def test_create_css_template_create_failed(mcp_server: object) -> None:
"""CssTemplateCreateFailedError is caught and returned as an error response."""
from superset.commands.css.exceptions import CssTemplateCreateFailedError
mock_command = MagicMock()
mock_command.run.side_effect = CssTemplateCreateFailedError()
with patch(
"superset.commands.css.create.CreateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = CreateCssTemplateRequest(
template_name="Failing Template",
css=".x { color: blue; }",
)
result = await client.call_tool(
"create_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] is None
assert data["error"] is not None
assert "Failed to create CSS template" in data["error"]
@pytest.mark.asyncio
async def test_create_css_template_unexpected_exception_is_reraised(
mcp_server: object,
) -> None:
"""Unexpected exceptions are re-raised (not swallowed as error responses)."""
from fastmcp.exceptions import ToolError
mock_command = MagicMock()
mock_command.run.side_effect = RuntimeError("unexpected database error")
with patch(
"superset.commands.css.create.CreateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
with pytest.raises((RuntimeError, ToolError)):
await client.call_tool(
"create_css_template",
{
"request": {
"template_name": "Theme",
"css": ".x { color: red; }",
}
},
)

View File

@@ -0,0 +1,219 @@
# 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.
from unittest.mock import MagicMock, Mock, patch
import pytest
from fastmcp import Client
from superset.mcp_service.app import mcp
from superset.mcp_service.css_template.schemas import UpdateCssTemplateRequest
from superset.utils import json
@pytest.fixture
def mcp_server():
return mcp
@pytest.fixture(autouse=True)
def mock_auth():
"""Mock authentication for all tests."""
with patch(
"superset.mcp_service.auth.get_user_from_request",
return_value=Mock(is_authenticated=True),
):
yield
# ---------------------------------------------------------------------------
# Schema tests
# ---------------------------------------------------------------------------
def test_update_css_template_request_valid_name_only() -> None:
req = UpdateCssTemplateRequest(id=1, template_name="New Name")
assert req.id == 1
assert req.template_name == "New Name"
assert req.css is None
def test_update_css_template_request_valid_css_only() -> None:
req = UpdateCssTemplateRequest(id=5, css=".body { color: blue; }")
assert req.id == 5
assert req.css == ".body { color: blue; }"
assert req.template_name is None
def test_update_css_template_request_strips_name_whitespace() -> None:
req = UpdateCssTemplateRequest(id=1, template_name=" Padded ")
assert req.template_name == "Padded"
def test_update_css_template_request_empty_name_fails() -> None:
from pydantic import ValidationError
with pytest.raises(ValidationError, match="template_name must not be empty"):
UpdateCssTemplateRequest(id=1, template_name=" ")
def test_update_css_template_request_name_too_long() -> None:
from pydantic import ValidationError
with pytest.raises(ValidationError):
UpdateCssTemplateRequest(id=1, template_name="x" * 251)
# ---------------------------------------------------------------------------
# Tool logic tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_update_css_template_success(mcp_server: object) -> None:
"""Happy path: template updated, all fields returned."""
mock_template = MagicMock()
mock_template.id = 3
mock_template.template_name = "Updated Theme"
mock_template.css = "body { background: #fff; }"
mock_command = MagicMock()
mock_command.run.return_value = mock_template
with patch(
"superset.commands.css.update.UpdateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = UpdateCssTemplateRequest(
id=3,
template_name="Updated Theme",
css="body { background: #fff; }",
)
result = await client.call_tool(
"update_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] == 3
assert data["template_name"] == "Updated Theme"
assert data["css"] == "body { background: #fff; }"
assert data["error"] is None
@pytest.mark.asyncio
async def test_update_css_template_no_fields_returns_error(
mcp_server: object,
) -> None:
"""Calling with neither template_name nor css returns a structured error."""
async with Client(mcp_server) as client:
result = await client.call_tool("update_css_template", {"request": {"id": 1}})
data = json.loads(result.content[0].text)
assert data["id"] is None
assert "At least one" in data["error"]
@pytest.mark.asyncio
async def test_update_css_template_not_found(mcp_server: object) -> None:
"""CssTemplateNotFoundError is caught and returned as an error response."""
from superset.commands.css.exceptions import CssTemplateNotFoundError
mock_command = MagicMock()
mock_command.run.side_effect = CssTemplateNotFoundError()
with patch(
"superset.commands.css.update.UpdateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = UpdateCssTemplateRequest(id=999, template_name="Ghost")
result = await client.call_tool(
"update_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] is None
assert "not found" in data["error"]
@pytest.mark.asyncio
async def test_update_css_template_invalid_error(mcp_server: object) -> None:
"""CssTemplateInvalidError is caught and returned as an error response."""
from superset.commands.css.exceptions import CssTemplateInvalidError
mock_command = MagicMock()
mock_command.run.side_effect = CssTemplateInvalidError()
with patch(
"superset.commands.css.update.UpdateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = UpdateCssTemplateRequest(id=1, template_name="Valid Name")
result = await client.call_tool(
"update_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] is None
assert data["error"] is not None
@pytest.mark.asyncio
async def test_update_css_template_update_failed(mcp_server: object) -> None:
"""CssTemplateUpdateFailedError is caught and returned as an error response."""
from superset.commands.css.exceptions import CssTemplateUpdateFailedError
mock_command = MagicMock()
mock_command.run.side_effect = CssTemplateUpdateFailedError()
with patch(
"superset.commands.css.update.UpdateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
request = UpdateCssTemplateRequest(id=1, css=".x { color: red; }")
result = await client.call_tool(
"update_css_template", {"request": request.model_dump()}
)
data = json.loads(result.content[0].text)
assert data["id"] is None
assert "Failed to update CSS template" in data["error"]
@pytest.mark.asyncio
async def test_update_css_template_unexpected_exception_is_reraised(
mcp_server: object,
) -> None:
"""Unexpected exceptions are re-raised (not swallowed as error responses)."""
from fastmcp.exceptions import ToolError
mock_command = MagicMock()
mock_command.run.side_effect = RuntimeError("unexpected database error")
with patch(
"superset.commands.css.update.UpdateCssTemplateCommand",
return_value=mock_command,
):
async with Client(mcp_server) as client:
with pytest.raises((RuntimeError, ToolError)):
await client.call_tool(
"update_css_template",
{"request": {"id": 1, "css": ".x { color: red; }"}},
)