feat(mcp): restore self-lookup via created_by_me flag (#39638)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/unit_tests/mcp_service/system/tool/test_get_current_user.py

@@ -461,7 +461,7 @@ class TestGetInstanceInfoCurrentUserViaMCP:
 
 
 def test_chart_filter_rejects_created_by_fk() -> None:
-    """Test that ChartFilter rejects user-directory columns."""
+    """created_by_fk is not a valid ChartFilter column; use created_by_me instead."""
     with pytest.raises(ValidationError):
         ChartFilter(col="created_by_fk", opr="eq", value=42)
 
@@ -473,7 +473,7 @@ def test_chart_filter_rejects_invalid_column():
 
 
 def test_dashboard_filter_rejects_created_by_fk():
-    """Test that DashboardFilter rejects user-directory columns."""
+    """created_by_fk is not a valid DashboardFilter column; use created_by_me."""
     with pytest.raises(ValidationError):
         DashboardFilter(col="created_by_fk", opr="eq", value=42)
 

tests/unit_tests/mcp_service/system/tool/test_mcp_core.py

@@ -18,6 +18,7 @@
 from datetime import datetime
 from types import SimpleNamespace
 from typing import Any, Dict, List
+from unittest.mock import Mock, patch
 
 import pytest
 from pydantic import BaseModel
@@ -126,6 +127,45 @@ def test_model_list_tool_with_filters_and_columns():
     assert "id" in result.columns_loaded
 
 
+def test_model_list_tool_keeps_single_filter_when_created_by_me_is_used():
+    current_user = Mock()
+    current_user.is_authenticated = True
+    current_user.id = 42
+
+    captured = {}
+
+    class CapturingDAO:
+        @classmethod
+        def list(cls, column_operators=None, **kwargs):
+            captured["filters"] = column_operators
+            return [], 0
+
+    tool = ModelListCore(
+        dao_class=CapturingDAO,
+        output_schema=DummyOutputSchema,
+        item_serializer=dummy_serializer,
+        filter_type=None,
+        default_columns=["id", "name"],
+        search_columns=["name"],
+        list_field_name="items",
+        output_list_schema=DummyListSchema,
+    )
+
+    with patch(
+        "superset.mcp_service.mcp_core.get_current_user",
+        return_value=current_user,
+    ):
+        tool.run_tool(
+            filters={"col": "name", "opr": "eq", "value": "foo"},
+            created_by_me=True,
+        )
+
+    assert len(captured["filters"]) == 2
+    assert captured["filters"][0].col == "created_by_fk"
+    assert captured["filters"][0].value == 42
+    assert captured["filters"][1] == {"col": "name", "opr": "eq", "value": "foo"}
+
+
 def test_model_list_tool_rejects_only_user_directory_select_columns():
     tool = ModelListCore(
         dao_class=DummyDAO,
@@ -177,6 +217,160 @@ def test_model_list_tool_allows_order_column_when_sortable_columns_not_declared(
     tool.run_tool(order_column="name")
 
 
+def test_model_list_tool_injects_current_user_id_for_created_by_me():
+    """created_by_me=True adds a created_by_fk filter with the current user's ID."""
+    current_user = Mock()
+    current_user.is_authenticated = True
+    current_user.id = 42
+
+    captured = {}
+
+    class CapturingDAO:
+        @classmethod
+        def list(cls, column_operators=None, **kwargs):
+            captured["filters"] = column_operators
+            return [], 0
+
+    tool = ModelListCore(
+        dao_class=CapturingDAO,
+        output_schema=DummyOutputSchema,
+        item_serializer=dummy_serializer,
+        filter_type=None,
+        default_columns=["id", "name"],
+        search_columns=["name"],
+        list_field_name="items",
+        output_list_schema=DummyListSchema,
+    )
+
+    with patch(
+        "superset.mcp_service.mcp_core.get_current_user",
+        return_value=current_user,
+    ):
+        tool.run_tool(created_by_me=True)
+
+    assert captured["filters"][0].col == "created_by_fk"
+    assert captured["filters"][0].value == 42
+
+
+def test_model_list_tool_created_by_me_requires_authenticated_user():
+    """created_by_me=True raises when no authenticated user is present."""
+    current_user = Mock()
+    current_user.is_authenticated = False
+
+    tool = ModelListCore(
+        dao_class=DummyDAO,
+        output_schema=DummyOutputSchema,
+        item_serializer=dummy_serializer,
+        filter_type=None,
+        default_columns=["id", "name"],
+        search_columns=["name"],
+        list_field_name="items",
+        output_list_schema=DummyListSchema,
+    )
+
+    with patch(
+        "superset.mcp_service.mcp_core.get_current_user",
+        return_value=current_user,
+    ):
+        with pytest.raises(ValueError, match="authenticated user"):
+            tool.run_tool(created_by_me=True)
+
+
+def test_model_list_tool_injects_current_user_id_for_owned_by_me():
+    """owned_by_me=True adds an owner filter with the current user's ID."""
+    current_user = Mock()
+    current_user.is_authenticated = True
+    current_user.id = 99
+
+    captured = {}
+
+    class CapturingDAO:
+        @classmethod
+        def list(cls, column_operators=None, **kwargs):
+            captured["filters"] = column_operators
+            return [], 0
+
+    tool = ModelListCore(
+        dao_class=CapturingDAO,
+        output_schema=DummyOutputSchema,
+        item_serializer=dummy_serializer,
+        filter_type=None,
+        default_columns=["id", "name"],
+        search_columns=["name"],
+        list_field_name="items",
+        output_list_schema=DummyListSchema,
+    )
+
+    with patch(
+        "superset.mcp_service.mcp_core.get_current_user",
+        return_value=current_user,
+    ):
+        tool.run_tool(owned_by_me=True)
+
+    assert captured["filters"][0].col == "owner"
+    assert captured["filters"][0].value == 99
+
+
+def test_model_list_tool_both_flags_uses_combined_or_filter():
+    """created_by_me=True + owned_by_me=True generates a single OR filter."""
+    current_user = Mock()
+    current_user.is_authenticated = True
+    current_user.id = 55
+
+    captured = {}
+
+    class CapturingDAO:
+        @classmethod
+        def list(cls, column_operators=None, **kwargs):
+            captured["filters"] = column_operators
+            return [], 0
+
+    tool = ModelListCore(
+        dao_class=CapturingDAO,
+        output_schema=DummyOutputSchema,
+        item_serializer=dummy_serializer,
+        filter_type=None,
+        default_columns=["id", "name"],
+        search_columns=["name"],
+        list_field_name="items",
+        output_list_schema=DummyListSchema,
+    )
+
+    with patch(
+        "superset.mcp_service.mcp_core.get_current_user",
+        return_value=current_user,
+    ):
+        tool.run_tool(created_by_me=True, owned_by_me=True)
+
+    assert len(captured["filters"]) == 1
+    assert captured["filters"][0].col == "created_by_fk_or_owner"
+    assert captured["filters"][0].value == 55
+
+
+def test_model_list_tool_owned_by_me_requires_authenticated_user():
+    """owned_by_me=True raises when no authenticated user is present."""
+    current_user = Mock()
+    current_user.is_authenticated = False
+
+    tool = ModelListCore(
+        dao_class=DummyDAO,
+        output_schema=DummyOutputSchema,
+        item_serializer=dummy_serializer,
+        filter_type=None,
+        default_columns=["id", "name"],
+        search_columns=["name"],
+        list_field_name="items",
+        output_list_schema=DummyListSchema,
+    )
+
+    with patch(
+        "superset.mcp_service.mcp_core.get_current_user",
+        return_value=current_user,
+    ):
+        with pytest.raises(ValueError, match="authenticated user"):
+            tool.run_tool(owned_by_me=True)
+
+
 def test_user_directory_fields_include_last_saved_relationships():
     assert "last_saved_by" in USER_DIRECTORY_FIELDS
     assert "last_saved_by_name" in USER_DIRECTORY_FIELDS