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
| 18 | |
| 19 | |
| 20 | class 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: |
no outgoing calls