Chris Coutinho
2d46959d01
Unit test fixes: - test_userinfo_routes: patch nextcloud_httpx_client instead of httpx.AsyncClient - test_instrument_tool: patch trace_operation in metrics module (where imported) - test_management_app_password_endpoints: patch nextcloud_httpx_client and get_settings at correct import locations - test_management_status_endpoint: patch detect_auth_mode and get_settings at correct import locations (api.management, not config/config_validators) - test_token_exchange: fix TokenBrokerService constructor args (client_id/ client_secret instead of encryption_key) CI: - Add Node.js setup and astrolabe build step (composer + npm ci + npm run build) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
218 lines
7.3 KiB
Python
218 lines
7.3 KiB
Python
"""
|
|
Unit tests for @instrument_tool decorator.
|
|
|
|
Tests that the decorator correctly instruments MCP tools with both
|
|
Prometheus metrics and OpenTelemetry tracing.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.observability.metrics import instrument_tool
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_metrics():
|
|
"""Mock Prometheus metrics."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.observability.metrics.record_tool_call"
|
|
) as mock_record,
|
|
patch(
|
|
"nextcloud_mcp_server.observability.metrics.record_tool_error"
|
|
) as mock_error,
|
|
):
|
|
yield {"record_tool_call": mock_record, "record_tool_error": mock_error}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_tracer():
|
|
"""Mock OpenTelemetry tracer."""
|
|
with patch(
|
|
"nextcloud_mcp_server.observability.metrics.trace_operation"
|
|
) as mock_trace:
|
|
# Configure mock to act as a context manager that allows exceptions to propagate
|
|
mock_trace.return_value.__enter__ = MagicMock(return_value=None)
|
|
mock_trace.return_value.__exit__ = MagicMock(
|
|
return_value=False
|
|
) # Return False to allow exceptions to propagate
|
|
yield mock_trace
|
|
|
|
|
|
class TestInstrumentToolDecorator:
|
|
"""Test the @instrument_tool decorator."""
|
|
|
|
async def test_decorator_creates_trace_span(self, mock_tracer, mock_metrics):
|
|
"""Test that decorator creates OpenTelemetry span with correct attributes."""
|
|
|
|
@instrument_tool
|
|
async def example_tool(query: str, limit: int = 10):
|
|
return {"results": []}
|
|
|
|
# Call the tool
|
|
await example_tool(query="test query", limit=5)
|
|
|
|
# Verify trace_operation was called with correct parameters
|
|
mock_tracer.assert_called_once()
|
|
call_args = mock_tracer.call_args
|
|
|
|
# Check span name
|
|
assert call_args[0][0] == "mcp.tool.example_tool"
|
|
|
|
# Check span attributes
|
|
attributes = call_args[1]["attributes"]
|
|
assert attributes["mcp.tool.name"] == "example_tool"
|
|
assert "query" in attributes["mcp.tool.args"]
|
|
assert "test query" in attributes["mcp.tool.args"]
|
|
assert "limit" in attributes["mcp.tool.args"]
|
|
|
|
# Verify record_exception parameter
|
|
assert call_args[1]["record_exception"] is True
|
|
|
|
async def test_decorator_sanitizes_sensitive_arguments(
|
|
self, mock_tracer, mock_metrics
|
|
):
|
|
"""Test that sensitive arguments are excluded from span attributes."""
|
|
|
|
@instrument_tool
|
|
async def example_tool(
|
|
query: str, password: str, token: str, api_key: str, ctx: object
|
|
):
|
|
return {"success": True}
|
|
|
|
# Call with sensitive parameters
|
|
await example_tool(
|
|
query="test",
|
|
password="secret123",
|
|
token="bearer_token",
|
|
api_key="api_key_123",
|
|
ctx=MagicMock(),
|
|
)
|
|
|
|
# Verify trace was created
|
|
mock_tracer.assert_called_once()
|
|
attributes = mock_tracer.call_args[1]["attributes"]
|
|
|
|
# Check that sensitive fields are NOT in attributes
|
|
tool_args = attributes["mcp.tool.args"]
|
|
assert "password" not in tool_args
|
|
assert "secret123" not in tool_args
|
|
assert "token" not in tool_args
|
|
assert "bearer_token" not in tool_args
|
|
assert "api_key" not in tool_args
|
|
assert "api_key_123" not in tool_args
|
|
assert "ctx" not in tool_args
|
|
|
|
# Check that non-sensitive field IS included
|
|
assert "query" in tool_args
|
|
assert "test" in tool_args
|
|
|
|
async def test_decorator_limits_argument_string_length(
|
|
self, mock_tracer, mock_metrics
|
|
):
|
|
"""Test that tool arguments are limited to 500 characters."""
|
|
|
|
@instrument_tool
|
|
async def example_tool(query: str):
|
|
return {"results": []}
|
|
|
|
# Create a very long query string (>500 chars)
|
|
long_query = "x" * 1000
|
|
|
|
await example_tool(query=long_query)
|
|
|
|
# Verify arguments were truncated
|
|
mock_tracer.assert_called_once()
|
|
attributes = mock_tracer.call_args[1]["attributes"]
|
|
tool_args = attributes["mcp.tool.args"]
|
|
|
|
assert len(tool_args) <= 500
|
|
|
|
async def test_decorator_records_success_metrics(self, mock_tracer, mock_metrics):
|
|
"""Test that successful tool execution records metrics."""
|
|
|
|
@instrument_tool
|
|
async def example_tool():
|
|
return {"success": True}
|
|
|
|
# Call the tool
|
|
await example_tool()
|
|
|
|
# Verify success metrics were recorded
|
|
mock_metrics["record_tool_call"].assert_called_once()
|
|
call_args = mock_metrics["record_tool_call"].call_args
|
|
assert call_args[0][0] == "example_tool" # tool_name
|
|
assert isinstance(call_args[0][1], float) # duration
|
|
assert call_args[0][2] == "success" # status
|
|
|
|
async def test_decorator_records_error_metrics(self, mock_tracer, mock_metrics):
|
|
"""Test that tool errors are recorded in metrics."""
|
|
|
|
@instrument_tool
|
|
async def failing_tool():
|
|
raise ValueError("Test error")
|
|
|
|
# Call the tool and expect exception
|
|
with pytest.raises(ValueError, match="Test error"):
|
|
await failing_tool()
|
|
|
|
# Verify error metrics were recorded
|
|
mock_metrics["record_tool_call"].assert_called_once()
|
|
call_args = mock_metrics["record_tool_call"].call_args
|
|
assert call_args[0][0] == "failing_tool" # tool_name
|
|
assert isinstance(call_args[0][1], float) # duration
|
|
assert call_args[0][2] == "error" # status
|
|
|
|
# Verify error type was recorded
|
|
mock_metrics["record_tool_error"].assert_called_once()
|
|
error_args = mock_metrics["record_tool_error"].call_args
|
|
assert error_args[0][0] == "failing_tool" # tool_name
|
|
assert error_args[0][1] == "ValueError" # error_type
|
|
|
|
async def test_decorator_preserves_function_metadata(
|
|
self, mock_tracer, mock_metrics
|
|
):
|
|
"""Test that decorator preserves function name and docstring."""
|
|
|
|
@instrument_tool
|
|
async def example_tool():
|
|
"""This is a test tool."""
|
|
return {"success": True}
|
|
|
|
# Verify function metadata is preserved
|
|
assert example_tool.__name__ == "example_tool"
|
|
assert example_tool.__doc__ == "This is a test tool."
|
|
|
|
async def test_decorator_preserves_return_value(self, mock_tracer, mock_metrics):
|
|
"""Test that decorator returns the original function's return value."""
|
|
|
|
@instrument_tool
|
|
async def example_tool(value: int):
|
|
return {"result": value * 2}
|
|
|
|
# Call the tool
|
|
result = await example_tool(value=5)
|
|
|
|
# Verify return value is unchanged
|
|
assert result == {"result": 10}
|
|
|
|
async def test_decorator_with_no_arguments(self, mock_tracer, mock_metrics):
|
|
"""Test decorator with tool that takes no arguments."""
|
|
|
|
@instrument_tool
|
|
async def no_args_tool():
|
|
return {"status": "ok"}
|
|
|
|
# Call the tool
|
|
await no_args_tool()
|
|
|
|
# Verify tracing works with no arguments
|
|
mock_tracer.assert_called_once()
|
|
attributes = mock_tracer.call_args[1]["attributes"]
|
|
|
|
# tool_args should be None when there are no kwargs
|
|
assert attributes["mcp.tool.args"] is None
|