This is an automated email from the ASF dual-hosted git repository. aminghadersohi pushed a commit to branch amin/mcp-create-css-template in repository https://gitbox.apache.org/repos/asf/superset.git
commit 9b6ee2522c089c6a03084640ad3922db97f1fb19 Author: Amin Ghadersohi <[email protected]> AuthorDate: Sat May 30 04:39:19 2026 +0000 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 --- superset/commands/css/create.py | 46 +++++ superset/commands/css/exceptions.py | 20 +- superset/commands/css/update.py | 53 +++++ superset/mcp_service/app.py | 4 + superset/mcp_service/css_template/schemas.py | 104 ++++++++++ superset/mcp_service/css_template/tool/__init__.py | 4 + .../css_template/tool/create_css_template.py | 95 +++++++++ .../css_template/tool/update_css_template.py | 111 +++++++++++ .../css_template/tool/test_create_css_template.py | 192 ++++++++++++++++++ .../css_template/tool/test_update_css_template.py | 219 +++++++++++++++++++++ 10 files changed, 847 insertions(+), 1 deletion(-) diff --git a/superset/commands/css/create.py b/superset/commands/css/create.py new file mode 100644 index 00000000000..a50e38992ae --- /dev/null +++ b/superset/commands/css/create.py @@ -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() diff --git a/superset/commands/css/exceptions.py b/superset/commands/css/exceptions.py index 95517581067..04eff648671 100644 --- a/superset/commands/css/exceptions.py +++ b/superset/commands/css/exceptions.py @@ -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.") diff --git a/superset/commands/css/update.py b/superset/commands/css/update.py new file mode 100644 index 00000000000..7bdb6ece6e6 --- /dev/null +++ b/superset/commands/css/update.py @@ -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() diff --git a/superset/mcp_service/app.py b/superset/mcp_service/app.py index 9164254f863..b870c43852b 100644 --- a/superset/mcp_service/app.py +++ b/superset/mcp_service/app.py @@ -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, diff --git a/superset/mcp_service/css_template/schemas.py b/superset/mcp_service/css_template/schemas.py index b17140e3f8c..93907e19033 100644 --- a/superset/mcp_service/css_template/schemas.py +++ b/superset/mcp_service/css_template/schemas.py @@ -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",)) diff --git a/superset/mcp_service/css_template/tool/__init__.py b/superset/mcp_service/css_template/tool/__init__.py index ecd6860e381..b8a01d2cbba 100644 --- a/superset/mcp_service/css_template/tool/__init__.py +++ b/superset/mcp_service/css_template/tool/__init__.py @@ -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", ] diff --git a/superset/mcp_service/css_template/tool/create_css_template.py b/superset/mcp_service/css_template/tool/create_css_template.py new file mode 100644 index 00000000000..b12be40cc72 --- /dev/null +++ b/superset/mcp_service/css_template/tool/create_css_template.py @@ -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 diff --git a/superset/mcp_service/css_template/tool/update_css_template.py b/superset/mcp_service/css_template/tool/update_css_template.py new file mode 100644 index 00000000000..7024d21e784 --- /dev/null +++ b/superset/mcp_service/css_template/tool/update_css_template.py @@ -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 diff --git a/tests/unit_tests/mcp_service/css_template/tool/test_create_css_template.py b/tests/unit_tests/mcp_service/css_template/tool/test_create_css_template.py new file mode 100644 index 00000000000..7af212a0d52 --- /dev/null +++ b/tests/unit_tests/mcp_service/css_template/tool/test_create_css_template.py @@ -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 + + [email protected] +def mcp_server(): + return mcp + + [email protected](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 +# --------------------------------------------------------------------------- + + [email protected] +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 + + [email protected] +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 + + [email protected] +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"] + + [email protected] +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; }", + } + }, + ) diff --git a/tests/unit_tests/mcp_service/css_template/tool/test_update_css_template.py b/tests/unit_tests/mcp_service/css_template/tool/test_update_css_template.py new file mode 100644 index 00000000000..c451ccf7835 --- /dev/null +++ b/tests/unit_tests/mcp_service/css_template/tool/test_update_css_template.py @@ -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 + + [email protected] +def mcp_server(): + return mcp + + [email protected](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 +# --------------------------------------------------------------------------- + + [email protected] +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 + + [email protected] +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"] + + [email protected] +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"] + + [email protected] +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 + + [email protected] +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"] + + [email protected] +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; }"}}, + )
