Files
superset2/tests/unit_tests/mcp_service/test_error_handling.py
Amin Ghadersohi 91a3214ed4 update: wip
2025-07-30 14:20:38 -04:00

84 lines
4.0 KiB
Python

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Unit tests for MCP tool error handling and parameter validation
"""
import logging
from unittest.mock import patch
import pytest
from superset.mcp_service.pydantic_schemas.dashboard_schemas import DashboardAvailableFilters, DashboardList
from superset.mcp_service.pydantic_schemas.dataset_schemas import DatasetAvailableFilters, DatasetList
from superset.mcp_service.tools.dashboard import get_dashboard_available_filters, list_dashboards
from superset.mcp_service.tools.dataset import get_dataset_available_filters, list_datasets
from fastmcp.exceptions import ToolError
from superset.daos.dashboard import DashboardDAO
from superset.daos.chart import ChartDAO
from superset.daos.dataset import DatasetDAO
from flask import Flask, g
from flask_login import AnonymousUserMixin
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class TestErrorHandling:
"""Test error handling and parameter validation in MCP tools"""
@patch('superset.daos.dashboard.DashboardDAO.list')
def test_list_dashboards_exception_handling(self, mock_list):
mock_list.side_effect = Exception("Unexpected error")
with pytest.raises(Exception) as excinfo:
list_dashboards()
assert "Unexpected error" in str(excinfo.value)
def test_get_dashboard_available_filters_exception_handling(self):
result = get_dashboard_available_filters()
assert isinstance(result, DashboardAvailableFilters)
assert hasattr(result, "filters")
assert hasattr(result, "operators")
assert hasattr(result, "columns")
@patch('superset.daos.dataset.DatasetDAO.list')
def test_list_datasets_exception_handling(self, mock_list):
mock_list.side_effect = Exception("API request failed")
with pytest.raises(Exception) as excinfo:
list_datasets()
assert "API request failed" in str(excinfo.value)
def test_list_dashboards_parameter_types(self):
from pydantic import ValidationError
with patch('superset.daos.dashboard.DashboardDAO.list') as mock_list:
mock_list.return_value = ([], 0)
with pytest.raises(ValidationError):
list_dashboards(filters='[{"col": "test", "opr": "eq", "value": "value"}]')
with pytest.raises(ValidationError):
list_dashboards(filters=[{"col": "test", "opr": "eq", "value": "value"}])
list_dashboards(select_columns="id,dashboard_title")
list_dashboards(select_columns=["id", "dashboard_title"])
def test_list_datasets_parameter_types(self):
from pydantic import ValidationError
with patch('superset.daos.dataset.DatasetDAO.list') as mock_list:
mock_list.return_value = ([], 0)
with pytest.raises(ValidationError):
list_datasets(filters='[{"col": "test", "opr": "eq", "value": "value"}]')
with pytest.raises(ValidationError):
list_datasets(filters=[{"col": "test", "opr": "eq", "value": "value"}])
list_datasets(select_columns="id,table_name")
list_datasets(select_columns=["id", "table_name"])
# Example: test for missing required param, extra param, and malformed input would be in protocol/integration tests