MCPcopy Create free account
hub / github.com/IBM/mcp-cli / TestChatContext

Class TestChatContext

src/mcp_cli/chat/testing.py:20–129  ·  view source on GitHub ↗

Test-specific ChatContext that works with stream_manager instead of ToolManager. This is used for: 1. The --test-mode flag in chat handler 2. Unit tests that need a mock context

Source from the content-addressed store, hash-verified

18
19
20class TestChatContext(ChatContext):
21 """
22 Test-specific ChatContext that works with stream_manager instead of ToolManager.
23
24 This is used for:
25 1. The --test-mode flag in chat handler
26 2. Unit tests that need a mock context
27 """
28
29 def __init__(self, stream_manager: Any, model_manager: ModelManager):
30 """Create test context with stream_manager."""
31 # Initialize base attributes without calling super().__init__
32 self.tool_manager = None # type: ignore[assignment] # Tests don't use ToolManager
33 self.stream_manager = stream_manager
34 self.model_manager = model_manager
35
36 # Conversation state
37 self.exit_requested = False
38 self.conversation_history: list = []
39
40 # Context management notices
41 self._pending_context_notices: list[str] = []
42
43 # ToolProcessor back-reference
44 self.tool_processor: Any = None
45
46 # Tool state
47 self.tools: list = []
48 self.internal_tools: list = []
49 self.server_info: list = []
50 self.tool_to_server_map: dict = {}
51 self.openai_tools: list = []
52 self.tool_name_mapping: dict = {}
53
54 logger.debug(f"TestChatContext created with {self.provider}/{self.model}")
55
56 @classmethod
57 def create_for_testing(
58 cls,
59 stream_manager: Any,
60 provider: str | None = None,
61 model: str | None = None,
62 ) -> "TestChatContext":
63 """Factory for test contexts."""
64 model_manager = ModelManager()
65
66 if provider and model:
67 model_manager.switch_model(provider, model)
68 elif provider:
69 model_manager.switch_provider(provider)
70 elif model:
71 # Switch model in current provider
72 current_provider = model_manager.get_active_provider()
73 model_manager.switch_model(current_provider, model)
74
75 return cls(stream_manager, model_manager)
76
77 async def _initialize_tools(self, on_progress=None) -> None:

Calls

no outgoing calls