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 3151ed7 feat: paginate MCP list operations (#141)
3151ed7 is described below
commit 3151ed792eb62261c249e56fe4efe46885e34c40
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 10:00:46 2026 +0800
feat: paginate MCP list operations (#141)
---
.env.example | 3 +
CHANGELOG.md | 2 +
README.md | 19 ++
doris_mcp_server/main.py | 2 +
doris_mcp_server/multiworker_app.py | 1 +
doris_mcp_server/pagination.py | 202 ++++++++++++++++++
doris_mcp_server/protocol.py | 98 ++++++++-
doris_mcp_server/tools/resources_manager.py | 66 +++++-
doris_mcp_server/utils/config.py | 12 ++
test/integration/test_real_doris_transports.py | 101 ++++++++-
test/protocol/pagination_fixture.py | 138 +++++++++++++
test/protocol/pagination_stdio_server.py | 39 ++++
test/protocol/test_mcp_v2_protocol.py | 32 +--
test/protocol/test_multiworker_config.py | 18 ++
test/protocol/test_protocol_pagination.py | 274 +++++++++++++++++++++++++
test/tools/test_resources_manager_cache.py | 43 +++-
16 files changed, 1008 insertions(+), 42 deletions(-)
diff --git a/.env.example b/.env.example
index 1cdaba6..6c11165 100644
--- a/.env.example
+++ b/.env.example
@@ -458,6 +458,9 @@ SERVER_PORT=3000
# /mcp/legacy. Stdio compatibility is unaffected by this switch.
ENABLE_LEGACY_HTTP_ADAPTER=false
+# Maximum resources, tools, or prompts returned in one list page (1-1000).
+MCP_LIST_PAGE_SIZE=100
+
# Temporary files directory
TEMP_FILES_DIR=tmp
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 667fc95..e363112 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,6 +35,8 @@ under **Unreleased** until a new version is selected and
published.
client/redirect host context.
- Per-request client capability enforcement, cache hints, typed protocol
errors, and deterministic product identity.
+- Permission-bound cursor pagination for resources, tools, and prompts on
+ Streamable HTTP and stdio.
- Real Doris process tests covering Streamable HTTP and stdio.
### Changed
diff --git a/README.md b/README.md
index 6186b8f..fc1a68a 100644
--- a/README.md
+++ b/README.md
@@ -208,6 +208,9 @@ export DORIS_PASSWORD="your_password"
# identified 2025-11-25 client still needs /mcp/legacy.
export ENABLE_LEGACY_HTTP_ADAPTER=false
+# Bound each resources/list, tools/list, and prompts/list response.
+export MCP_LIST_PAGE_SIZE=100
+
# Token Management Interface (Security-Critical)
export TOKEN_ADMIN="$(python -c 'import secrets;
print(secrets.token_urlsafe(32))')"
export TOKEN_MANAGEMENT_ADMIN_TOKEN="$(python -c 'import secrets;
print(secrets.token_urlsafe(32))')"
@@ -283,6 +286,8 @@ cp .env.example .env
* `ENABLE_LEGACY_HTTP_ADAPTER`: Expose the isolated
`2025-11-25` migration adapter at `/mcp/legacy` (default: false);
modern traffic always uses `POST /mcp`
+ * `MCP_LIST_PAGE_SIZE`: Maximum resources, tools, or prompts returned
+ per protocol page (default: 100; range: 1-1000)
* **Authentication Configuration (Enhanced in v0.6.0)**:
* `ENABLE_TOKEN_AUTH`: Enable token-based authentication (default: false)
* `ENABLE_JWT_AUTH`: Enable JWT authentication (default: false)
@@ -541,6 +546,20 @@ Stdio carries the same JSON-RPC request metadata in the
message body, but it
does not use HTTP headers. Do not write logs or other diagnostics to stdout in
stdio mode; stdout is reserved for MCP protocol messages.
+### List Pagination
+
+`resources/list`, `tools/list`, and `prompts/list` return at most
+`MCP_LIST_PAGE_SIZE` entries per response on both Streamable HTTP and stdio.
+Results are ordered by their stable resource URI, tool name, or prompt name.
+When `nextCursor` is present, pass that opaque value as the next request's
+`cursor`; do not parse or construct cursors.
+
+A cursor is bound to its list type, the current visible-list snapshot, and the
+current authorization context. Reusing it for another list, after visibility
+changes, or under another principal returns `Invalid Params`. Restart the
+listing without a cursor in that case. This prevents a multi-page traversal
+from silently duplicating, dropping, or crossing permission-scoped entries.
+
### Migrating from MCP 2025-11-25
1. Upgrade the client to a `2026-07-28`-capable MCP SDK.
diff --git a/doris_mcp_server/main.py b/doris_mcp_server/main.py
index 095311f..f70d62b 100644
--- a/doris_mcp_server/main.py
+++ b/doris_mcp_server/main.py
@@ -83,6 +83,7 @@ def _multiworker_environment(
"ENABLE_LEGACY_HTTP_ADAPTER": str(
config.enable_legacy_http_adapter
).lower(),
+ "MCP_LIST_PAGE_SIZE": str(config.mcp_list_page_size),
"SERVER_NAME": config.server_name,
"TRANSPORT": "http",
"WORKERS": str(workers),
@@ -134,6 +135,7 @@ class DorisServer:
name=config.server_name,
version=config.server_version,
logger=self.logger,
+ list_page_size=config.mcp_list_page_size,
)
async def start_stdio(self) -> None:
diff --git a/doris_mcp_server/multiworker_app.py
b/doris_mcp_server/multiworker_app.py
index 0d64b4b..38dc6e7 100644
--- a/doris_mcp_server/multiworker_app.py
+++ b/doris_mcp_server/multiworker_app.py
@@ -163,6 +163,7 @@ async def initialize_worker() -> None:
name=config.server_name,
version=config.server_version,
logger=logger,
+ list_page_size=config.mcp_list_page_size,
)
# Create the exact modern/legacy HTTP boundary for this worker.
diff --git a/doris_mcp_server/pagination.py b/doris_mcp_server/pagination.py
new file mode 100644
index 0000000..11c4abc
--- /dev/null
+++ b/doris_mcp_server/pagination.py
@@ -0,0 +1,202 @@
+# 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.
+"""Stateless, permission-bound pagination for MCP list operations."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import json
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from hashlib import sha256
+from hmac import compare_digest
+from typing import Any, Generic, Literal, TypeVar
+
+from .utils.security import AuthContext
+
+DEFAULT_LIST_PAGE_SIZE = 100
+MAX_LIST_PAGE_SIZE = 1000
+_CURSOR_VERSION = 1
+_MAX_CURSOR_LENGTH = 2048
+_CURSOR_FIELDS = {
+ "version",
+ "collection",
+ "after",
+ "snapshot",
+ "authorization",
+}
+
+CursorCollection = Literal["resources", "tools", "prompts"]
+T = TypeVar("T")
+
+
+class PaginationCursorError(ValueError):
+ """An invalid cursor supplied by an MCP client."""
+
+ def __init__(self, reason: str) -> None:
+ super().__init__(reason)
+ self.reason = reason
+
+
+@dataclass(frozen=True)
+class PaginationPage(Generic[T]):
+ """One deterministic page and the opaque cursor for its successor."""
+
+ items: list[T]
+ next_cursor: str | None
+
+
+def _canonical_json(value: Any) -> bytes:
+ return json.dumps(
+ value,
+ ensure_ascii=False,
+ separators=(",", ":"),
+ sort_keys=True,
+ ).encode("utf-8")
+
+
+def _digest(value: Any) -> str:
+ return sha256(_canonical_json(value)).hexdigest()
+
+
+def _authorization_fingerprint(auth_context: AuthContext | None) -> str:
+ if auth_context is None:
+ return _digest({"authMethod": "anonymous", "principal": "anonymous"})
+
+ security_level = getattr(auth_context.security_level, "value", "")
+ return _digest(
+ {
+ "authMethod": auth_context.auth_method,
+ "tokenId": auth_context.token_id,
+ "userId": auth_context.user_id,
+ "roles": sorted(auth_context.roles),
+ "permissions": sorted(auth_context.permissions),
+ "securityLevel": security_level,
+ "dorisUser": auth_context.doris_user,
+ "oauthClientId": auth_context.oauth_client_id,
+ "oauthScopes": sorted(auth_context.oauth_scopes),
+ "oauthTokenId": auth_context.oauth_token_id,
+ "oauthIssuer": auth_context.oauth_issuer,
+ "oauthResource": auth_context.oauth_resource,
+ "oauthAudiences": sorted(auth_context.oauth_audiences),
+ "poolKey": auth_context.pool_key,
+ "dorisOAuthDatabaseTools": {
+ "enabled": auth_context.doris_oauth_db_tools_enabled,
+ "allowlist":
sorted(auth_context.doris_oauth_db_tool_allowlist),
+ },
+ "dorisOAuthQueryTools": {
+ "enabled": auth_context.doris_oauth_query_tools_enabled,
+ "allowlist":
sorted(auth_context.doris_oauth_query_tool_allowlist),
+ },
+ "dorisOAuthExplainTools": {
+ "enabled": auth_context.doris_oauth_explain_tools_enabled,
+ "allowlist":
sorted(auth_context.doris_oauth_explain_tool_allowlist),
+ },
+ }
+ )
+
+
+def _encode_cursor(payload: dict[str, Any]) -> str:
+ return (
+
base64.urlsafe_b64encode(_canonical_json(payload)).decode("ascii").rstrip("=")
+ )
+
+
+def _decode_cursor(cursor: str) -> dict[str, Any]:
+ if not cursor or len(cursor) > _MAX_CURSOR_LENGTH:
+ raise PaginationCursorError("invalid")
+ try:
+ padding = "=" * (-len(cursor) % 4)
+ raw = base64.b64decode(
+ (cursor + padding).encode("ascii"),
+ altchars=b"-_",
+ validate=True,
+ )
+ payload = json.loads(raw.decode("utf-8"))
+ except (
+ UnicodeEncodeError,
+ UnicodeDecodeError,
+ binascii.Error,
+ json.JSONDecodeError,
+ ) as exc:
+ raise PaginationCursorError("invalid") from exc
+
+ if not isinstance(payload, dict) or set(payload) != _CURSOR_FIELDS:
+ raise PaginationCursorError("invalid")
+ if payload.get("version") != _CURSOR_VERSION:
+ raise PaginationCursorError("invalid")
+ if not all(
+ isinstance(payload.get(field), str)
+ for field in ("collection", "after", "snapshot", "authorization")
+ ):
+ raise PaginationCursorError("invalid")
+ return payload
+
+
+def paginate(
+ items: Sequence[T],
+ *,
+ collection: CursorCollection,
+ cursor: str | None,
+ page_size: int,
+ identifier: Callable[[T], str],
+ auth_context: AuthContext | None,
+) -> PaginationPage[T]:
+ """Return a deterministic keyset page or reject an unsafe continuation."""
+ if not 1 <= page_size <= MAX_LIST_PAGE_SIZE:
+ raise ValueError(f"page_size must be in the range
1-{MAX_LIST_PAGE_SIZE}")
+
+ ordered = sorted(items, key=identifier)
+ identifiers = [identifier(item) for item in ordered]
+ if len(identifiers) != len(set(identifiers)):
+ raise RuntimeError(f"{collection} list contains duplicate identifiers")
+
+ snapshot = _digest(identifiers)
+ authorization = _authorization_fingerprint(auth_context)
+ start = 0
+
+ if cursor is not None:
+ payload = _decode_cursor(cursor)
+ if payload["collection"] != collection:
+ raise PaginationCursorError("wrong_collection")
+ if not compare_digest(payload["authorization"], authorization):
+ raise PaginationCursorError("authorization_context_changed")
+ if not compare_digest(payload["snapshot"], snapshot):
+ raise PaginationCursorError("stale")
+ try:
+ start = identifiers.index(payload["after"]) + 1
+ except ValueError as exc:
+ raise PaginationCursorError("stale") from exc
+ if start >= len(ordered):
+ raise PaginationCursorError("invalid")
+
+ page_items = ordered[start : start + page_size]
+ end = start + len(page_items)
+ next_cursor = None
+ if page_items and end < len(ordered):
+ next_cursor = _encode_cursor(
+ {
+ "version": _CURSOR_VERSION,
+ "collection": collection,
+ "after": identifiers[end - 1],
+ "snapshot": snapshot,
+ "authorization": authorization,
+ }
+ )
+
+ return PaginationPage(items=page_items, next_cursor=next_cursor)
diff --git a/doris_mcp_server/protocol.py b/doris_mcp_server/protocol.py
index 0712879..082778d 100644
--- a/doris_mcp_server/protocol.py
+++ b/doris_mcp_server/protocol.py
@@ -20,8 +20,8 @@ from __future__ import annotations
import json
import logging
-from collections.abc import Iterable, Mapping
-from typing import Any, Protocol
+from collections.abc import Callable, Iterable, Mapping, Sequence
+from typing import Any, Protocol, TypeVar
from mcp.server import Server, ServerRequestContext
from mcp.server.caching import CacheHint
@@ -53,6 +53,14 @@ from mcp.types import (
)
from .auth.operation_policy import OperationAuthorizationError,
authorize_operation
+from .pagination import (
+ DEFAULT_LIST_PAGE_SIZE,
+ MAX_LIST_PAGE_SIZE,
+ CursorCollection,
+ PaginationCursorError,
+ PaginationPage,
+ paginate,
+)
from .utils.redaction import (
redact_error_payload,
redact_sensitive_text,
@@ -60,6 +68,8 @@ from .utils.redaction import (
)
from .utils.security import get_current_auth_context
+_ListItemT = TypeVar("_ListItemT")
+
class ResourcesManager(Protocol):
async def list_resources(self) -> list[Resource]: ...
@@ -186,6 +196,31 @@ def _decode_resource_request_error(payload: str) ->
tuple[str, str] | None:
return error_code, message
+def _paginate_list_or_raise(
+ items: Sequence[_ListItemT],
+ *,
+ collection: CursorCollection,
+ cursor: str | None,
+ page_size: int,
+ identifier: Callable[[_ListItemT], str],
+) -> PaginationPage[_ListItemT]:
+ try:
+ return paginate(
+ items,
+ collection=collection,
+ cursor=cursor,
+ page_size=page_size,
+ identifier=identifier,
+ auth_context=get_current_auth_context(),
+ )
+ except PaginationCursorError as exc:
+ raise MCPError(
+ code=INVALID_PARAMS,
+ message="Invalid pagination cursor",
+ data={"cursorError": exc.reason},
+ ) from exc
+
+
def create_doris_mcp_server(
*,
resources_manager: ResourcesManager,
@@ -194,20 +229,37 @@ def create_doris_mcp_server(
name: str,
version: str,
logger: logging.Logger,
+ list_page_size: int = DEFAULT_LIST_PAGE_SIZE,
required_client_capabilities: Mapping[str, ClientCapabilities] | None =
None,
required_tool_capabilities: Mapping[str, ClientCapabilities] | None = None,
) -> Server:
"""Create the one low-level SDK v2 server used by every transport."""
+ if not 1 <= list_page_size <= MAX_LIST_PAGE_SIZE:
+ raise ValueError(f"list_page_size must be in the range
1-{MAX_LIST_PAGE_SIZE}")
async def list_resources(
ctx: ServerRequestContext,
params: PaginatedRequestParams | None,
) -> ListResourcesResult:
- del ctx, params
+ del ctx
authorize_operation(get_current_auth_context(), "list_resources")
resources = await resources_manager.list_resources()
- logger.info("Returning %d resources", len(resources))
- return ListResourcesResult(resources=resources)
+ page = _paginate_list_or_raise(
+ resources,
+ collection="resources",
+ cursor=params.cursor if params else None,
+ page_size=list_page_size,
+ identifier=lambda resource: str(resource.uri),
+ )
+ logger.info(
+ "Returning %d of %d resources",
+ len(page.items),
+ len(resources),
+ )
+ return ListResourcesResult(
+ resources=page.items,
+ next_cursor=page.next_cursor,
+ )
async def read_resource(
ctx: ServerRequestContext,
@@ -242,11 +294,21 @@ def create_doris_mcp_server(
ctx: ServerRequestContext,
params: PaginatedRequestParams | None,
) -> ListToolsResult:
- del ctx, params
+ del ctx
authorize_operation(get_current_auth_context(), "list_tools")
tools = await tools_manager.list_tools()
- logger.info("Returning %d tools", len(tools))
- return ListToolsResult(tools=tools)
+ page = _paginate_list_or_raise(
+ tools,
+ collection="tools",
+ cursor=params.cursor if params else None,
+ page_size=list_page_size,
+ identifier=lambda tool: tool.name,
+ )
+ logger.info("Returning %d of %d tools", len(page.items), len(tools))
+ return ListToolsResult(
+ tools=page.items,
+ next_cursor=page.next_cursor,
+ )
async def call_tool(
ctx: ServerRequestContext,
@@ -279,11 +341,25 @@ def create_doris_mcp_server(
ctx: ServerRequestContext,
params: PaginatedRequestParams | None,
) -> ListPromptsResult:
- del ctx, params
+ del ctx
authorize_operation(get_current_auth_context(), "list_prompts")
prompts = await prompts_manager.list_prompts()
- logger.info("Returning %d prompts", len(prompts))
- return ListPromptsResult(prompts=prompts)
+ page = _paginate_list_or_raise(
+ prompts,
+ collection="prompts",
+ cursor=params.cursor if params else None,
+ page_size=list_page_size,
+ identifier=lambda prompt: prompt.name,
+ )
+ logger.info(
+ "Returning %d of %d prompts",
+ len(page.items),
+ len(prompts),
+ )
+ return ListPromptsResult(
+ prompts=page.items,
+ next_cursor=page.next_cursor,
+ )
async def get_prompt(
ctx: ServerRequestContext,
diff --git a/doris_mcp_server/tools/resources_manager.py
b/doris_mcp_server/tools/resources_manager.py
index a7f231e..b6de177 100644
--- a/doris_mcp_server/tools/resources_manager.py
+++ b/doris_mcp_server/tools/resources_manager.py
@@ -23,6 +23,7 @@ import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import datetime
+from hashlib import sha256
from typing import Any, cast
from urllib.parse import quote, unquote
@@ -75,10 +76,16 @@ class ViewMetadata:
class MetadataCache:
"""Metadata cache manager"""
- def __init__(self, ttl_seconds: int = 300, enabled: bool = False):
+ def __init__(
+ self,
+ ttl_seconds: int = 300,
+ enabled: bool = False,
+ max_entries: int = 128,
+ ):
self.cache: dict[str, tuple[Any, float]] = {}
self.ttl = ttl_seconds
self.enabled = enabled
+ self.max_entries = max_entries
async def get(self, key: str) -> Any | None:
if not self.enabled:
@@ -94,6 +101,9 @@ class MetadataCache:
async def set(self, key: str, value: Any) -> None:
if not self.enabled:
return
+ if key not in self.cache and len(self.cache) >= self.max_entries:
+ oldest_key = min(self.cache, key=lambda item: self.cache[item][1])
+ del self.cache[oldest_key]
self.cache[key] = (value, datetime.now().timestamp())
@@ -132,10 +142,52 @@ class DorisResourcesManager:
def __init__(self, connection_manager: DorisConnectionManager):
self.connection_manager = connection_manager
- # Resource metadata cache is disabled until it is identity-aware.
- # Static token-bound DB and Doris OAuth can route to different Doris
- # users; global cache keys would leak metadata across identities.
- self.metadata_cache = MetadataCache(enabled=False)
+ self.metadata_cache = MetadataCache(enabled=True)
+
+ def _metadata_cache_scope(self) -> str:
+ auth_context = get_auth_context()
+ original_config = getattr(
+ self.connection_manager,
+ "original_db_config",
+ {},
+ )
+ endpoint = {
+ field: original_config.get(field)
+ for field in ("host", "port", "user", "database")
+ }
+
+ token_manager = getattr(self.connection_manager, "token_manager", None)
+ token = getattr(auth_context, "token", "") if auth_context else ""
+ if token_manager and token:
+ token_config = token_manager.get_database_config_by_token(token)
+ if token_config:
+ endpoint = {
+ field: getattr(token_config, field, None)
+ for field in ("host", "port", "user", "database")
+ }
+
+ identity = {
+ "authMethod": getattr(auth_context, "auth_method", ""),
+ "tokenId": getattr(auth_context, "token_id", ""),
+ "userId": getattr(auth_context, "user_id", ""),
+ "roles": sorted(getattr(auth_context, "roles", [])),
+ "permissions": sorted(getattr(auth_context, "permissions", [])),
+ "dorisUser": getattr(auth_context, "doris_user", ""),
+ "oauthClientId": getattr(auth_context, "oauth_client_id", ""),
+ "oauthScopes": sorted(getattr(auth_context, "oauth_scopes", [])),
+ "oauthResource": getattr(auth_context, "oauth_resource", ""),
+ "oauthAudiences": sorted(
+ getattr(auth_context, "oauth_audiences", [])
+ ),
+ "poolKey": getattr(auth_context, "pool_key", ""),
+ }
+ payload = json.dumps(
+ {"endpoint": endpoint, "identity": identity},
+ ensure_ascii=False,
+ separators=(",", ":"),
+ sort_keys=True,
+ )
+ return sha256(payload.encode("utf-8")).hexdigest()
def _is_doris_oauth_context(self) -> bool:
return getattr(get_auth_context(), "auth_method", "") == "doris_oauth"
@@ -364,7 +416,7 @@ class DorisResourcesManager:
async def _get_table_metadata(self) -> list[TableMetadata]:
"""Get metadata for all tables"""
- cache_key = "table_metadata"
+ cache_key = f"{self._metadata_cache_scope()}:table_metadata"
cached = await self.metadata_cache.get(cache_key)
if cached:
return cast(list[TableMetadata], cached)
@@ -467,7 +519,7 @@ class DorisResourcesManager:
async def _get_view_metadata(self) -> list[ViewMetadata]:
"""Get metadata for all views"""
- cache_key = "view_metadata"
+ cache_key = f"{self._metadata_cache_scope()}:view_metadata"
cached = await self.metadata_cache.get(cache_key)
if cached:
return cast(list[ViewMetadata], cached)
diff --git a/doris_mcp_server/utils/config.py b/doris_mcp_server/utils/config.py
index 64d676c..00dd3dc 100644
--- a/doris_mcp_server/utils/config.py
+++ b/doris_mcp_server/utils/config.py
@@ -603,6 +603,7 @@ class DorisConfig:
mcp_allowed_hosts: list[str] = field(default_factory=list)
mcp_allowed_origins: list[str] = field(default_factory=list)
enable_legacy_http_adapter: bool = False
+ mcp_list_page_size: int = 100
transport: str = "stdio"
workers: int = 1
@@ -1148,6 +1149,12 @@ class DorisConfig:
os.getenv("ENABLE_LEGACY_HTTP_ADAPTER", "false").lower() ==
"true"
)
_mark_source(config, "enable_legacy_http_adapter", "env")
+ if "MCP_LIST_PAGE_SIZE" in os.environ:
+ config.mcp_list_page_size = _env_int(
+ "MCP_LIST_PAGE_SIZE",
+ config.mcp_list_page_size,
+ )
+ _mark_source(config, "mcp_list_page_size", "env")
config.temp_files_dir = os.getenv("TEMP_FILES_DIR",
config.temp_files_dir)
return config
@@ -1165,6 +1172,7 @@ class DorisConfig:
"mcp_allowed_hosts",
"mcp_allowed_origins",
"enable_legacy_http_adapter",
+ "mcp_list_page_size",
"temp_files_dir",
"transport",
"workers",
@@ -1239,6 +1247,7 @@ class DorisConfig:
"mcp_allowed_hosts": self.mcp_allowed_hosts,
"mcp_allowed_origins": self.mcp_allowed_origins,
"enable_legacy_http_adapter": self.enable_legacy_http_adapter,
+ "mcp_list_page_size": self.mcp_list_page_size,
"temp_files_dir": self.temp_files_dir,
"database": {
"host": self.database.host,
@@ -1433,6 +1442,9 @@ class DorisConfig:
if self.database.http_max_response_bytes <= 0:
errors.append("Doris HTTP response byte limit must be greater than
0")
+ if not 1 <= self.mcp_list_page_size <= 1000:
+ errors.append("MCP list page size must be in the range 1-1000")
+
# Validate security configuration
if self.security.auth_type not in ["token", "basic", "oauth", "jwt"]:
errors.append("Authentication type must be one of token, basic,
oauth, or jwt")
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index 73a5b0a..0a76f1c 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -288,17 +288,25 @@ async def _http_process(environment: dict[str, str]) ->
AsyncIterator[str]:
@asynccontextmanager
-async def _http_client(environment: dict[str, str]) -> AsyncIterator[Client]:
+async def _http_client(
+ environment: dict[str, str],
+ *,
+ read_timeout_seconds: int = 15,
+) -> AsyncIterator[Client]:
async with _http_process(environment) as base_url:
async with Client(
f"{base_url}/mcp",
- read_timeout_seconds=15,
+ read_timeout_seconds=read_timeout_seconds,
) as client:
yield client
@asynccontextmanager
-async def _stdio_client(environment: dict[str, str]) -> AsyncIterator[Client]:
+async def _stdio_client(
+ environment: dict[str, str],
+ *,
+ read_timeout_seconds: int = 15,
+) -> AsyncIterator[Client]:
parameters = StdioServerParameters(
command=sys.executable,
args=["-m", "doris_mcp_server", "--transport", "stdio"],
@@ -311,7 +319,7 @@ async def _stdio_client(environment: dict[str, str]) ->
AsyncIterator[Client]:
client = await stack.enter_async_context(
Client(
stdio_client(parameters, errlog=log_file),
- read_timeout_seconds=15,
+ read_timeout_seconds=read_timeout_seconds,
)
)
except Exception:
@@ -329,10 +337,18 @@ async def _stdio_client(environment: dict[str, str]) ->
AsyncIterator[Client]:
def _transport_client(
transport: str,
environment: dict[str, str],
+ *,
+ read_timeout_seconds: int = 15,
) -> Any:
if transport == "http":
- return _http_client(environment)
- return _stdio_client(environment)
+ return _http_client(
+ environment,
+ read_timeout_seconds=read_timeout_seconds,
+ )
+ return _stdio_client(
+ environment,
+ read_timeout_seconds=read_timeout_seconds,
+ )
async def _exec_query(
@@ -353,6 +369,36 @@ async def _exec_query(
return result, result.structured_content
+async def _collect_list_pages(
+ list_method: Any,
+ *,
+ result_field: str,
+ identifier: Any,
+ max_page_size: int,
+) -> tuple[list[str], int]:
+ cursor = None
+ seen_cursors: set[str] = set()
+ identifiers: list[str] = []
+ page_count = 0
+
+ while True:
+ result = await list_method(cursor=cursor, cache_mode="bypass")
+ page = getattr(result, result_field)
+ assert 0 < len(page) <= max_page_size
+ identifiers.extend(identifier(item) for item in page)
+ page_count += 1
+
+ cursor = result.next_cursor
+ if cursor is None:
+ break
+ assert cursor not in seen_cursors
+ seen_cursors.add(cursor)
+
+ assert identifiers == sorted(identifiers)
+ assert len(identifiers) == len(set(identifiers))
+ return identifiers, page_count
+
+
async def test_real_doris_liveness_and_readiness(
doris_sandbox: DorisSandbox,
) -> None:
@@ -379,6 +425,49 @@ async def test_real_doris_liveness_and_readiness(
}
[email protected]("transport", ["http", "stdio"])
+async def test_real_doris_protocol_lists_paginate_without_loss(
+ transport: str,
+ doris_sandbox: DorisSandbox,
+) -> None:
+ environment = _server_environment(
+ doris_sandbox.settings,
+ user=doris_sandbox.settings.user,
+ password=doris_sandbox.settings.password,
+ )
+ page_size = 100
+ environment["MCP_LIST_PAGE_SIZE"] = str(page_size)
+ results: dict[str, tuple[list[str], int]] = {}
+
+ async with _transport_client(
+ transport,
+ environment,
+ read_timeout_seconds=60,
+ ) as client:
+ for list_method, result_field, identifier in (
+ (
+ client.list_resources,
+ "resources",
+ lambda resource: str(resource.uri),
+ ),
+ (client.list_tools, "tools", lambda tool: tool.name),
+ (client.list_prompts, "prompts", lambda prompt: prompt.name),
+ ):
+ identifiers, page_count = await _collect_list_pages(
+ list_method,
+ result_field=result_field,
+ identifier=identifier,
+ max_page_size=page_size,
+ )
+ results[result_field] = (identifiers, page_count)
+
+ resource_identifiers, resource_page_count = results["resources"]
+ assert len(resource_identifiers) > page_size
+ assert resource_page_count > 1
+ assert results["tools"][0]
+ assert results["prompts"][0]
+
+
@pytest.mark.parametrize("transport", ["http", "stdio"])
async def test_real_doris_read_write_permission_timeout_and_recovery(
transport: str,
diff --git a/test/protocol/pagination_fixture.py
b/test/protocol/pagination_fixture.py
new file mode 100644
index 0000000..3aca503
--- /dev/null
+++ b/test/protocol/pagination_fixture.py
@@ -0,0 +1,138 @@
+# 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.
+"""Static managers used to verify MCP list pagination across transports."""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from mcp.types import (
+ GetPromptResult,
+ Prompt,
+ PromptMessage,
+ Resource,
+ TextContent,
+ Tool,
+)
+
+from doris_mcp_server import __version__
+from doris_mcp_server.protocol import create_doris_mcp_server
+
+RESOURCE_URIS = [
+ "doris://table/alpha",
+ "doris://table/bravo",
+ "doris://table/charlie",
+ "doris://table/delta",
+ "doris://table/echo",
+]
+TOOL_NAMES = ["alpha", "bravo", "charlie", "delta", "echo"]
+PROMPT_NAMES = ["alpha", "bravo", "charlie", "delta", "echo"]
+
+
+@dataclass
+class PaginationManagers:
+ resources: PaginationResourcesManager
+ tools: PaginationToolsManager
+ prompts: PaginationPromptsManager
+
+
+class PaginationResourcesManager:
+ def __init__(self) -> None:
+ self.resources = [
+ Resource(
+ uri=uri,
+ name=uri.rsplit("/", 1)[-1],
+ mime_type="application/json",
+ )
+ for uri in reversed(RESOURCE_URIS)
+ ]
+
+ async def list_resources(self) -> list[Resource]:
+ return list(self.resources)
+
+ async def read_resource(self, uri: str) -> str:
+ return json.dumps({"uri": uri})
+
+
+class PaginationToolsManager:
+ def __init__(self) -> None:
+ self.tools = [
+ Tool(
+ name=name,
+ description=f"{name} test tool",
+ input_schema={"type": "object", "properties": {}},
+ )
+ for name in reversed(TOOL_NAMES)
+ ]
+
+ async def list_tools(self) -> list[Tool]:
+ return list(self.tools)
+
+ async def call_tool(self, name: str, arguments: dict[str, Any]) -> str:
+ return json.dumps({"name": name, "arguments": arguments})
+
+
+class PaginationPromptsManager:
+ def __init__(self) -> None:
+ self.prompts = [
+ Prompt(name=name, description=f"{name} test prompt")
+ for name in reversed(PROMPT_NAMES)
+ ]
+
+ async def list_prompts(self) -> list[Prompt]:
+ return list(self.prompts)
+
+ async def get_prompt(
+ self,
+ name: str,
+ arguments: dict[str, Any],
+ ) -> GetPromptResult:
+ return GetPromptResult(
+ description=name,
+ messages=[
+ PromptMessage(
+ role="user",
+ content=TextContent(
+ type="text",
+ text=json.dumps(arguments, sort_keys=True),
+ ),
+ )
+ ],
+ )
+
+
+def create_pagination_server(*, page_size: int = 2):
+ resources = PaginationResourcesManager()
+ tools = PaginationToolsManager()
+ prompts = PaginationPromptsManager()
+ server = create_doris_mcp_server(
+ resources_manager=resources,
+ tools_manager=tools,
+ prompts_manager=prompts,
+ name="doris-mcp-pagination-test",
+ version=__version__,
+ logger=logging.getLogger(__name__),
+ list_page_size=page_size,
+ )
+ return server, PaginationManagers(
+ resources=resources,
+ tools=tools,
+ prompts=prompts,
+ )
diff --git a/test/protocol/pagination_stdio_server.py
b/test/protocol/pagination_stdio_server.py
new file mode 100644
index 0000000..142a873
--- /dev/null
+++ b/test/protocol/pagination_stdio_server.py
@@ -0,0 +1,39 @@
+# 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.
+"""True subprocess stdio fixture for MCP list pagination."""
+
+from __future__ import annotations
+
+import asyncio
+
+from mcp.server.stdio import stdio_server
+
+from test.protocol.pagination_fixture import create_pagination_server
+
+
+async def main() -> None:
+ server, _ = create_pagination_server(page_size=2)
+ async with stdio_server() as (read_stream, write_stream):
+ await server.run(
+ read_stream,
+ write_stream,
+ server.create_initialization_options(),
+ )
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/test/protocol/test_mcp_v2_protocol.py
b/test/protocol/test_mcp_v2_protocol.py
index c0248da..30ee3f3 100644
--- a/test/protocol/test_mcp_v2_protocol.py
+++ b/test/protocol/test_mcp_v2_protocol.py
@@ -48,6 +48,16 @@ from doris_mcp_server.protocol import (
from test.protocol.stdio_capability_server import OneToolManager as
ProfileToolManager
REQUIRED_EXTENSION = "io.apache.doris/read"
+PROFILE_TOOL_NAMES = sorted(
+ [
+ "echo",
+ "get_sql_profile",
+ "monitor_data_freshness",
+ "analyze_data_access_patterns",
+ "analyze_columns",
+ "get_monitoring_metrics",
+ ]
+)
def test_transport_security_accepts_explicit_deployment_allowlists():
@@ -996,14 +1006,9 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
stdio_client(server_params),
extensions=[advertise(REQUIRED_EXTENSION)],
) as capable:
- assert [tool.name for tool in (await capable.list_tools()).tools] == [
- "echo",
- "get_sql_profile",
- "monitor_data_freshness",
- "analyze_data_access_patterns",
- "analyze_columns",
- "get_monitoring_metrics",
- ]
+ assert [
+ tool.name for tool in (await capable.list_tools()).tools
+ ] == PROFILE_TOOL_NAMES
secret = "stdio-secret-sec-016"
error_result = await capable.call_tool(
"echo",
@@ -1024,14 +1029,9 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
assert error_result.structured_content["token"] == "[REDACTED]"
async with Client(stdio_client(server_params), mode="legacy") as legacy:
- assert [tool.name for tool in (await legacy.list_tools()).tools] == [
- "echo",
- "get_sql_profile",
- "monitor_data_freshness",
- "analyze_data_access_patterns",
- "analyze_columns",
- "get_monitoring_metrics",
- ]
+ assert [
+ tool.name for tool in (await legacy.list_tools()).tools
+ ] == PROFILE_TOOL_NAMES
legacy_error = await legacy.read_resource("doris://table/missing")
assert (
json.loads(legacy_error.contents[0].text)["error_code"]
diff --git a/test/protocol/test_multiworker_config.py
b/test/protocol/test_multiworker_config.py
index 495f55a..9c4d6ca 100644
--- a/test/protocol/test_multiworker_config.py
+++ b/test/protocol/test_multiworker_config.py
@@ -27,6 +27,22 @@ def
test_legacy_http_adapter_is_default_off_and_requires_explicit_env(monkeypatc
assert enabled.to_dict()["enable_legacy_http_adapter"] is True
+def test_mcp_list_page_size_is_configurable_and_bounded(monkeypatch):
+ monkeypatch.delenv("MCP_LIST_PAGE_SIZE", raising=False)
+ assert DorisConfig.from_env().mcp_list_page_size == 100
+
+ monkeypatch.setenv("MCP_LIST_PAGE_SIZE", "17")
+ configured = DorisConfig.from_env()
+ assert configured.mcp_list_page_size == 17
+ assert configured.to_dict()["mcp_list_page_size"] == 17
+ assert configured.validate() == []
+
+ configured.mcp_list_page_size = 0
+ assert "MCP list page size must be in the range 1-1000" in
configured.validate()
+ configured.mcp_list_page_size = 1001
+ assert "MCP list page size must be in the range 1-1000" in
configured.validate()
+
+
def test_multiworker_environment_preserves_resolved_parent_config(monkeypatch):
config = DorisConfig()
config.database.host = "127.0.0.1"
@@ -38,6 +54,7 @@ def
test_multiworker_environment_preserves_resolved_parent_config(monkeypatch):
config.mcp_allowed_hosts = ["mcp.example.test", "mcp.example.test:*"]
config.mcp_allowed_origins = ["https://client.example.test";]
config.enable_legacy_http_adapter = True
+ config.mcp_list_page_size = 17
worker_env = _multiworker_environment(
config,
@@ -64,6 +81,7 @@ def
test_multiworker_environment_preserves_resolved_parent_config(monkeypatch):
]
assert child_config.mcp_allowed_origins == ["https://client.example.test";]
assert child_config.enable_legacy_http_adapter is True
+ assert child_config.mcp_list_page_size == 17
assert child_config.server_name == "doris-mcp-server"
assert child_config.server_version == __version__
assert child_config.transport == "http"
diff --git a/test/protocol/test_protocol_pagination.py
b/test/protocol/test_protocol_pagination.py
new file mode 100644
index 0000000..f0ef5d2
--- /dev/null
+++ b/test/protocol/test_protocol_pagination.py
@@ -0,0 +1,274 @@
+# 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.
+"""Stable, permission-bound pagination for MCP list operations."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import httpx2
+import pytest
+from mcp import Client, MCPError, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from mcp.types import (
+ INVALID_PARAMS,
+ LATEST_PROTOCOL_VERSION,
+ PaginatedRequestParams,
+ Tool,
+)
+
+from doris_mcp_server.http_transport import DorisMCPHTTPTransport
+from doris_mcp_server.utils.security import (
+ AuthContext,
+ reset_auth_context,
+ set_current_auth_context,
+)
+from test.protocol.pagination_fixture import (
+ PROMPT_NAMES,
+ RESOURCE_URIS,
+ TOOL_NAMES,
+ create_pagination_server,
+)
+
+
+async def _collect_pages(
+ list_method,
+ *,
+ result_field: str,
+ item_key,
+) -> tuple[list[str], int]:
+ cursor = None
+ seen_cursors: set[str] = set()
+ identifiers: list[str] = []
+ page_count = 0
+
+ while True:
+ result = await list_method(cursor=cursor, cache_mode="bypass")
+ page_count += 1
+ page = getattr(result, result_field)
+ assert 0 < len(page) <= 2
+ identifiers.extend(item_key(item) for item in page)
+
+ cursor = result.next_cursor
+ if cursor is None:
+ break
+ assert cursor not in seen_cursors
+ seen_cursors.add(cursor)
+
+ return identifiers, page_count
+
+
+async def _assert_all_lists_are_complete(client: Client) -> None:
+ resource_uris, resource_pages = await _collect_pages(
+ client.list_resources,
+ result_field="resources",
+ item_key=lambda resource: str(resource.uri),
+ )
+ tool_names, tool_pages = await _collect_pages(
+ client.list_tools,
+ result_field="tools",
+ item_key=lambda tool: tool.name,
+ )
+ prompt_names, prompt_pages = await _collect_pages(
+ client.list_prompts,
+ result_field="prompts",
+ item_key=lambda prompt: prompt.name,
+ )
+
+ assert resource_uris == RESOURCE_URIS
+ assert tool_names == TOOL_NAMES
+ assert prompt_names == PROMPT_NAMES
+ assert len(resource_uris) == len(set(resource_uris))
+ assert len(tool_names) == len(set(tool_names))
+ assert len(prompt_names) == len(set(prompt_names))
+ assert (resource_pages, tool_pages, prompt_pages) == (3, 3, 3)
+
+
[email protected]("mode", ["modern", "legacy"])
[email protected]
+async def test_all_list_operations_page_without_duplicates_or_loss(mode: str):
+ server, _ = create_pagination_server()
+ client_context = (
+ Client(server) if mode == "modern" else Client(server, mode="legacy")
+ )
+
+ async with client_context as client:
+ await _assert_all_lists_are_complete(client)
+
+
[email protected]
+async def test_streamable_http_carries_opaque_cursor_across_all_list_methods():
+ server, _ = create_pagination_server()
+ app = DorisMCPHTTPTransport(app=server, security_settings=None)
+
+ async with (
+ app.run(),
+ httpx2.ASGITransport(app.handle_request) as transport,
+ httpx2.AsyncClient(
+ transport=transport,
+ base_url="http://127.0.0.1:3000";,
+ ) as client,
+ ):
+ for method, result_field, item_key, expected in (
+ (
+ "resources/list",
+ "resources",
+ lambda item: item["uri"],
+ RESOURCE_URIS,
+ ),
+ ("tools/list", "tools", lambda item: item["name"], TOOL_NAMES),
+ ("prompts/list", "prompts", lambda item: item["name"],
PROMPT_NAMES),
+ ):
+ cursor = None
+ identifiers: list[str] = []
+ while True:
+ params: dict[str, Any] = {
+ "_meta": {
+ "io.modelcontextprotocol/protocolVersion":
"2026-07-28",
+ "io.modelcontextprotocol/clientCapabilities": {},
+ "io.modelcontextprotocol/clientInfo": {
+ "name": "pagination-http-test",
+ "version": "1.0.0",
+ },
+ }
+ }
+ if cursor is not None:
+ params["cursor"] = cursor
+ response = await client.post(
+ "/mcp",
+ json={
+ "jsonrpc": "2.0",
+ "id": len(identifiers) + 1,
+ "method": method,
+ "params": params,
+ },
+ headers={
+ "Accept": "application/json, text/event-stream",
+ "Content-Type": "application/json",
+ "Mcp-Protocol-Version": "2026-07-28",
+ "Mcp-Method": method,
+ },
+ )
+ assert response.status_code == 200
+ assert "mcp-session-id" not in response.headers
+ result = response.json()["result"]
+ identifiers.extend(item_key(item) for item in
result[result_field])
+ cursor = result.get("nextCursor")
+ if cursor is None:
+ break
+
+ assert identifiers == expected
+
+
[email protected]
+async def test_true_subprocess_stdio_pages_all_list_methods():
+ project_root = Path(__file__).resolve().parents[2]
+ server_params = StdioServerParameters(
+ command=sys.executable,
+ args=["-m", "test.protocol.pagination_stdio_server"],
+ cwd=project_root,
+ )
+
+ async with Client(stdio_client(server_params)) as client:
+ await _assert_all_lists_are_complete(client)
+
+
[email protected]
+async def test_cursor_rejects_wrong_collection_malformed_and_stale_snapshots():
+ server, managers = create_pagination_server()
+
+ async with Client(server) as client:
+ first_tools = await client.list_tools(cache_mode="bypass")
+ assert first_tools.next_cursor is not None
+
+ with pytest.raises(MCPError) as wrong_collection:
+ await client.list_resources(
+ cursor=first_tools.next_cursor,
+ cache_mode="bypass",
+ )
+ assert wrong_collection.value.code == INVALID_PARAMS
+ assert wrong_collection.value.data == {"cursorError":
"wrong_collection"}
+
+ with pytest.raises(MCPError) as malformed:
+ await client.list_tools(
+ cursor="not-a-valid-pagination-cursor",
+ cache_mode="bypass",
+ )
+ assert malformed.value.code == INVALID_PARAMS
+ assert malformed.value.data == {"cursorError": "invalid"}
+
+ managers.tools.tools.append(
+ Tool(
+ name="foxtrot",
+ description="new item invalidates the snapshot",
+ input_schema={"type": "object", "properties": {}},
+ )
+ )
+ with pytest.raises(MCPError) as stale:
+ await client.list_tools(
+ cursor=first_tools.next_cursor,
+ cache_mode="bypass",
+ )
+ assert stale.value.code == INVALID_PARAMS
+ assert stale.value.data == {"cursorError": "stale"}
+
+
[email protected]
+async def test_cursor_is_bound_to_the_authorization_context():
+ server, _ = create_pagination_server(page_size=1)
+ entry = server.get_request_handler("tools/list")
+ assert entry is not None
+ context = SimpleNamespace(protocol_version=LATEST_PROTOCOL_VERSION)
+
+ first_context = AuthContext(
+ token_id="token-a",
+ user_id="user-a",
+ permissions=["tool:list"],
+ roles=["analyst"],
+ auth_method="token",
+ )
+ token = set_current_auth_context(first_context)
+ try:
+ first_page = await entry.handler(context, PaginatedRequestParams())
+ finally:
+ reset_auth_context(token)
+ assert first_page.next_cursor is not None
+
+ second_context = AuthContext(
+ token_id="token-b",
+ user_id="user-b",
+ permissions=["tool:list"],
+ roles=["analyst"],
+ auth_method="token",
+ )
+ token = set_current_auth_context(second_context)
+ try:
+ with pytest.raises(MCPError) as changed_context:
+ await entry.handler(
+ context,
+ PaginatedRequestParams(cursor=first_page.next_cursor),
+ )
+ finally:
+ reset_auth_context(token)
+
+ assert changed_context.value.code == INVALID_PARAMS
+ assert changed_context.value.data == {
+ "cursorError": "authorization_context_changed"
+ }
diff --git a/test/tools/test_resources_manager_cache.py
b/test/tools/test_resources_manager_cache.py
index 4dcff8c..87c5192 100644
--- a/test/tools/test_resources_manager_cache.py
+++ b/test/tools/test_resources_manager_cache.py
@@ -196,14 +196,53 @@ async def test_metadata_cache_disabled_by_default():
@pytest.mark.asyncio
-async def test_resources_manager_does_not_reuse_global_metadata_cache():
+async def test_resources_manager_reuses_identity_scoped_metadata_cache():
connection_manager = FakeConnectionManager()
manager = DorisResourcesManager(connection_manager)
first = await manager._get_table_metadata()
second = await manager._get_table_metadata()
- assert manager.metadata_cache.enabled is False
+ assert manager.metadata_cache.enabled is True
+ assert [table.name for table in first] == ["orders_1"]
+ assert [table.name for table in second] == ["orders_1"]
+ assert connection_manager.acquires == 1
+ assert connection_manager.releases == 1
+
+
[email protected]
+async def test_resources_manager_does_not_share_metadata_across_identities():
+ connection_manager = FakeConnectionManager()
+ manager = DorisResourcesManager(connection_manager)
+
+ first_token = set_current_auth_context(
+ AuthContext(
+ token_id="token-a",
+ user_id="user-a",
+ roles=["reader"],
+ permissions=["resource:list"],
+ auth_method="token",
+ )
+ )
+ try:
+ first = await manager._get_table_metadata()
+ finally:
+ reset_auth_context(first_token)
+
+ second_token = set_current_auth_context(
+ AuthContext(
+ token_id="token-b",
+ user_id="user-b",
+ roles=["reader"],
+ permissions=["resource:list"],
+ auth_method="token",
+ )
+ )
+ try:
+ second = await manager._get_table_metadata()
+ finally:
+ reset_auth_context(second_token)
+
assert [table.name for table in first] == ["orders_1"]
assert [table.name for table in second] == ["orders_2"]
assert connection_manager.acquires == 2
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
