MCP SQL Server

"""Unit tests for ExploreTableService.""" import pytest from mcp_sql.tools.explore_table import ExploreTableService @pytest.mark.asyncio class TestExploreTableService: """Test cases for ExploreTableService.""" async def test_name_property(self, tool_dependencies): """Test that the tool has the correct name.""" service = ExploreTableService(**tool_dependencies) assert service.name == "explore_table" async def test_description_property(self, tool_dependencies): """Test that the tool has a description.""" service = ExploreTableService(**tool_dependencies) assert len(service.description) > 0 assert "table" in service.description.lower() async def test_execute_returns_sample_data(self, tool_dependencies, mock_context): """Test that execute returns sample table data.""" service = ExploreTableService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", database="test_db", server_name="test_server", limit=5 ) assert isinstance(result, list) assert len(result) == 3 # Based on mock data assert all(isinstance(row, dict) for row in result) assert "id" in result[0] assert "name" in result[0] async def test_execute_with_custom_limit(self, tool_dependencies, mock_context): """Test execute with custom limit.""" service = ExploreTableService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", database="test_db", server_name="test_server", limit=10 ) assert isinstance(result, list) tool_dependencies["executor"].explore_table.assert_called_once() # Check that limit was passed correctly call_args = tool_dependencies["executor"].explore_table.call_args assert call_args[0][2] == 10 # Third argument is limit async def test_execute_without_database(self, tool_dependencies, mock_context, mock_credentials): """Test execute without database name.""" mock_credentials.database = None service = ExploreTableService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", server_name="test_server" ) assert isinstance(result, list) assert len(result) == 1 assert "error" in result[0] assert "Database and table names are required" in result[0]["error"] async def test_execute_without_table(self, tool_dependencies, mock_context): """Test execute without table name.""" service = ExploreTableService(**tool_dependencies) result = await service.execute( mock_context, table="", database="test_db", server_name="test_server" ) assert isinstance(result, list) assert len(result) == 1 assert "error" in result[0] async def test_execute_with_invalid_credentials(self, tool_dependencies_invalid, mock_context): """Test execute with invalid credentials.""" service = ExploreTableService(**tool_dependencies_invalid) result = await service.execute( mock_context, table="test_table", database="test_db" ) assert isinstance(result, list) assert "error" in result[0] assert "Missing credentials" in result[0]["error"] async def test_execute_calls_executor(self, tool_dependencies, mock_context): """Test that execute calls the query executor.""" service = ExploreTableService(**tool_dependencies) await service.execute( mock_context, table="test_table", database="test_db", server_name="test_server" ) tool_dependencies["executor"].explore_table.assert_called_once()