mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-04-14 06:59:46 +00:00
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""Tests for MCP connection lifecycle in AgentLoop."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from nanobot.agent.loop import AgentLoop
|
|
from nanobot.bus.queue import MessageBus
|
|
|
|
|
|
def _make_loop(tmp_path, *, mcp_servers: dict | None = None) -> AgentLoop:
|
|
bus = MessageBus()
|
|
provider = MagicMock()
|
|
provider.get_default_model.return_value = "test-model"
|
|
provider.generation.max_tokens = 4096
|
|
return AgentLoop(
|
|
bus=bus,
|
|
provider=provider,
|
|
workspace=tmp_path,
|
|
model="test-model",
|
|
mcp_servers=mcp_servers or {"test": object()},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_mcp_retries_when_no_servers_connect(tmp_path, monkeypatch: pytest.MonkeyPatch):
|
|
loop = _make_loop(tmp_path)
|
|
attempts = 0
|
|
|
|
async def _fake_connect(_servers, _registry):
|
|
nonlocal attempts
|
|
attempts += 1
|
|
return {}
|
|
|
|
monkeypatch.setattr("nanobot.agent.tools.mcp.connect_mcp_servers", _fake_connect)
|
|
|
|
await loop._connect_mcp()
|
|
await loop._connect_mcp()
|
|
|
|
assert attempts == 2
|
|
assert loop._mcp_connected is False
|
|
assert loop._mcp_stacks == {}
|