52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
import pytest
|
|
from beaver_gateway.mcp.types import McpServer
|
|
from beaver_gateway.mcp.wrap import build_python_tool_server
|
|
from fastmcp import Client
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
from beaver_agent.hands.komodo import Komodo
|
|
|
|
|
|
@pytest.fixture
|
|
def server():
|
|
tool = Komodo(url="http://127.0.0.1:9", key="k", secret="s")
|
|
return build_python_tool_server(
|
|
McpServer.python_tool(name="komodo", tools=[tool.komodo])
|
|
)
|
|
|
|
|
|
async def test_only_listed_actions_exist(server):
|
|
async with Client(server) as client:
|
|
(tool,) = await client.list_tools()
|
|
actions = tool.inputSchema["properties"]["action"]["enum"]
|
|
assert (
|
|
"prune" not in actions
|
|
and "destroy" not in actions
|
|
and "terminal" not in actions
|
|
)
|
|
assert {"logs", "deploy", "restart", "exec"} <= set(actions)
|
|
with pytest.raises(ToolError):
|
|
await client.call_tool(
|
|
"komodo", {"action": "prune", "stack": "beaver-agent"}
|
|
)
|
|
|
|
|
|
async def test_exec_denies_infrastructure_before_any_request(server):
|
|
async with Client(server) as client:
|
|
result = await client.call_tool(
|
|
"komodo", {"action": "exec", "container": "dell-periphery", "command": "id"}
|
|
)
|
|
assert "запрещён" in result.content[0].text
|
|
|
|
|
|
async def test_missing_parameter_is_an_error(server):
|
|
async with Client(server) as client:
|
|
with pytest.raises(ToolError, match="stack"):
|
|
await client.call_tool("komodo", {"action": "logs"})
|
|
|
|
|
|
def test_clip_keeps_head_and_tail():
|
|
k = Komodo(url="u", key="k", secret="s", max_chars=30)
|
|
out = k._clip("a" * 100)
|
|
assert out.startswith("a" * 20) and out.endswith("a" * 10) and "обрезано" in out
|