Coverage for src / mcp_server_langgraph / core / test_helpers.py: 87%
45 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 00:43 +0000
1"""
2Test Helper Functions
4This module provides convenient helper functions for creating test instances
5of agents, servers, settings, and other components.
7Usage:
8 from mcp_server_langgraph.core.test_helpers import (
9 create_test_agent,
10 create_test_server,
11 create_test_settings,
12 create_test_container,
13 create_mock_llm_response,
14 create_mock_mcp_request,
15 )
17 # In your tests
18 def test_my_feature():
19 agent = create_test_agent()
20 assert agent.invoke({"message": "test"}) is not None
21"""
23from __future__ import annotations
25from datetime import datetime, timedelta, UTC
26from typing import Any
27from unittest.mock import MagicMock
29from mcp_server_langgraph.core.config import Settings
30from mcp_server_langgraph.core.container import ApplicationContainer
31from mcp_server_langgraph.core.container import create_test_container as _create_container
33# ==============================================================================
34# Container Helpers
35# ==============================================================================
38def create_test_container(settings: Settings | None = None) -> ApplicationContainer:
39 """
40 Create a test container with no-op providers.
42 This is a convenience wrapper around the container module's create_test_container.
44 Args:
45 settings: Optional custom settings
47 Returns:
48 ApplicationContainer configured for testing
50 Example:
51 container = create_test_container()
52 agent = create_agent(container.settings, container.get_telemetry())
53 """
54 return _create_container(settings=settings)
57# ==============================================================================
58# Settings Helpers
59# ==============================================================================
62def create_test_settings(**kwargs: Any) -> Settings:
63 """
64 Create test settings with safe defaults.
66 Args:
67 **kwargs: Override any settings attributes
69 Returns:
70 Settings configured for testing
72 Example:
73 settings = create_test_settings(model_name="test-model", log_level="ERROR")
74 """
75 defaults = {
76 "environment": "test",
77 "log_level": "DEBUG",
78 "jwt_secret_key": "test-secret-key-for-testing-only",
79 "hipaa_integrity_secret": "test-hipaa-secret-for-testing-only",
80 "openfga_store_id": "",
81 "openfga_model_id": "",
82 "anthropic_api_key": "test-anthropic-key",
83 "enable_tracing": False,
84 "enable_metrics": False,
85 }
86 defaults.update(kwargs)
87 return Settings(**defaults) # type: ignore[arg-type]
90# ==============================================================================
91# Agent Helpers
92# ==============================================================================
95def create_test_agent(settings: Settings | None = None, container: ApplicationContainer | None = None) -> Any:
96 """
97 Create a test agent instance with dependency injection support.
99 Now uses the create_agent() factory function which supports containers.
101 Args:
102 settings: Optional custom settings (if container not provided)
103 container: Optional container to use for dependencies
105 Returns:
106 Agent instance (LangGraph CompiledStateGraph)
108 Example:
109 agent = create_test_agent()
110 result = agent.invoke({"messages": [{"role": "user", "content": "test"}]})
111 """
112 # Import the new DI-enabled agent creation function
113 from mcp_server_langgraph.core.agent import create_agent
115 # Use the new factory function with container support
116 agent = create_agent(settings=settings, container=container)
118 return agent
121# ==============================================================================
122# Server Helpers
123# ==============================================================================
126def create_test_server(container: ApplicationContainer | None = None) -> Any:
127 """
128 Create a test MCP server instance.
130 Args:
131 container: Optional container to use for dependencies
133 Returns:
134 MCP server instance
136 Example:
137 server = create_test_server()
138 # Use server in tests
139 """
140 if container is None:
141 container = create_test_container()
143 # For now, return a mock server
144 # TODO: When we refactor server to use container, this will create a real test server
145 mock_server = MagicMock()
146 mock_server.server = MagicMock()
147 mock_server.auth = container.get_auth()
148 mock_server.settings = container.settings
150 return mock_server
153# ==============================================================================
154# Mock Helpers
155# ==============================================================================
158def create_mock_llm_response(content: str = "Test response", model: str = "test-model", **kwargs: Any) -> Any:
159 """
160 Create a mock LLM response compatible with LangChain.
162 Args:
163 content: Response content
164 model: Model name
165 **kwargs: Additional response attributes
167 Returns:
168 Mock message object
170 Example:
171 response = create_mock_llm_response(content="Hello, world!")
172 """
173 from langchain_core.messages import AIMessage
175 return AIMessage(content=content, response_metadata={"model": model, **kwargs})
178def create_mock_llm_stream(chunks: list[str]) -> list[Any]:
179 """
180 Create a mock LLM stream response.
182 Args:
183 chunks: List of content chunks to stream
185 Returns:
186 List of mock chunk objects
188 Example:
189 stream = create_mock_llm_stream(["Hello", " ", "World"])
190 for chunk in stream:
191 print(chunk.content)
192 """
193 from langchain_core.messages import AIMessageChunk
195 return [AIMessageChunk(content=chunk) for chunk in chunks]
198def create_mock_mcp_request(
199 method: str = "tools/call", params: dict[str, Any] | None = None, request_id: int = 1
200) -> dict[str, Any]:
201 """
202 Create a mock MCP JSON-RPC request.
204 Args:
205 method: JSON-RPC method name
206 params: Method parameters
207 request_id: Request ID
209 Returns:
210 Dict representing MCP request
212 Example:
213 request = create_mock_mcp_request(
214 method="tools/call",
215 params={"name": "chat", "arguments": {"message": "test"}}
216 )
217 """
218 return {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params or {}}
221def create_mock_jwt_token(user_id: str = "test-user", expiry_hours: int = 1) -> str:
222 """
223 Create a mock JWT token for testing.
225 Args:
226 user_id: User ID to include in token
227 expiry_hours: Hours until token expires
229 Returns:
230 JWT token string
232 Example:
233 token = create_mock_jwt_token(user_id="alice")
234 # Use token in auth tests
235 """
236 import jwt
238 payload = {
239 "sub": user_id,
240 "exp": datetime.now(UTC) + timedelta(hours=expiry_hours),
241 "iat": datetime.now(UTC),
242 }
244 # nosemgrep: python.jwt.security.jwt-hardcode.jwt-python-hardcoded-secret
245 return jwt.encode(payload, "test-secret-key", algorithm="HS256") # type: ignore[no-any-return]
248# ==============================================================================
249# Assertion Helpers
250# ==============================================================================
253def assert_valid_mcp_response(response: dict[str, Any]) -> None:
254 """
255 Assert that a response is a valid MCP JSON-RPC response.
257 Args:
258 response: Response to validate
260 Raises:
261 AssertionError: If response is invalid
263 Example:
264 response = server.handle_request(request)
265 assert_valid_mcp_response(response)
266 """
267 assert "jsonrpc" in response
268 assert response["jsonrpc"] == "2.0"
269 assert "id" in response
270 assert ("result" in response) or ("error" in response)
273def assert_valid_agent_state(state: dict[str, Any]) -> None:
274 """
275 Assert that a state dict is a valid LangGraph agent state.
277 Args:
278 state: State to validate
280 Raises:
281 AssertionError: If state is invalid
283 Example:
284 state = agent.invoke({"messages": [...]})
285 assert_valid_agent_state(state)
286 """
287 assert "messages" in state
288 assert isinstance(state["messages"], list)