From 101ef743eb867b14e6bb3f0de2a21547cd36907f Mon Sep 17 00:00:00 2001 From: FreeOnePlus Date: Thu, 30 Jul 2026 14:57:46 +0800 Subject: [PATCH] test: enforce domain coverage floors --- .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_github_actions_contract.py | 23 +- test/tools/test_manager_routing_boundaries.py | 237 ++++++++++++++ 12 files changed, 1439 insertions(+), 5 deletions(-) create mode 100644 test/auth/test_auth_middleware_boundaries.py create mode 100644 test/auth/test_jwt_runtime_boundaries.py create mode 100644 test/auth/test_token_handlers_boundaries.py create mode 100644 test/auth/test_token_validator_boundaries.py create mode 100644 test/deployment/check_coverage_domains.py create mode 100644 test/deployment/test_coverage_domains.py create mode 100644 test/tools/test_manager_routing_boundaries.py 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"} + + +@pytest.mark.asyncio +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"}) + + +@pytest.mark.asyncio +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" + + +@pytest.mark.asyncio +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"] + + +@pytest.mark.asyncio +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 + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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} + + +@pytest.mark.asyncio +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") + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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", + ) + + +@pytest.mark.asyncio +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 + + +@pytest.mark.asyncio +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" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("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() + + +@pytest.mark.asyncio +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" + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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",) + + +@pytest.mark.asyncio +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" + + +@pytest.mark.asyncio +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", + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("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" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("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 + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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 + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("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) + + +@pytest.mark.asyncio +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 + + +@pytest.mark.asyncio +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, + } + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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"), + ] + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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" + + +@pytest.mark.asyncio +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"]