This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new 79bef3f test: add real Doris transport matrix (#102)
79bef3f is described below
commit 79bef3fa1391fa9ddf2815bb3552d854ff8ec43c
Author: Yijia Su <[email protected]>
AuthorDate: Wed Jul 29 18:27:36 2026 +0800
test: add real Doris transport matrix (#102)
---
doris_mcp_server/utils/db.py | 7 +-
doris_mcp_server/utils/query_executor.py | 6 +-
test/integration/test_real_doris_transports.py | 411 +++++++++++++++++++++++++
test/utils/test_doris_user_pool_manager.py | 12 +
test/utils/test_query_error_classification.py | 29 ++
5 files changed, 461 insertions(+), 4 deletions(-)
diff --git a/doris_mcp_server/utils/db.py b/doris_mcp_server/utils/db.py
index 1151a98..32d849c 100644
--- a/doris_mcp_server/utils/db.py
+++ b/doris_mcp_server/utils/db.py
@@ -1619,16 +1619,17 @@ class DorisConnectionManager:
async def _warmup_pool(self):
"""Warm up connection pool by creating initial connections"""
- self.logger.info(f"🔥 Warming up connection pool with
{self.pool_warmup_size} connections")
+ warmup_size = min(self.pool_warmup_size, self.maxsize)
+ self.logger.info(f"🔥 Warming up connection pool with {warmup_size}
connections")
warmup_connections = []
try:
# Acquire connections to force pool to create them
- for i in range(self.pool_warmup_size):
+ for i in range(warmup_size):
try:
conn = await self.pool.acquire()
warmup_connections.append(conn)
- self.logger.debug(f"Warmed up connection
{i+1}/{self.pool_warmup_size}")
+ self.logger.debug(f"Warmed up connection
{i+1}/{warmup_size}")
except Exception as e:
self.logger.warning(f"Failed to warm up connection {i+1}:
{e}")
break
diff --git a/doris_mcp_server/utils/query_executor.py
b/doris_mcp_server/utils/query_executor.py
index 43cfb1b..269cc9d 100644
--- a/doris_mcp_server/utils/query_executor.py
+++ b/doris_mcp_server/utils/query_executor.py
@@ -996,7 +996,11 @@ class DorisQueryExecutor:
"error_type": "syntax_error",
"user_message": "SQL syntax error. Please check your query
syntax."
}
- elif "access denied" in error_msg_lower or "permission" in
error_msg_lower:
+ elif (
+ "access denied" in error_msg_lower
+ or "command denied" in error_msg_lower
+ or "permission" in error_msg_lower
+ ):
return {
"error_type": "permission_denied",
"user_message": "Access denied. You don't have permission to
execute this query."
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
new file mode 100644
index 0000000..6135156
--- /dev/null
+++ b/test/integration/test_real_doris_transports.py
@@ -0,0 +1,411 @@
+#!/usr/bin/env python3
+# 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.
+"""Opt-in production-process integration tests against a real Doris instance.
+
+Set ``DORIS_REAL_INTEGRATION=1`` and provide ``DORIS_REAL_HOST``,
+``DORIS_REAL_USER``, and ``DORIS_REAL_DATABASE``. ``DORIS_REAL_PORT`` defaults
+to 9030 and ``DORIS_REAL_PASSWORD`` may be empty.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+import secrets
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from collections.abc import AsyncIterator
+from contextlib import AsyncExitStack, asynccontextmanager, suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pymysql
+import pytest
+from mcp import Client, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+pytestmark = [
+ pytest.mark.integration,
+ pytest.mark.skipif(
+ os.getenv("DORIS_REAL_INTEGRATION") != "1",
+ reason="set DORIS_REAL_INTEGRATION=1 to run real Doris tests",
+ ),
+]
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+SAFE_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+
+@dataclass(frozen=True)
+class RealDorisSettings:
+ host: str
+ port: int
+ user: str
+ password: str
+ database: str
+
+
+@dataclass
+class DorisSandbox:
+ settings: RealDorisSettings
+ admin_connection: pymysql.Connection
+ table: str
+ readonly_user: str
+ readonly_password: str
+ marker: str
+
+ @property
+ def qualified_table(self) -> str:
+ return f"`{self.settings.database}`.`{self.table}`"
+
+ def kill_connection(self, connection_id: int) -> None:
+ with self.admin_connection.cursor() as cursor:
+ cursor.execute(f"KILL CONNECTION {int(connection_id)}")
+
+
+def _real_doris_settings() -> RealDorisSettings:
+ required = {
+ name: os.getenv(name, "").strip()
+ for name in (
+ "DORIS_REAL_HOST",
+ "DORIS_REAL_USER",
+ "DORIS_REAL_DATABASE",
+ )
+ }
+ missing = [name for name, value in required.items() if not value]
+ if missing:
+ pytest.fail(
+ "missing real Doris settings: " + ", ".join(sorted(missing)),
+ pytrace=False,
+ )
+ if not SAFE_IDENTIFIER.fullmatch(required["DORIS_REAL_DATABASE"]):
+ pytest.fail("DORIS_REAL_DATABASE must be a safe SQL identifier",
pytrace=False)
+ return RealDorisSettings(
+ host=required["DORIS_REAL_HOST"],
+ port=int(os.getenv("DORIS_REAL_PORT", "9030")),
+ user=required["DORIS_REAL_USER"],
+ password=os.getenv("DORIS_REAL_PASSWORD", ""),
+ database=required["DORIS_REAL_DATABASE"],
+ )
+
+
[email protected]
+def doris_sandbox() -> DorisSandbox:
+ settings = _real_doris_settings()
+ suffix = secrets.token_hex(6)
+ table = f"mcp_real_it_{suffix}"
+ readonly_user = f"mcp_it_ro_{suffix}"
+ readonly_password = secrets.token_hex(16)
+ marker = secrets.token_hex(12)
+ user_identity = f"'{readonly_user}'@'%'"
+ qualified_table = f"`{settings.database}`.`{table}`"
+ admin_connection = pymysql.connect(
+ host=settings.host,
+ port=settings.port,
+ user=settings.user,
+ password=settings.password,
+ database=settings.database,
+ autocommit=True,
+ )
+
+ try:
+ with admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"""
+ CREATE TABLE {qualified_table} (
+ id BIGINT,
+ marker VARCHAR(64)
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ )
+ cursor.execute(
+ f"CREATE USER {user_identity} IDENTIFIED BY
'{readonly_password}'"
+ )
+ cursor.execute(f"GRANT SELECT_PRIV ON {qualified_table} TO
{user_identity}")
+
+ yield DorisSandbox(
+ settings=settings,
+ admin_connection=admin_connection,
+ table=table,
+ readonly_user=readonly_user,
+ readonly_password=readonly_password,
+ marker=marker,
+ )
+ finally:
+ with suppress(Exception), admin_connection.cursor() as cursor:
+ cursor.execute(f"DROP USER IF EXISTS {user_identity}")
+ with suppress(Exception), admin_connection.cursor() as cursor:
+ cursor.execute(f"DROP TABLE IF EXISTS {qualified_table}")
+ admin_connection.close()
+
+
+def _server_environment(
+ settings: RealDorisSettings,
+ *,
+ user: str,
+ password: str,
+) -> dict[str, str]:
+ return {
+ **os.environ,
+ "DORIS_HOST": settings.host,
+ "DORIS_PORT": str(settings.port),
+ "DORIS_USER": user,
+ "DORIS_PASSWORD": password,
+ "DORIS_DATABASE": settings.database,
+ "DORIS_MAX_CONNECTIONS": "1",
+ "DORIS_CONNECTION_TIMEOUT": "3",
+ "ENABLE_SECURITY_CHECK": "false",
+ "ENABLE_TOKEN_AUTH": "false",
+ "ENABLE_JWT_AUTH": "false",
+ "ENABLE_OAUTH_AUTH": "false",
+ "OAUTH_ENABLED": "false",
+ "ENABLE_DORIS_OAUTH_AUTH": "false",
+ "LOG_LEVEL": "ERROR",
+ "PYTHONUNBUFFERED": "1",
+ }
+
+
+def _free_tcp_port() -> int:
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ return int(sock.getsockname()[1])
+
+
+def _read_process_log(log_file: Any) -> str:
+ log_file.flush()
+ log_file.seek(0)
+ return log_file.read().decode("utf-8", errors="replace")[-4000:]
+
+
+async def _stop_process(process: asyncio.subprocess.Process) -> None:
+ if process.returncode is not None:
+ return
+ process.terminate()
+ try:
+ await asyncio.wait_for(process.wait(), timeout=10)
+ except TimeoutError:
+ process.kill()
+ await process.wait()
+
+
+async def _wait_for_http_server(
+ process: asyncio.subprocess.Process,
+ port: int,
+ log_file: Any,
+) -> None:
+ deadline = time.monotonic() + 30
+ async with httpx.AsyncClient(timeout=0.5) as client:
+ while time.monotonic() < deadline:
+ if process.returncode is not None:
+ pytest.fail(
+ "HTTP MCP process exited during startup:\n"
+ + _read_process_log(log_file),
+ pytrace=False,
+ )
+ try:
+ response = await client.get(f"http://127.0.0.1:{port}/health";)
+ if response.status_code == 200:
+ return
+ except httpx.HTTPError:
+ pass
+ await asyncio.sleep(0.1)
+ pytest.fail(
+ "HTTP MCP process did not become healthy:\n" +
_read_process_log(log_file),
+ pytrace=False,
+ )
+
+
+@asynccontextmanager
+async def _http_client(environment: dict[str, str]) -> AsyncIterator[Client]:
+ port = _free_tcp_port()
+ with tempfile.TemporaryFile() as log_file:
+ process = await asyncio.create_subprocess_exec(
+ sys.executable,
+ "-m",
+ "doris_mcp_server",
+ "--transport",
+ "http",
+ "--host",
+ "127.0.0.1",
+ "--port",
+ str(port),
+ cwd=PROJECT_ROOT,
+ env=environment,
+ stdout=log_file,
+ stderr=subprocess.STDOUT,
+ )
+ try:
+ await _wait_for_http_server(process, port, log_file)
+ async with Client(
+ f"http://127.0.0.1:{port}/mcp";,
+ read_timeout_seconds=15,
+ ) as client:
+ yield client
+ finally:
+ await _stop_process(process)
+
+
+@asynccontextmanager
+async def _stdio_client(environment: dict[str, str]) -> AsyncIterator[Client]:
+ parameters = StdioServerParameters(
+ command=sys.executable,
+ args=["-m", "doris_mcp_server", "--transport", "stdio"],
+ cwd=PROJECT_ROOT,
+ env=environment,
+ )
+ with tempfile.TemporaryFile(mode="w+") as log_file:
+ stack = AsyncExitStack()
+ try:
+ client = await stack.enter_async_context(
+ Client(
+ stdio_client(parameters, errlog=log_file),
+ read_timeout_seconds=15,
+ )
+ )
+ except Exception:
+ log_file.seek(0)
+ pytest.fail(
+ "STDIO MCP process failed during startup:\n" +
log_file.read()[-4000:],
+ pytrace=False,
+ )
+ try:
+ yield client
+ finally:
+ await stack.aclose()
+
+
+def _transport_client(
+ transport: str,
+ environment: dict[str, str],
+) -> Any:
+ if transport == "http":
+ return _http_client(environment)
+ return _stdio_client(environment)
+
+
+async def _exec_query(
+ client: Client,
+ sql: str,
+ *,
+ timeout: int = 5,
+) -> tuple[Any, dict[str, Any]]:
+ result = await client.call_tool(
+ "exec_query",
+ {
+ "sql": sql,
+ "max_rows": 20,
+ "timeout": timeout,
+ },
+ )
+ assert isinstance(result.structured_content, dict)
+ return result, result.structured_content
+
+
[email protected]("transport", ["http", "stdio"])
+async def test_real_doris_read_write_permission_timeout_and_recovery(
+ transport: str,
+ doris_sandbox: DorisSandbox,
+) -> None:
+ admin_environment = _server_environment(
+ doris_sandbox.settings,
+ user=doris_sandbox.settings.user,
+ password=doris_sandbox.settings.password,
+ )
+ async with _transport_client(transport, admin_environment) as client:
+ read_result, read_payload = await _exec_query(client, "SELECT 42 AS
answer")
+ assert read_result.is_error is False
+ assert read_payload["success"] is True
+ assert read_payload["data"][0]["answer"] == 42
+
+ write_result, write_payload = await _exec_query(
+ client,
+ f"""
+ INSERT INTO {doris_sandbox.qualified_table}
+ VALUES (1, '{doris_sandbox.marker}')
+ """,
+ )
+ assert write_result.is_error is False
+ assert write_payload["success"] is True
+
+ verify_result, verify_payload = await _exec_query(
+ client,
+ f"""
+ SELECT marker
+ FROM {doris_sandbox.qualified_table}
+ WHERE id = 1
+ """,
+ )
+ assert verify_result.is_error is False
+ assert verify_payload["data"] == [{"marker": doris_sandbox.marker}]
+
+ timeout_result, timeout_payload = await _exec_query(
+ client,
+ "SELECT SLEEP(2) AS slept",
+ timeout=1,
+ )
+ assert timeout_result.is_error is True
+ assert timeout_payload["success"] is False
+ assert timeout_payload["error_type"] == "timeout"
+
+ _, connection_payload = await _exec_query(
+ client,
+ "SELECT CONNECTION_ID() AS connection_id",
+ )
+ connection_id = int(connection_payload["data"][0]["connection_id"])
+ doris_sandbox.kill_connection(connection_id)
+
+ recovered_result, recovered_payload = await _exec_query(
+ client,
+ "SELECT 1 AS recovered",
+ )
+ assert recovered_result.is_error is False
+ assert recovered_payload["success"] is True
+ assert recovered_payload["data"][0]["recovered"] == 1
+
+ readonly_environment = _server_environment(
+ doris_sandbox.settings,
+ user=doris_sandbox.readonly_user,
+ password=doris_sandbox.readonly_password,
+ )
+ async with _transport_client(transport, readonly_environment) as client:
+ allowed_result, allowed_payload = await _exec_query(
+ client,
+ f"SELECT marker FROM {doris_sandbox.qualified_table} WHERE id = 1",
+ )
+ assert allowed_result.is_error is False
+ assert allowed_payload["data"] == [{"marker": doris_sandbox.marker}]
+
+ denied_result, denied_payload = await _exec_query(
+ client,
+ f"""
+ INSERT INTO {doris_sandbox.qualified_table}
+ VALUES (2, 'must-not-write')
+ """,
+ )
+ assert denied_result.is_error is True
+ assert denied_payload["success"] is False
+ assert denied_payload["error_type"] == "permission_denied"
diff --git a/test/utils/test_doris_user_pool_manager.py
b/test/utils/test_doris_user_pool_manager.py
index f046804..9d64b42 100644
--- a/test/utils/test_doris_user_pool_manager.py
+++ b/test/utils/test_doris_user_pool_manager.py
@@ -90,6 +90,18 @@ def manager():
return DorisConnectionManager(manager_config())
[email protected]
+async def test_pool_warmup_does_not_exceed_configured_capacity(manager):
+ pool = FakePool("global")
+ manager.pool = pool
+ manager.maxsize = 1
+
+ await manager._warmup_pool()
+
+ assert pool.acquire_calls == 1
+ assert len(pool.release_calls) == 1
+
+
def doris_context(user="alice", token=""):
return AuthContext(
user_id=user,
diff --git a/test/utils/test_query_error_classification.py
b/test/utils/test_query_error_classification.py
new file mode 100644
index 0000000..a715c7a
--- /dev/null
+++ b/test/utils/test_query_error_classification.py
@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+# 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.
+
+from doris_mcp_server.utils.query_executor import DorisQueryExecutor
+
+
+def test_doris_command_denied_is_classified_as_permission_error() -> None:
+ executor = DorisQueryExecutor(object())
+
+ error = executor._analyze_error(
+ "LOAD command denied to user 'readonly'@'127.0.0.1' for table 'db.t'"
+ )
+
+ assert error["error_type"] == "permission_denied"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
