71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
from mulit_agent_app.application.agent_service import AgentService
|
|
from mulit_agent_app.config import AgentConfig
|
|
from mulit_agent_app.infrastructure.database import AgentDatabase
|
|
|
|
|
|
class _Session:
|
|
def __init__(self) -> None:
|
|
self.acks: list[tuple[str, int, str, str]] = []
|
|
|
|
def send_mcp_ack(self, tool_id: str, version: int, status: str, error: str = "") -> None:
|
|
self.acks.append((tool_id, version, status, error))
|
|
|
|
|
|
class _Executor:
|
|
def __init__(self) -> None:
|
|
self.desktop_enabled = False
|
|
|
|
def set_desktop_filesystem_enabled(self, enabled: bool) -> None:
|
|
self.desktop_enabled = enabled
|
|
|
|
|
|
def test_mcp_snapshot_persists_version_and_reacknowledges_after_reconnect(tmp_path):
|
|
config = AgentConfig(tmp_path, tmp_path / "settings.json", tmp_path / "agent.sqlite3", tmp_path / "agent.log")
|
|
service = AgentService.__new__(AgentService)
|
|
service._config = config
|
|
service._database = AgentDatabase(config.database_path)
|
|
service._database.initialize()
|
|
service.session = _Session()
|
|
service.task_executor = _Executor()
|
|
snapshot = {
|
|
"tools": [
|
|
{
|
|
"id": "desktop-filesystem",
|
|
"deployment_version": 3,
|
|
"enabled": True,
|
|
"scope": "desktop-directories",
|
|
}
|
|
]
|
|
}
|
|
|
|
service.apply_mcp_snapshot(snapshot)
|
|
service.apply_mcp_snapshot(snapshot)
|
|
|
|
assert service._database.get_meta("mcp.desktop-filesystem.enabled") == "true"
|
|
assert service._database.get_meta("mcp.desktop-filesystem.version") == "3"
|
|
assert service.task_executor.desktop_enabled is True
|
|
assert service.session.acks == [
|
|
("desktop-filesystem", 3, "applied", ""),
|
|
("desktop-filesystem", 3, "applied", ""),
|
|
]
|
|
|
|
|
|
def test_mcp_snapshot_rejects_lower_version_without_changing_enabled_state(tmp_path):
|
|
config = AgentConfig(tmp_path, tmp_path / "settings.json", tmp_path / "agent.sqlite3", tmp_path / "agent.log")
|
|
service = AgentService.__new__(AgentService)
|
|
service._config = config
|
|
service._database = AgentDatabase(config.database_path)
|
|
service._database.initialize()
|
|
service._database.set_meta("mcp.desktop-filesystem.enabled", "true")
|
|
service._database.set_meta("mcp.desktop-filesystem.version", "4")
|
|
service.session = _Session()
|
|
service.task_executor = _Executor()
|
|
|
|
service.apply_mcp_snapshot(
|
|
{"tools": [{"id": "desktop-filesystem", "deployment_version": 3, "enabled": True, "scope": "desktop-directories"}]}
|
|
)
|
|
|
|
assert service._database.get_meta("mcp.desktop-filesystem.version") == "4"
|
|
assert service.task_executor.desktop_enabled is False
|
|
assert service.session.acks == [("desktop-filesystem", 3, "failed", "拒绝低版本桌面文件工具部署。")]
|