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 ad4649d test: enforce domain coverage floors (#152)
ad4649d is described below
commit ad4649d14fe703e01f0251a2807759efe2d2bb82
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 15:07:36 2026 +0800
test: enforce domain coverage floors (#152)
---
.github/workflows/ci.yml | 5 +
CHANGELOG.md | 2 +
README.md | 13 +
pyproject.toml | 1 +
test/auth/test_auth_middleware_boundaries.py | 197 +++++++++++++++
test/auth/test_jwt_runtime_boundaries.py | 257 ++++++++++++++++++++
test/auth/test_token_handlers_boundaries.py | 306 ++++++++++++++++++++++++
test/auth/test_token_validator_boundaries.py | 172 +++++++++++++
test/deployment/check_coverage_domains.py | 142 +++++++++++
test/deployment/test_coverage_domains.py | 89 +++++++
test/deployment/test_github_actions_contract.py | 23 +-
test/tools/test_manager_routing_boundaries.py | 237 ++++++++++++++++++
12 files changed, 1439 insertions(+), 5 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b91d397..adb7090 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -66,6 +66,11 @@ jobs:
run: uv sync --frozen --group dev
- name: Run full test suite
run: uv run pytest -q -W error
+ - name: Enforce domain coverage floors
+ run: |
+ uv run coverage json -o "$RUNNER_TEMP/coverage.json"
+ uv run python test/deployment/check_coverage_domains.py \
+ "$RUNNER_TEMP/coverage.json"
package:
name: Build and wheel smoke
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 97828c4..31ec2a3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -76,6 +76,8 @@ under **Unreleased** until a new version is selected and
published.
- Persisted static bearer tokens as self-describing SHA-256/SHA-512 digests,
with atomic owner-only writes and one-way migration from legacy plaintext
token files.
+- Added coverage gates for the protocol, authentication, and core manager
+ domains at 80%, alongside a 55% whole-repository floor.
### Fixed
diff --git a/README.md b/README.md
index 44ffec3..f1cab45 100644
--- a/README.md
+++ b/README.md
@@ -1580,6 +1580,19 @@ async def test_new_tool():
print(result)
```
+Run the release test gate with:
+
+```bash
+uv run pytest -q -W error
+uv run coverage json -o coverage.json
+uv run python test/deployment/check_coverage_domains.py coverage.json
+```
+
+The test suite enforces at least 55% coverage across the repository. The
+generated coverage report is also checked at an 80% floor for the protocol,
+authentication, and core manager domains, so high-risk runtime code cannot be
+hidden by unrelated coverage.
+
## MCP Client
The project includes a unified MCP client (`doris_mcp_client/`) for testing
and integration purposes. The client supports multiple connection modes and
provides a convenient interface for interacting with the MCP server.
diff --git a/pyproject.toml b/pyproject.toml
index a739c6c..11d2ea8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -257,6 +257,7 @@ omit = [
]
[tool.coverage.report]
+fail_under = 55
exclude_lines = [
"pragma: no cover",
"def __repr__",
diff --git a/test/auth/test_auth_middleware_boundaries.py
b/test/auth/test_auth_middleware_boundaries.py
new file mode 100644
index 0000000..925beaa
--- /dev/null
+++ b/test/auth/test_auth_middleware_boundaries.py
@@ -0,0 +1,197 @@
+# 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.
+"""ASGI and MCP boundaries for the JWT authentication middleware."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+import pytest
+
+from doris_mcp_server.auth.auth_middleware import (
+ AuthenticationError,
+ AuthMiddleware,
+ AuthorizationError,
+)
+from doris_mcp_server.utils.auth_credentials import BearerCredentials
+from doris_mcp_server.utils.security import AuthContext, SecurityLevel
+
+
+class _JWTManager:
+ async def validate_token(self, token: str, token_type: str) -> dict[str,
Any]:
+ if token == "invalid":
+ raise ValueError("invalid token")
+ assert token_type == "access"
+ return {
+ "payload": {
+ "jti": "jwt-1",
+ "sub": "reader-1",
+ "roles": ["reader"],
+ "permissions": ["read_data"],
+ "security_level": "internal",
+ "iat": int(datetime.now(UTC).timestamp()),
+ }
+ }
+
+
+async def _receive() -> dict[str, Any]:
+ return {"type": "http.disconnect"}
+
+
[email protected]
+async def test_header_helpers_and_mcp_authentication() -> None:
+ middleware = AuthMiddleware(_JWTManager())
+
+ assert middleware.extract_token_from_header("Bearer jwt-value") ==
"jwt-value"
+ assert middleware.extract_token_from_header("Basic ignored") is None
+
+ context = await middleware.authenticate_mcp_request({"X-Auth-Token":
"jwt-value"})
+ assert context.user_id == "reader-1"
+ assert context.auth_method == "jwt"
+ assert context.token == ""
+
+ headers = await middleware.create_auth_response_headers(context)
+ assert headers == {
+ "X-Auth-User": "reader-1",
+ "X-Auth-Roles": "reader",
+ "X-Auth-Session": "jwt-1",
+ "X-Auth-Security-Level": "internal",
+ }
+
+ with pytest.raises(ValueError, match="Missing JWT token"):
+ await middleware.authenticate_request(BearerCredentials())
+ with pytest.raises(ValueError, match="JWT authentication failed"):
+ await middleware.authenticate_mcp_request({"Authorization": "Bearer
invalid"})
+
+
[email protected]
+async def test_http_middleware_adds_identity_headers() -> None:
+ middleware = AuthMiddleware(_JWTManager())
+
+ async def app(
+ scope: dict[str, Any],
+ receive: Any,
+ send: Any,
+ ) -> None:
+ assert scope["auth_context"].user_id == "reader-1"
+ await send(
+ {
+ "type": "http.response.start",
+ "status": 200,
+ "headers": [(b"x-existing", b"kept")],
+ }
+ )
+ await send({"type": "http.response.body", "body": b"ok"})
+
+ wrapped = middleware.create_http_middleware(app, skip_paths=["/public"])
+ sent: list[dict[str, Any]] = []
+
+ async def send(message: dict[str, Any]) -> None:
+ sent.append(message)
+
+ await wrapped(
+ {
+ "type": "http",
+ "path": "/mcp",
+ "headers": [(b"authorization", b"Bearer jwt-value")],
+ },
+ _receive,
+ send,
+ )
+
+ response_headers = dict(sent[0]["headers"])
+ assert sent[0]["status"] == 200
+ assert response_headers[b"x-existing"] == b"kept"
+ assert response_headers[b"X-Auth-User"] == b"reader-1"
+ assert response_headers[b"X-Auth-Security-Level"] == b"internal"
+
+
[email protected]
+async def test_http_middleware_skips_public_and_non_http_scopes() -> None:
+ middleware = AuthMiddleware(_JWTManager())
+ observed: list[str] = []
+
+ async def app(scope: dict[str, Any], receive: Any, send: Any) -> None:
+ observed.append(scope["type"])
+
+ wrapped = middleware.create_http_middleware(app, skip_paths=["/public"])
+
+ async def send(_message: dict[str, Any]) -> None:
+ raise AssertionError("pass-through requests must not emit their own
response")
+
+ await wrapped(
+ {"type": "http", "path": "/public/child", "headers": []},
+ _receive,
+ send,
+ )
+ await wrapped(
+ {"type": "websocket", "path": "/mcp", "headers": []},
+ _receive,
+ send,
+ )
+
+ assert observed == ["http", "websocket"]
+
+
[email protected]
+async def test_http_middleware_returns_bearer_challenge_on_failure() -> None:
+ middleware = AuthMiddleware(_JWTManager())
+
+ async def app(scope: dict[str, Any], receive: Any, send: Any) -> None:
+ raise AssertionError("unauthenticated request reached the application")
+
+ wrapped = middleware.create_http_middleware(app, skip_paths=["/public"])
+ sent: list[dict[str, Any]] = []
+
+ async def send(message: dict[str, Any]) -> None:
+ sent.append(message)
+
+ await wrapped(
+ {
+ "type": "http",
+ "path": "/mcp",
+ "headers": [(b"authorization", b"Bearer invalid")],
+ },
+ _receive,
+ send,
+ )
+
+ assert sent[0]["status"] == 401
+ assert dict(sent[0]["headers"])[b"www-authenticate"] == b"Bearer"
+ assert b"Authentication failed" in sent[1]["body"]
+
+
+def test_authentication_and_authorization_errors_keep_codes() -> None:
+ authentication = AuthenticationError("bad credentials")
+ authorization = AuthorizationError("forbidden")
+
+ assert authentication.message == "bad credentials"
+ assert authentication.error_code == "AUTH_FAILED"
+ assert authorization.message == "forbidden"
+ assert authorization.error_code == "ACCESS_DENIED"
+
+
+def test_auth_response_headers_use_security_level_value() -> None:
+ context = AuthContext(
+ user_id="reader-1",
+ roles=["reader", "analyst"],
+ session_id="session-1",
+ security_level=SecurityLevel.INTERNAL,
+ )
+
+ assert context.security_level.value == "internal"
diff --git a/test/auth/test_jwt_runtime_boundaries.py
b/test/auth/test_jwt_runtime_boundaries.py
new file mode 100644
index 0000000..61dfeff
--- /dev/null
+++ b/test/auth/test_jwt_runtime_boundaries.py
@@ -0,0 +1,257 @@
+# 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.
+"""JWT lifecycle coverage for supported authentication runtime behavior."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import jwt
+import pytest
+
+from doris_mcp_server.auth import jwt_manager as jwt_manager_module
+from doris_mcp_server.auth.jwt_manager import JWTManager
+
+JWT_SECRET = "test-only-jwt-signing-key-with-sufficient-length"
+
+
+def _security_config() -> SimpleNamespace:
+ return SimpleNamespace(
+ jwt_algorithm="HS256",
+ jwt_issuer="doris-mcp-test",
+ jwt_audience="doris-mcp-clients",
+ jwt_access_token_expiry=300,
+ jwt_refresh_token_expiry=600,
+ enable_token_refresh=True,
+ enable_token_revocation=True,
+ jwt_verify_signature=True,
+ jwt_require_exp=True,
+ jwt_require_iat=True,
+ jwt_require_nbf=False,
+ jwt_verify_audience=True,
+ jwt_verify_issuer=True,
+ jwt_leeway=0,
+ )
+
+
+def _manager(
+ monkeypatch: pytest.MonkeyPatch,
+) -> tuple[JWTManager, SimpleNamespace, SimpleNamespace]:
+ key_manager = SimpleNamespace(
+ key_rotation_interval=0,
+ initialize=AsyncMock(return_value=True),
+ get_private_key=Mock(return_value=JWT_SECRET),
+ get_public_key=Mock(return_value=JWT_SECRET),
+ is_key_expired=AsyncMock(return_value=False),
+ rotate_keys=AsyncMock(),
+ get_key_info=AsyncMock(return_value={"key_id": "test-key"}),
+ export_public_key_pem=AsyncMock(return_value="test-public-key"),
+ )
+
+ async def validate_claims(payload: dict) -> dict:
+ return {"valid": True, "user_id": payload.get("sub"), "payload":
payload}
+
+ validator = SimpleNamespace(
+ start=AsyncMock(),
+ stop=AsyncMock(),
+ validate_claims=AsyncMock(side_effect=validate_claims),
+ revoke_token=AsyncMock(),
+ get_validation_stats=AsyncMock(return_value={"validated": 1}),
+ )
+ monkeypatch.setattr(
+ jwt_manager_module,
+ "KeyManager",
+ Mock(return_value=key_manager),
+ )
+ monkeypatch.setattr(
+ jwt_manager_module,
+ "TokenValidator",
+ Mock(return_value=validator),
+ )
+
+ manager = JWTManager(SimpleNamespace(security=_security_config()))
+ return manager, key_manager, validator
+
+
[email protected]
+async def test_jwt_manager_initializes_and_shuts_down_owned_tasks(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ manager, key_manager, validator = _manager(monkeypatch)
+
+ assert await manager.initialize() is True
+ key_manager.initialize.assert_awaited_once()
+ validator.start.assert_awaited_once()
+
+ manager._key_rotation_task = asyncio.create_task(asyncio.sleep(60))
+ await manager.shutdown()
+
+ assert manager._key_rotation_task.cancelled()
+ validator.stop.assert_awaited_once()
+
+
[email protected]
+async def test_jwt_manager_full_token_lifecycle(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ manager, _, validator = _manager(monkeypatch)
+ user = {
+ "user_id": "reader-1",
+ "roles": ["reader"],
+ "permissions": ["read_data"],
+ "security_level": "internal",
+ }
+
+ tokens = await manager.generate_tokens(user, {"tenant": "tenant-a"})
+
+ assert tokens["token_type"] == "Bearer"
+ assert tokens["user_id"] == "reader-1"
+ assert tokens["refresh_token"]
+ validated = await manager.validate_token(tokens["access_token"])
+ assert validated["payload"]["tenant"] == "tenant-a"
+
+ refreshed = await manager.refresh_token(tokens["refresh_token"])
+ assert refreshed["user_id"] == "reader-1"
+
+ token_info = await manager.get_token_info(tokens["access_token"])
+ assert token_info["sub"] == "reader-1"
+ assert token_info["token_type"] == "access"
+ assert token_info["is_expired"] is False
+
+ revocable = jwt.encode(
+ {"jti": "revocable", "exp": int(time.time()) + 60},
+ JWT_SECRET,
+ algorithm="HS256",
+ )
+ assert await manager.revoke_token(revocable) is True
+ validator.revoke_token.assert_awaited_once_with(
+ "revocable",
+ pytest.approx(time.time() + 60, abs=2),
+ )
+
+ assert await manager.get_public_key_info() == {
+ "algorithm": "HS256",
+ "public_key_pem": "test-public-key",
+ "key_info": {"key_id": "test-key"},
+ }
+ stats = await manager.get_manager_stats()
+ assert stats["jwt_config"]["enable_refresh"] is True
+ assert stats["key_manager"] == {"key_id": "test-key"}
+ assert stats["validator"] == {"validated": 1}
+
+
[email protected]
+async def test_jwt_manager_rejects_invalid_lifecycle_inputs(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ manager, key_manager, _ = _manager(monkeypatch)
+ current_time = int(time.time())
+
+ wrong_type = jwt.encode(
+ {
+ "iss": manager.issuer,
+ "aud": manager.audience,
+ "iat": current_time,
+ "exp": current_time + 60,
+ "token_type": "refresh",
+ },
+ JWT_SECRET,
+ algorithm="HS256",
+ )
+ with pytest.raises(ValueError, match="Invalid token type"):
+ await manager.validate_token(wrong_type)
+
+ expired = jwt.encode(
+ {
+ "iss": manager.issuer,
+ "aud": manager.audience,
+ "iat": current_time - 120,
+ "exp": current_time - 60,
+ "token_type": "access",
+ },
+ JWT_SECRET,
+ algorithm="HS256",
+ )
+ with pytest.raises(ValueError, match="Token has expired"):
+ await manager.validate_token(expired)
+ with pytest.raises(ValueError, match="Invalid token"):
+ await manager.validate_token("not-a-jwt")
+
+ key_manager.get_public_key.return_value = None
+ with pytest.raises(ValueError, match="verification key is not
initialized"):
+ await manager.validate_token(wrong_type)
+ key_manager.get_public_key.return_value = JWT_SECRET
+
+ key_manager.get_private_key.return_value = None
+ with pytest.raises(RuntimeError, match="signing key is not initialized"):
+ await manager.generate_tokens({"user_id": "reader-1"})
+ key_manager.get_private_key.return_value = JWT_SECRET
+
+ manager.enable_refresh = False
+ with pytest.raises(ValueError, match="refresh is disabled"):
+ await manager.refresh_token("unused")
+
+ manager.enable_revocation = False
+ assert await manager.revoke_token("unused") is False
+ manager.enable_revocation = True
+ missing_claims = jwt.encode(
+ {"exp": current_time + 60},
+ JWT_SECRET,
+ algorithm="HS256",
+ )
+ assert await manager.revoke_token(missing_claims) is False
+ assert await manager.revoke_token("not-a-jwt") is False
+
+ with pytest.raises(jwt.InvalidTokenError):
+ await manager.decode_token_unsafe("not-a-jwt")
+
+
[email protected]
+async def test_jwt_manager_initialization_failure_paths(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ manager, key_manager, validator = _manager(monkeypatch)
+
+ key_manager.initialize.return_value = False
+ assert await manager.initialize() is False
+ validator.start.assert_not_awaited()
+
+ key_manager.initialize.side_effect = RuntimeError("key store unavailable")
+ assert await manager.initialize() is False
+
+ validator.stop.side_effect = RuntimeError("validator shutdown failed")
+ await manager.shutdown()
+
+
[email protected]
+async def test_jwt_manager_rotates_expired_keys(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ manager, key_manager, _ = _manager(monkeypatch)
+ key_manager.is_key_expired.return_value = True
+
+ async def cancel_after_iteration(_delay: float) -> None:
+ raise asyncio.CancelledError
+
+ monkeypatch.setattr(jwt_manager_module.asyncio, "sleep",
cancel_after_iteration)
+
+ await manager._auto_key_rotation()
+
+ key_manager.rotate_keys.assert_awaited_once()
diff --git a/test/auth/test_token_handlers_boundaries.py
b/test/auth/test_token_handlers_boundaries.py
new file mode 100644
index 0000000..12edf71
--- /dev/null
+++ b/test/auth/test_token_handlers_boundaries.py
@@ -0,0 +1,306 @@
+# 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.
+
+"""HTTP boundary coverage for token-management handlers."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import httpx
+import pytest
+from starlette.applications import Starlette
+from starlette.routing import Route
+
+from doris_mcp_server.auth.token_handlers import TokenHandlers
+
+
+def _manager(*, enabled: bool = True) -> SimpleNamespace:
+ return SimpleNamespace(
+ auth_provider=SimpleNamespace(token_manager=object() if enabled else
None),
+ create_token=AsyncMock(return_value="generated-token"),
+ revoke_token=AsyncMock(return_value=True),
+ list_tokens=AsyncMock(return_value=[{"token_id": "reader"}]),
+ get_token_stats=Mock(return_value={"active_tokens": 1}),
+ cleanup_expired_tokens=AsyncMock(return_value=2),
+ )
+
+
+def _app(manager: SimpleNamespace) -> Starlette:
+ handlers = TokenHandlers(manager, config=None)
+ return Starlette(
+ routes=[
+ Route(
+ "/token/create",
+ handlers.handle_create_token,
+ methods=["GET", "POST"],
+ ),
+ Route(
+ "/token/revoke",
+ handlers.handle_revoke_token,
+ methods=["DELETE"],
+ ),
+ Route(
+ "/token/revoke/{token_id}",
+ handlers.handle_revoke_token,
+ methods=["DELETE"],
+ ),
+ Route("/token/list", handlers.handle_list_tokens, methods=["GET"]),
+ Route("/token/stats", handlers.handle_token_stats,
methods=["GET"]),
+ Route(
+ "/token/cleanup",
+ handlers.handle_cleanup_tokens,
+ methods=["POST"],
+ ),
+ ]
+ )
+
+
+def _client(manager: SimpleNamespace) -> httpx.AsyncClient:
+ return httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=_app(manager)),
+ base_url="http://testserver";,
+ )
+
+
[email protected]
+async def test_create_token_from_get_query_includes_database_config():
+ manager = _manager()
+
+ async with _client(manager) as client:
+ response = await client.get(
+ "/token/create",
+ params={
+ "token_id": "reader",
+ "expires_hours": "12",
+ "description": "read only",
+ "custom_token": "chosen-token",
+ "db_host": "doris.internal",
+ "db_port": "9031",
+ "db_user": "readonly",
+ "db_password": "secret",
+ "db_database": "analytics",
+ "db_fe_http_port": "8031",
+ },
+ )
+
+ assert response.status_code == 200
+ assert response.json()["token"] == "generated-token"
+ call = manager.create_token.await_args
+ assert call.kwargs["token_id"] == "reader"
+ assert call.kwargs["expires_hours"] == 12
+ assert call.kwargs["custom_token"] == "chosen-token"
+ database = call.kwargs["database_config"]
+ assert database.host == "doris.internal"
+ assert database.port == 9031
+ assert database.user == "readonly"
+ assert database.password == "secret"
+ assert database.database == "analytics"
+ assert database.fe_http_port == 8031
+
+
[email protected]
+async def test_create_token_from_post_json_uses_defaults():
+ manager = _manager()
+
+ async with _client(manager) as client:
+ response = await client.post(
+ "/token/create",
+ json={
+ "token_id": "writer",
+ "database_config": {"host": "doris.internal"},
+ },
+ )
+
+ assert response.status_code == 200
+ database = manager.create_token.await_args.kwargs["database_config"]
+ assert database.host == "doris.internal"
+ assert database.port == 9030
+ assert database.user == "root"
+ assert database.database == "information_schema"
+
+
[email protected]
[email protected](
+ ("request_kind", "payload", "expected_error"),
+ [
+ ("json", {}, "token_id is required"),
+ ("query", {"token_id": "reader", "expires_hours": "soon"}, "integer"),
+ (
+ "json",
+ {"token_id": "reader", "database_config": {"port": "bad"}},
+ "Invalid database configuration",
+ ),
+ ],
+)
+async def test_create_token_rejects_invalid_inputs(
+ request_kind: str,
+ payload: dict[str, object],
+ expected_error: str,
+):
+ manager = _manager()
+
+ async with _client(manager) as client:
+ if request_kind == "query":
+ response = await client.get("/token/create", params=payload)
+ else:
+ response = await client.post("/token/create", json=payload)
+
+ assert response.status_code == 400
+ assert expected_error in response.json()["error"]
+ manager.create_token.assert_not_awaited()
+
+
[email protected]
+async def
test_create_token_rejects_invalid_json_and_reports_service_failures():
+ manager = _manager()
+
+ async with _client(manager) as client:
+ invalid_json = await client.post(
+ "/token/create",
+ content=b"{",
+ headers={"content-type": "application/json"},
+ )
+ bad_query_database = await client.get(
+ "/token/create",
+ params={"token_id": "reader", "db_host": "doris", "db_port":
"bad"},
+ )
+ manager.create_token.side_effect = RuntimeError("do not expose")
+ create_failure = await client.post(
+ "/token/create",
+ json={"token_id": "reader"},
+ )
+
+ assert invalid_json.status_code == 400
+ assert invalid_json.json()["error"] == "Invalid JSON body"
+ assert bad_query_database.status_code == 500
+ assert bad_query_database.json()["error"] == "Internal server error"
+ assert create_failure.status_code == 400
+ assert create_failure.json()["error"] == "Token creation failed"
+
+
[email protected]
+async def test_create_token_reports_disabled_authentication():
+ manager = _manager(enabled=False)
+
+ async with _client(manager) as client:
+ response = await client.post("/token/create", json={"token_id":
"reader"})
+
+ assert response.status_code == 503
+ manager.create_token.assert_not_awaited()
+
+
[email protected]
+async def test_revoke_token_supports_query_and_path_outcomes():
+ manager = _manager()
+
+ async with _client(manager) as client:
+ success = await client.delete("/token/revoke", params={"token_id":
"reader"})
+ manager.revoke_token.return_value = False
+ not_found = await client.delete("/token/revoke/missing")
+ missing = await client.delete("/token/revoke")
+
+ assert success.status_code == 200
+ assert success.json()["success"] is True
+ assert not_found.status_code == 404
+ assert not_found.json()["success"] is False
+ assert missing.status_code == 400
+ assert missing.json()["error"] == "token_id is required"
+ assert manager.revoke_token.await_args_list[0].args == ("reader",)
+ assert manager.revoke_token.await_args_list[1].args == ("missing",)
+
+
[email protected]
+async def test_revoke_token_reports_disabled_and_internal_failures():
+ disabled = _manager(enabled=False)
+ failing = _manager()
+ failing.revoke_token.side_effect = RuntimeError("database unavailable")
+
+ async with _client(disabled) as client:
+ unavailable = await client.delete(
+ "/token/revoke", params={"token_id": "reader"}
+ )
+ async with _client(failing) as client:
+ failed = await client.delete("/token/revoke", params={"token_id":
"reader"})
+
+ assert unavailable.status_code == 503
+ assert failed.status_code == 500
+ assert failed.json()["error"] == "Internal server error"
+
+
[email protected]
+async def test_list_stats_and_cleanup_success_paths():
+ manager = _manager()
+
+ async with _client(manager) as client:
+ listed = await client.get("/token/list")
+ stats = await client.get("/token/stats")
+ cleaned = await client.post("/token/cleanup")
+
+ assert listed.json() == {
+ "success": True,
+ "count": 1,
+ "tokens": [{"token_id": "reader"}],
+ }
+ assert stats.json() == {"success": True, "stats": {"active_tokens": 1}}
+ assert cleaned.json() == {
+ "success": True,
+ "cleaned_count": 2,
+ "message": "Cleaned up 2 expired tokens",
+ }
+
+
[email protected]
[email protected](
+ ("path", "method", "mock_name"),
+ [
+ ("/token/list", "GET", "list_tokens"),
+ ("/token/stats", "GET", "get_token_stats"),
+ ("/token/cleanup", "POST", "cleanup_expired_tokens"),
+ ],
+)
+async def test_list_stats_and_cleanup_hide_internal_failures(
+ path: str,
+ method: str,
+ mock_name: str,
+):
+ manager = _manager()
+ getattr(manager, mock_name).side_effect = RuntimeError("sensitive failure")
+
+ async with _client(manager) as client:
+ response = await client.request(method, path)
+
+ assert response.status_code == 500
+ assert response.json()["error"] == "Internal server error"
+
+
[email protected]
[email protected](
+ ("path", "method"),
+ [
+ ("/token/list", "GET"),
+ ("/token/stats", "GET"),
+ ("/token/cleanup", "POST"),
+ ],
+)
+async def test_read_and_cleanup_handlers_require_token_manager(path: str,
method: str):
+ manager = _manager(enabled=False)
+
+ async with _client(manager) as client:
+ response = await client.request(method, path)
+
+ assert response.status_code == 503
+ assert response.json()["error"] == "Token authentication is not enabled"
diff --git a/test/auth/test_token_validator_boundaries.py
b/test/auth/test_token_validator_boundaries.py
new file mode 100644
index 0000000..067ffc3
--- /dev/null
+++ b/test/auth/test_token_validator_boundaries.py
@@ -0,0 +1,172 @@
+# 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.
+"""Claim, blacklist, and rate-limit boundaries for JWT validation."""
+
+from __future__ import annotations
+
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from doris_mcp_server.auth.token_validators import (
+ RateLimiter,
+ TokenBlacklist,
+ TokenValidator,
+)
+
+
+def _security_config() -> SimpleNamespace:
+ return SimpleNamespace(
+ jwt_verify_signature=True,
+ jwt_verify_audience=True,
+ jwt_verify_issuer=True,
+ jwt_require_exp=True,
+ jwt_require_iat=True,
+ jwt_require_nbf=False,
+ jwt_leeway=0,
+ jwt_audience="doris-mcp-clients",
+ jwt_issuer="doris-mcp-test",
+ )
+
+
+def _validator() -> TokenValidator:
+ return TokenValidator(SimpleNamespace(security=_security_config()))
+
+
+def _valid_payload(**overrides: object) -> dict[str, object]:
+ now = time.time()
+ payload: dict[str, object] = {
+ "iss": "doris-mcp-test",
+ "aud": "doris-mcp-clients",
+ "exp": now + 120,
+ "iat": now,
+ "jti": "token-1",
+ "sub": "reader-1",
+ }
+ payload.update(overrides)
+ return payload
+
+
[email protected]
+async def test_blacklist_lifecycle_and_cleanup() -> None:
+ blacklist = TokenBlacklist(cleanup_interval=1)
+ now = time.time()
+
+ await blacklist.add_token("expired", now - 1)
+ await blacklist.add_token("active", now + 60)
+ assert await blacklist.is_blacklisted("active") is True
+ assert await blacklist.remove_token("missing") is False
+
+ stats = await blacklist.get_stats()
+ assert stats["total_blacklisted"] == 2
+ assert stats["active_blacklisted"] == 1
+ assert stats["expired_blacklisted"] == 1
+ assert await blacklist.cleanup_expired() == 1
+ assert await blacklist.remove_token("active") is True
+
+ await blacklist.start()
+ assert blacklist._cleanup_task is not None
+ await blacklist.stop()
+ assert blacklist._cleanup_task.cancelled()
+
+
[email protected]
+async def test_rate_limiter_enforces_window_and_reports_usage() -> None:
+ limiter = RateLimiter(max_requests=1, time_window=60)
+
+ assert await limiter.is_allowed("reader-1") is True
+ assert await limiter.is_allowed("reader-1") is False
+ usage = await limiter.get_usage("reader-1")
+ assert usage == {
+ "user_id": "reader-1",
+ "requests_in_window": 1,
+ "max_requests": 1,
+ "time_window": 60,
+ "remaining_requests": 0,
+ }
+
+ limiter._request_history["reader-2"] = [0.0]
+ assert await limiter.is_allowed("reader-2") is True
+
+
[email protected]
+async def test_validator_accepts_valid_claims_and_reports_runtime_stats() ->
None:
+ validator = _validator()
+ payload = _valid_payload(aud=["another-client", "doris-mcp-clients"])
+
+ result = await validator.validate_claims(payload)
+
+ assert result == {
+ "valid": True,
+ "user_id": "reader-1",
+ "payload": payload,
+ }
+ await validator.revoke_token("revoked", time.time() + 60)
+ stats = await validator.get_validation_stats()
+ assert stats["blacklist"]["active_blacklisted"] == 1
+ assert stats["validation_config"]["verify_signature"] is True
+ usage = await validator.get_user_rate_limit_info("reader-1")
+ assert usage["requests_in_window"] == 1
+
+ await validator.start()
+ await validator.stop()
+
+
[email protected]
[email protected](
+ ("payload", "message"),
+ [
+ (_valid_payload(iss="wrong"), "Invalid issuer"),
+ (_valid_payload(aud="wrong"), "Invalid audience"),
+ (_valid_payload(aud=["wrong"]), "not in"),
+ (_valid_payload(exp=None), "Missing 'exp' claim"),
+ (_valid_payload(exp=1), "Token has expired"),
+ (_valid_payload(nbf=None), "Missing 'nbf' claim"),
+ (_valid_payload(nbf=time.time() + 60), "Token not yet valid"),
+ (_valid_payload(iat=None), "Missing 'iat' claim"),
+ (_valid_payload(iat=time.time() + 60), "Token issued in the future"),
+ ],
+)
+async def test_validator_rejects_invalid_registered_claims(
+ payload: dict[str, object],
+ message: str,
+) -> None:
+ validator = _validator()
+
+ with pytest.raises(ValueError, match=message):
+ await validator.validate_claims(payload)
+
+
[email protected]
+async def test_validator_rejects_revoked_and_rate_limited_tokens() -> None:
+ validator = _validator()
+ await validator.blacklist.add_token("token-1", time.time() + 60)
+ with pytest.raises(ValueError, match="revoked"):
+ await validator.validate_claims(_valid_payload())
+
+ validator = _validator()
+ validator.rate_limiter.max_requests = 0
+ with pytest.raises(ValueError, match="Rate limit exceeded"):
+ await validator.validate_claims(_valid_payload())
+
+
+def test_validator_accepts_direct_security_config() -> None:
+ validator = TokenValidator(_security_config())
+
+ assert validator.expected_audience == "doris-mcp-clients"
+ assert validator.expected_issuer == "doris-mcp-test"
diff --git a/test/deployment/check_coverage_domains.py
b/test/deployment/check_coverage_domains.py
new file mode 100644
index 0000000..a5d01a0
--- /dev/null
+++ b/test/deployment/check_coverage_domains.py
@@ -0,0 +1,142 @@
+# 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.
+"""Enforce coverage floors for the highest-risk runtime domains."""
+
+from __future__ import annotations
+
+import json
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+@dataclass(frozen=True)
+class CoverageDomain:
+ minimum: float
+ exact_files: frozenset[str] = frozenset()
+ prefixes: tuple[str, ...] = ()
+
+
+DOMAINS = {
+ "protocol": CoverageDomain(
+ minimum=80.0,
+ exact_files=frozenset(
+ {
+ "doris_mcp_server/health.py",
+ "doris_mcp_server/http_transport.py",
+ "doris_mcp_server/pagination.py",
+ "doris_mcp_server/protocol.py",
+ "doris_mcp_server/schema_validation.py",
+ "doris_mcp_server/state_handles.py",
+ "doris_mcp_server/trace_context.py",
+ }
+ ),
+ ),
+ "authentication": CoverageDomain(
+ minimum=80.0,
+ prefixes=("doris_mcp_server/auth/",),
+ ),
+ "core_managers": CoverageDomain(
+ minimum=80.0,
+ exact_files=frozenset(
+ {
+ "doris_mcp_server/tools/prompts_manager.py",
+ "doris_mcp_server/tools/resources_manager.py",
+ "doris_mcp_server/tools/tool_catalog.py",
+ "doris_mcp_server/tools/tool_registry.py",
+ "doris_mcp_server/tools/tools_manager.py",
+ }
+ ),
+ ),
+}
+
+
+def _normalized_files(report: dict[str, Any]) -> dict[str, dict[str, Any]]:
+ return {
+ name.replace("\\", "/"): details
+ for name, details in report.get("files", {}).items()
+ }
+
+
+def evaluate_domains(report: dict[str, Any]) -> dict[str, float]:
+ """Return domain percentages or raise when the report is incomplete."""
+ files = _normalized_files(report)
+ percentages: dict[str, float] = {}
+
+ for name, domain in DOMAINS.items():
+ missing = domain.exact_files.difference(files)
+ if missing:
+ raise ValueError(
+ f"{name} coverage is missing required files: {',
'.join(sorted(missing))}"
+ )
+
+ selected = {
+ path: details
+ for path, details in files.items()
+ if path in domain.exact_files
+ or any(path.startswith(prefix) for prefix in domain.prefixes)
+ }
+ if not selected:
+ raise ValueError(f"{name} coverage selected no measured files")
+
+ statements = sum(
+ int(details["summary"]["num_statements"]) for details in
selected.values()
+ )
+ covered = sum(
+ int(details["summary"]["covered_lines"]) for details in
selected.values()
+ )
+ if statements == 0:
+ raise ValueError(f"{name} coverage has no executable statements")
+ percentages[name] = covered * 100.0 / statements
+
+ return percentages
+
+
+def enforce_domains(report: dict[str, Any]) -> dict[str, float]:
+ """Raise when any domain is below its declared minimum."""
+ percentages = evaluate_domains(report)
+ failures = [
+ f"{name}={percentages[name]:.2f}% < {domain.minimum:.2f}%"
+ for name, domain in DOMAINS.items()
+ if percentages[name] < domain.minimum
+ ]
+ if failures:
+ raise ValueError("coverage domain gate failed: " + "; ".join(failures))
+ return percentages
+
+
+def main(argv: list[str] | None = None) -> int:
+ args = sys.argv[1:] if argv is None else argv
+ if len(args) != 1:
+ print("usage: check_coverage_domains.py COVERAGE_JSON",
file=sys.stderr)
+ return 2
+
+ try:
+ report = json.loads(Path(args[0]).read_text(encoding="utf-8"))
+ percentages = enforce_domains(report)
+ except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as
exc:
+ print(str(exc), file=sys.stderr)
+ return 1
+
+ for name, percentage in percentages.items():
+ print(f"{name}: {percentage:.2f}% (minimum
{DOMAINS[name].minimum:.2f}%)")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/test/deployment/test_coverage_domains.py
b/test/deployment/test_coverage_domains.py
new file mode 100644
index 0000000..29a90fc
--- /dev/null
+++ b/test/deployment/test_coverage_domains.py
@@ -0,0 +1,89 @@
+# 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 __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from test.deployment.check_coverage_domains import (
+ DOMAINS,
+ enforce_domains,
+ evaluate_domains,
+ main,
+)
+
+
+def _report(*, covered: int = 8, statements: int = 10) -> dict:
+ files: dict[str, dict] = {}
+ for domain in DOMAINS.values():
+ paths = set(domain.exact_files)
+ if domain.prefixes:
+ paths.add(f"{domain.prefixes[0]}representative.py")
+ for path in paths:
+ files[path] = {
+ "summary": {
+ "covered_lines": covered,
+ "num_statements": statements,
+ }
+ }
+ return {"files": files}
+
+
+def test_domain_gate_accepts_each_domain_at_eighty_percent() -> None:
+ assert enforce_domains(_report()) == {
+ "protocol": 80.0,
+ "authentication": 80.0,
+ "core_managers": 80.0,
+ }
+
+
+def test_domain_gate_includes_every_authentication_module() -> None:
+ report = _report()
+ report["files"]["doris_mcp_server/auth/uncovered.py"] = {
+ "summary": {"covered_lines": 0, "num_statements": 10}
+ }
+
+ percentages = evaluate_domains(report)
+
+ assert percentages["authentication"] < 80.0
+ with pytest.raises(ValueError, match="authentication="):
+ enforce_domains(report)
+
+
+def test_domain_gate_fails_closed_when_required_file_is_absent() -> None:
+ report = _report()
+ report["files"].pop("doris_mcp_server/protocol.py")
+
+ with pytest.raises(ValueError, match="missing required files"):
+ evaluate_domains(report)
+
+
+def test_cli_reports_failures_and_success(
+ tmp_path: Path,
+ capsys: pytest.CaptureFixture[str],
+) -> None:
+ report_path = tmp_path / "coverage.json"
+ report_path.write_text(json.dumps(_report()), encoding="utf-8")
+ assert main([str(report_path)]) == 0
+ assert "authentication: 80.00%" in capsys.readouterr().out
+
+ report_path.write_text("{}", encoding="utf-8")
+ assert main([str(report_path)]) == 1
+ assert "missing required files" in capsys.readouterr().err
diff --git a/test/deployment/test_github_actions_contract.py
b/test/deployment/test_github_actions_contract.py
index 2311038..36082ea 100644
--- a/test/deployment/test_github_actions_contract.py
+++ b/test/deployment/test_github_actions_contract.py
@@ -16,12 +16,16 @@
# under the License.
import re
+import tomllib
from pathlib import Path
import yaml
+from test.deployment.check_coverage_domains import DOMAINS
+
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
WORKFLOW_PATH = REPOSITORY_ROOT / ".github" / "workflows" / "ci.yml"
+PYPROJECT_PATH = REPOSITORY_ROOT / "pyproject.toml"
FULL_COMMIT_ACTION = re.compile(r"^[^@\s]+@[0-9a-f]{40}$")
CHECKOUT_ACTION = "actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803"
CONFORMANCE_COMMIT = "49103de6ed70804e940637bf3e9e29e4a3f54e64"
@@ -82,13 +86,15 @@ def
test_quality_test_and_package_gates_cover_the_release_contract():
assert "uv sync --frozen --group dev" in tests
assert "uv run pytest -q -W error" in tests
+ assert "uv run coverage json" in tests
+ assert "test/deployment/check_coverage_domains.py" in tests
assert "uv sync --frozen --group dev" in package
assert "uv build" in package
- assert "test \"$wheel_count\" = \"1\"" in package
+ assert 'test "$wheel_count" = "1"' in package
assert 'cd "$smoke_root"' in package
- assert "doris-mcp-server\" --version" in package
- assert "doris-mcp-client\" --help" in package
+ assert 'doris-mcp-server" --version' in package
+ assert 'doris-mcp-client" --help' in package
assert "import doris_mcp_client, doris_mcp_server" in package
@@ -98,8 +104,7 @@ def
test_conformance_gate_is_official_pinned_and_has_no_failure_baseline():
checkout = next(
step
for step in job["steps"]
- if step.get("with", {}).get("repository")
- == "modelcontextprotocol/conformance"
+ if step.get("with", {}).get("repository") ==
"modelcontextprotocol/conformance"
)
assert checkout["with"]["ref"] == CONFORMANCE_COMMIT
@@ -111,3 +116,11 @@ def
test_conformance_gate_is_official_pinned_and_has_no_failure_baseline():
assert "--scenario server-stateless" in commands
assert "--url http://127.0.0.1:39124/mcp"; in commands
assert "expected-failures" not in commands
+
+
+def test_coverage_contract_has_full_repository_and_domain_floors() -> None:
+ pyproject = tomllib.loads(PYPROJECT_PATH.read_text(encoding="utf-8"))
+
+ assert pyproject["tool"]["coverage"]["report"]["fail_under"] == 55
+ assert set(DOMAINS) == {"protocol", "authentication", "core_managers"}
+ assert {domain.minimum for domain in DOMAINS.values()} == {80.0}
diff --git a/test/tools/test_manager_routing_boundaries.py
b/test/tools/test_manager_routing_boundaries.py
new file mode 100644
index 0000000..75d7695
--- /dev/null
+++ b/test/tools/test_manager_routing_boundaries.py
@@ -0,0 +1,237 @@
+# 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.
+
+"""Branch coverage for the manager's thin routing helpers."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, call
+
+import pytest
+
+from doris_mcp_server.tools.tools_manager import DorisToolsManager
+
+
+def _manager() -> DorisToolsManager:
+ manager = object.__new__(DorisToolsManager)
+ manager.monitoring_tools = SimpleNamespace(
+ get_monitoring_metrics=AsyncMock(
+ side_effect=[
+ {"success": True, "kind": "definitions"},
+ {"success": True, "timestamp": "2026-07-30T00:00:00Z"},
+ ]
+ )
+ )
+ manager.memory_tracker = SimpleNamespace(
+ get_realtime_memory_stats=AsyncMock(
+ return_value={
+ "success": True,
+ "timestamp": "2026-07-30T00:00:00Z",
+ }
+ ),
+ get_historical_memory_stats=AsyncMock(
+ return_value={"success": True, "points": []}
+ ),
+ )
+ manager.data_governance_tools = SimpleNamespace(
+ trace_column_lineage=AsyncMock(
+ side_effect=lambda **kwargs: {
+ "success": True,
+ "analysis_timestamp": "2026-07-30T00:00:00Z",
+ **kwargs,
+ }
+ )
+ )
+ return manager
+
+
[email protected]
+async def test_monitoring_metrics_routes_definitions_data_and_both():
+ definitions_manager = _manager()
+ definitions_manager.monitoring_tools.get_monitoring_metrics.side_effect =
None
+ definitions_manager.monitoring_tools.get_monitoring_metrics.return_value =
{
+ "success": True
+ }
+ definitions = await definitions_manager._get_monitoring_metrics_tool(
+ {
+ "content_type": "definitions",
+ "role": "fe",
+ "monitor_type": "system",
+ "priority": "all",
+ }
+ )
+
+ data_manager = _manager()
+ data_manager.monitoring_tools.get_monitoring_metrics.side_effect = None
+ data_manager.monitoring_tools.get_monitoring_metrics.return_value = {
+ "success": True
+ }
+ data = await data_manager._get_monitoring_metrics_tool(
+ {
+ "content_type": "data",
+ "role": "be",
+ "monitor_type": "query",
+ "priority": "core",
+ "include_raw_metrics": True,
+ }
+ )
+
+ both_manager = _manager()
+ both = await both_manager._get_monitoring_metrics_tool({"content_type":
"both"})
+
+ assert definitions == {"success": True}
+
definitions_manager.monitoring_tools.get_monitoring_metrics.assert_awaited_once_with(
+ "fe", "system", "all", info_only=True, format_type="prometheus"
+ )
+ assert data == {"success": True}
+
data_manager.monitoring_tools.get_monitoring_metrics.assert_awaited_once_with(
+ "be",
+ "query",
+ "core",
+ info_only=False,
+ format_type="prometheus",
+ include_raw_metrics=True,
+ )
+ assert both["content_type"] == "both"
+ assert both["timestamp"] == "2026-07-30T00:00:00Z"
+ assert both["_execution_info"] == {
+ "combined_response": True,
+ "definitions_available": True,
+ "data_available": True,
+ }
+
+
[email protected]
+async def test_monitoring_metrics_rejects_unknown_content_type():
+ manager = _manager()
+
+ result = await manager._get_monitoring_metrics_tool({"content_type":
"unknown"})
+
+ assert "Invalid content_type" in result["error"]
+ manager.monitoring_tools.get_monitoring_metrics.assert_not_awaited()
+
+
[email protected]
+async def test_memory_stats_routes_realtime_historical_and_both():
+ manager = _manager()
+
+ realtime = await manager._get_memory_stats_tool(
+ {
+ "data_type": "realtime",
+ "tracker_type": "process",
+ "include_details": False,
+ }
+ )
+ historical = await manager._get_memory_stats_tool(
+ {
+ "data_type": "historical",
+ "tracker_names": ["query"],
+ "time_range": "24h",
+ }
+ )
+ both = await manager._get_memory_stats_tool({"data_type": "both"})
+
+ assert realtime["success"] is True
+ assert historical == {"success": True, "points": []}
+ assert both["data_type"] == "both"
+ assert both["timestamp"] == "2026-07-30T00:00:00Z"
+ assert both["_execution_info"] == {
+ "combined_response": True,
+ "realtime_available": True,
+ "historical_available": True,
+ }
+ assert manager.memory_tracker.get_realtime_memory_stats.await_args_list ==
[
+ call("process", False),
+ call("overview", True),
+ ]
+ assert manager.memory_tracker.get_historical_memory_stats.await_args_list
== [
+ call(["query"], "24h"),
+ call(None, "1h"),
+ ]
+
+
[email protected]
+async def test_memory_stats_rejects_unknown_data_type():
+ manager = _manager()
+
+ result = await manager._get_memory_stats_tool({"data_type": "unknown"})
+
+ assert "Invalid data_type" in result["error"]
+ manager.memory_tracker.get_realtime_memory_stats.assert_not_awaited()
+ manager.memory_tracker.get_historical_memory_stats.assert_not_awaited()
+
+
[email protected]
+async def
test_column_lineage_requires_targets_and_validates_single_specification():
+ manager = _manager()
+
+ missing = await manager._trace_column_lineage_tool({})
+ invalid = await manager._trace_column_lineage_tool(
+ {"target_columns": "too.many.name.parts"}
+ )
+ two_part = await manager._trace_column_lineage_tool(
+ {
+ "target_columns": "orders.customer_id",
+ "analysis_depth": 2,
+ "catalog_name": "internal",
+ }
+ )
+ three_part = await manager._trace_column_lineage_tool(
+ {"target_columns": "analytics.orders.customer_id"}
+ )
+
+ assert missing == {"error": "target_columns parameter is required"}
+ assert "Invalid column specification" in invalid["error"]
+ assert two_part["table_name"] == "orders"
+ assert two_part["column_name"] == "customer_id"
+ assert two_part["db_name"] is None
+ assert two_part["depth"] == 2
+ assert three_part["db_name"] == "analytics"
+
+
[email protected]
+async def test_column_lineage_collects_multi_target_results_and_errors():
+ manager = _manager()
+
+ async def trace(**kwargs):
+ if kwargs["column_name"] == "broken":
+ raise RuntimeError("lineage unavailable")
+ return {
+ "success": True,
+ "analysis_timestamp": "2026-07-30T00:00:00Z",
+ **kwargs,
+ }
+
+ manager.data_governance_tools.trace_column_lineage.side_effect = trace
+ result = await manager._trace_column_lineage_tool(
+ {
+ "target_columns": [
+ "orders.customer_id",
+ "analytics.orders.total",
+ "invalid",
+ "orders.broken",
+ ],
+ "analysis_depth": 4,
+ }
+ )
+
+ assert result["multi_column_lineage"] is True
+ assert result["column_count"] == 4
+ assert result["analysis_timestamp"] == "2026-07-30T00:00:00Z"
+ assert result["results"]["orders.customer_id"]["db_name"] is None
+ assert result["results"]["analytics.orders.total"]["db_name"] ==
"analytics"
+ assert "Invalid column specification" in
result["results"]["invalid"]["error"]
+ assert "lineage unavailable" in result["results"]["orders.broken"]["error"]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
