add ci tests

Author: iskakaushik

.github/workflows/ci.yaml

@@ -0,0 +1,56 @@
+name: CI
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      clickhouse:
+        image: clickhouse/clickhouse-server:24.10
+        ports:
+          - 9000:9000
+          - 8123:8123
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Setup Python
+        run: uv python install 3.13
+
+      - name: Install Project
+        run: uv sync
+
+      # Wait for ClickHouse to be ready
+      - name: Wait for ClickHouse service
+        run: |
+          for i in {1..30}; do
+            if clickhouse-client --query "SELECT 1"; then
+              echo "ClickHouse is ready"
+              exit 0
+            fi
+            echo "Waiting for ClickHouse to be ready..."
+            sleep 2
+          done
+          echo "ClickHouse did not become ready in time" && exit 1
+
+      - name: Run tests
+        env:
+          CLICKHOUSE_HOST: "localhost"
+          CLICKHOUSE_PORT: "8123"
+          CLICKHOUSE_USER: "default"
+          CLICKHOUSE_PASSWORD: ""
+        run: |
+          uv run pytest tests
+
+      - name: Lint with Ruff
+        run: ruff check .

mcp_clickhouse/__init__.py

@@ -0,0 +1,13 @@
+from .mcp_server import (
+    list_databases,
+    list_tables,
+    run_select_query,
+    create_clickhouse_client,
+)
+
+__all__ = [
+    "list_databases",
+    "list_tables",
+    "run_select_query",
+    "create_clickhouse_client",
+]

mcp_clickhouse/mcp_server.py

@@ -10,15 +10,14 @@
 
 # Configure logging
 logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
 logger = logging.getLogger(MCP_SERVER_NAME)
 
 load_dotenv()
 
 deps = [
-    "clickhouse-connect", 
+    "clickhouse-connect",
     "python-dotenv",
     "uvicorn",
 ]

pyproject.toml

@@ -20,7 +20,8 @@ Home = "https://github.com/iskakaushik/mcp-clickhouse"
 
 [project.optional-dependencies]
 dev = [
-    "ruff"
+    "ruff",
+    "pytest"
 ]
 
 [tool.hatch.build.targets.wheel]

tests/test_tool.py

@@ -0,0 +1,71 @@
+import unittest
+from dotenv import load_dotenv
+from mcp_clickhouse import list_databases, list_tables, run_select_query, create_clickhouse_client
+
+load_dotenv()
+
+
+class TestClickhouseTools(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        """Set up the environment before tests."""
+        cls.client = create_clickhouse_client()
+
+        # Prepare test database and table
+        cls.test_db = "test_tool_db"
+        cls.test_table = "test_table"
+        cls.client.command(f"CREATE DATABASE IF NOT EXISTS {cls.test_db}")
+        cls.client.command(f"""
+            CREATE TABLE IF NOT EXISTS {cls.test_db}.{cls.test_table} (
+                id UInt32,
+                name String
+            ) ENGINE = MergeTree()
+            ORDER BY id
+        """)
+        cls.client.command(f"""
+            INSERT INTO {cls.test_db}.{cls.test_table} (id, name) VALUES (1, 'Alice'), (2, 'Bob')
+        """)
+
+    @classmethod
+    def tearDownClass(cls):
+        """Clean up the environment after tests."""
+        cls.client.command(f"DROP DATABASE IF EXISTS {cls.test_db}")
+
+    def test_list_databases(self):
+        """Test listing databases."""
+        result = list_databases()
+        self.assertIn(self.test_db, result)
+
+    def test_list_tables_without_like(self):
+        """Test listing tables without a 'LIKE' filter."""
+        result = list_tables(self.test_db)
+        self.assertIsInstance(result, list)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["name"], self.test_table)
+
+    def test_list_tables_with_like(self):
+        """Test listing tables with a 'LIKE' filter."""
+        result = list_tables(self.test_db, like=f"{self.test_table}%")
+        self.assertIsInstance(result, list)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["name"], self.test_table)
+
+    def test_run_select_query_success(self):
+        """Test running a SELECT query successfully."""
+        query = f"SELECT * FROM {self.test_db}.{self.test_table}"
+        result = run_select_query(query)
+        self.assertIsInstance(result, list)
+        self.assertEqual(len(result), 2)
+        self.assertEqual(result[0]["id"], 1)
+        self.assertEqual(result[0]["name"], "Alice")
+
+    def test_run_select_query_failure(self):
+        """Test running a SELECT query with an error."""
+        query = f"SELECT * FROM {self.test_db}.non_existent_table"
+        result = run_select_query(query)
+        self.assertIsInstance(result, str)
+        self.assertIn("error running query", result)
+
+
+if __name__ == "__main__":
+    unittest.main()