MCP SQL Server

"""Unit tests for QueryColumnsService.""" import pytest from mcp_sql.tools.query_columns import QueryColumnsService @pytest.mark.asyncio class TestQueryColumnsService: """Test cases for QueryColumnsService.""" async def test_name_property(self, tool_dependencies): """Test that the tool has the correct name.""" service = QueryColumnsService(**tool_dependencies) assert service.name == "query_table_with_columns" async def test_description_property(self, tool_dependencies): """Test that the tool has a description.""" service = QueryColumnsService(**tool_dependencies) assert len(service.description) > 0 assert "column" in service.description.lower() async def test_execute_returns_column_data(self, tool_dependencies, mock_context): """Test that execute returns data for specified columns.""" service = QueryColumnsService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", columns=["id", "name"], database="test_db", server_name="test_server", limit=10 ) assert isinstance(result, list) assert len(result) == 2 # 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_single_column(self, tool_dependencies, mock_context): """Test execute with a single column.""" service = QueryColumnsService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", columns=["id"], database="test_db", server_name="test_server" ) assert isinstance(result, list) tool_dependencies["executor"].query_columns.assert_called_once() async def test_execute_with_custom_limit(self, tool_dependencies, mock_context): """Test execute with custom limit.""" service = QueryColumnsService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", columns=["id", "name"], database="test_db", server_name="test_server", limit=50 ) assert isinstance(result, list) # Check that limit was passed correctly call_args = tool_dependencies["executor"].query_columns.call_args assert call_args[0][3] == 50 # Fourth 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 = QueryColumnsService(**tool_dependencies) result = await service.execute( mock_context, table="test_table", columns=["id"], 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 = QueryColumnsService(**tool_dependencies) result = await service.execute( mock_context, table="", columns=["id"], database="test_db", server_name="test_server" ) assert isinstance(result, list) assert "error" in result[0] async def test_execute_with_invalid_credentials(self, tool_dependencies_invalid, mock_context): """Test execute with invalid credentials.""" service = QueryColumnsService(**tool_dependencies_invalid) result = await service.execute( mock_context, table="test_table", columns=["id"], 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 = QueryColumnsService(**tool_dependencies) await service.execute( mock_context, table="test_table", columns=["id", "name"], database="test_db", server_name="test_server" ) tool_dependencies["executor"].query_columns.assert_called_once()