68 lines
1.8 KiB
Python
68 lines
1.8 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pytest
|
|
from src.app_controller import AppController
|
|
from src import mcp_client
|
|
from src import ai_client
|
|
from src import models
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_external_mcp_e2e_refresh_and_call(tmp_path, monkeypatch):
|
|
# 1. Setup mock config and mock server script
|
|
config_file = tmp_path / "config.toml"
|
|
monkeypatch.setattr(models, "CONFIG_PATH", str(config_file))
|
|
|
|
mock_script = Path("scripts/mock_mcp_server.py").absolute()
|
|
|
|
mcp_config_file = tmp_path / "mcp_config.json"
|
|
mcp_data = {
|
|
"mcpServers": {
|
|
"e2e-server": {
|
|
"command": "python",
|
|
"args": [str(mock_script)],
|
|
"auto_start": True
|
|
}
|
|
}
|
|
}
|
|
mcp_config_file.write_text(json.dumps(mcp_data))
|
|
|
|
config_content = f"""
|
|
[ai]
|
|
mcp_config_path = "{mcp_config_file.as_posix()}"
|
|
[projects]
|
|
paths = []
|
|
active = ""
|
|
"""
|
|
config_file.write_text(config_content)
|
|
|
|
# 2. Initialize AppController
|
|
ctrl = AppController()
|
|
monkeypatch.setattr(ctrl, "_load_active_project", lambda: None)
|
|
ctrl.project = {}
|
|
|
|
# We need to mock start_services or just manually call what we need
|
|
ctrl.init_state()
|
|
|
|
# Trigger refresh event manually (since we don't have the background thread running in unit test)
|
|
await ctrl.refresh_external_mcps()
|
|
|
|
# 3. Verify tools are discovered
|
|
manager = mcp_client.get_external_mcp_manager()
|
|
tools = manager.get_all_tools()
|
|
assert "echo" in tools
|
|
|
|
# 4. Mock pre_tool_callback to auto-approve
|
|
mock_pre_tool = lambda desc, base, qa: "Approved"
|
|
|
|
# 5. Call execute_single_tool_call_async (via ai_client)
|
|
name, cid, out, orig = await ai_client._execute_single_tool_call_async(
|
|
"echo", {"message": "hello"}, "id1", ".", mock_pre_tool, None, 0
|
|
)
|
|
|
|
assert "ECHO: {'message': 'hello'}" in out
|
|
|
|
# Cleanup
|
|
await manager.stop_all()
|